rag-rat-sync 0.21.0

Peer/p2p transport for the rag-rat op log: an iroh QUIC session that exchanges signed account entries between peers, feeding each through the op-log ingest seams (phase D, #406).
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
//! The node-authorization handshake that gates a session (phase D, #881).
//!
//! Runs BEFORE [`crate::session::run_session`], over the same bidirectional stream, and reveals
//! nothing about the account (not even confirmation that this peer hosts it) until the remote is
//! authorized. Each side presents a signed transport-node ↔ account-device binding and verifies the
//! other's under its OWN admission policy — authorization is **mutual**, so a peer handed a
//! poisoned address never streams its inventory to an impostor.
//!
//! Ordering is asymmetric to close the metadata gap without deadlocking (the transfer phase that
//! follows stays concurrent):
//! - the **acceptor** reads the dialer's binding and verifies it BEFORE sending its own — an
//!   unauthorized dialer learns nothing, not even the acceptor's binding;
//! - the **dialer** sends its binding (to the node iroh already authenticated it dialed) then
//!   verifies the acceptor's before proceeding to the data phase.
//!
//! A failure closes with one UNIFORM error regardless of cause (wrong account / not on roster / bad
//! signature / stale) so a peer cannot probe "does this server host account A / know device D".

use std::time::Duration;

use tokio::io::{AsyncRead, AsyncWrite};

use crate::codec::{self, CodecError};
use crate::wire::Frame;

/// How long a peer may take to send its auth frame before the handshake aborts. Deliberately much
/// shorter than the data-phase idle timeout: an unauthorized peer that connects and stalls must not
/// occupy the single-session accept slot for long.
pub const DEFAULT_PRE_AUTH_TIMEOUT: Duration = Duration::from_secs(10);

/// The largest frame accepted during the auth phase, checked against the length prefix BEFORE any
/// allocation. A valid auth frame is a ~600-byte cap (domain + tag + account id + a
/// [`crate::wire::MAX_AUTH_BINDING_BYTES`] binding); 1 KiB leaves headroom while keeping the
/// pre-auth allocation an unauthenticated peer can force to ~1 KiB, not the data-phase
/// [`crate::codec::MAX_FRAME_BYTES`].
const MAX_AUTH_FRAME_BYTES: u32 = 1024;

/// How a peer decides whether to admit a connection. Local policy, never negotiated on the wire —
/// an impostor cannot assert `Open` to exempt itself from the other side's `Closed`. Per-ACCOUNT,
/// not per-endpoint: one transport node may legitimately serve several accounts, so a public
/// account being `Open` must not open a private one on the same endpoint.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuthPolicy {
    /// Admit any peer. For a public, read-only knowledge base (a later slice) or a trusted LAN.
    Open,
    /// Admit only a peer whose binding verifies against this account's roster and the connection's
    /// authenticated remote node id. The default for a private account.
    Closed,
    /// Admit a not-yet-roster peer via a one-time invite token — the onboarding/pairing flow. Not
    /// implemented in this slice (it needs the token issue/redeem exchange); a session configured
    /// with it fails closed rather than silently admitting.
    InviteToken,
}

/// Which end of the connection this peer is — determines the send/verify order above.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuthRole {
    Dialer,
    Acceptor,
}

/// The non-stream inputs to one auth handshake, bundled so [`run_auth_phase`] takes the stream
/// pair, the authorizer, and this — rather than a long argument train.
#[derive(Debug, Clone, Copy)]
pub struct AuthConfig {
    /// Which end this peer is (sets the send/verify order).
    pub role: AuthRole,
    /// The account this session is scoped to (named in our own auth frame).
    pub account_id: [u8; 32],
    /// Our own transport node id (bound into the binding we present).
    pub local_node: [u8; 32],
    /// The peer's iroh-authenticated transport node id (checked against the binding it presents).
    pub remote_node: [u8; 32],
    /// How we admit the peer.
    pub policy: AuthPolicy,
    /// The CURRENT wall-clock (ms) for this handshake — used to stamp the binding we mint and to
    /// check the peer's binding freshness. Per-handshake, NOT a store-construction timestamp: a
    /// reused store would otherwise mint stale bindings and never advance the replay window.
    pub now_ms: i64,
    /// How long we wait for the peer's auth frame before aborting.
    pub pre_auth_timeout: Duration,
}

/// The account-authorization capability the transport needs from the store: mint our own binding,
/// and verify a peer's. Both resolve against the store's account + live fold; kept separate from
/// [`crate::session::SyncStore`] so the auth phase is testable without the data-phase machinery.
pub trait NodeAuth {
    /// Our signed binding vouching that `local_node` is this account's local device, stamped
    /// `now_ms`. `Err` if this store cannot authorize (e.g. no local account/device yet — the
    /// onboarding case).
    fn local_binding(&self, local_node: &[u8; 32], now_ms: i64) -> anyhow::Result<Vec<u8>>;

    /// Whether `binding` authorizes a peer whose iroh-authenticated node key is `remote_node`,
    /// judged fresh against `now_ms`. Returns a plain bool — the store collapses the internal
    /// failure taxonomy so the wire stays uniform. `Err` is reserved for a real fault (e.g. the
    /// DB read failed), not a rejected peer.
    fn authorize(
        &self,
        binding: &[u8],
        remote_node: &[u8; 32],
        now_ms: i64,
    ) -> anyhow::Result<bool>;
}

/// A node-authorization handshake that did not admit the connection.
#[derive(Debug)]
pub enum AuthError {
    /// The transport failed or the peer sent an unreadable frame.
    Codec(CodecError),
    /// The peer's binding did not satisfy our admission policy — the UNIFORM refusal (cause
    /// hidden).
    Unauthorized,
    /// The peer sent no auth frame within the pre-auth deadline.
    Timeout,
    /// The peer sent something other than an auth frame to open, or a policy we cannot serve.
    Protocol(String),
}

impl std::fmt::Display for AuthError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AuthError::Codec(e) => write!(f, "sync auth transport: {e}"),
            AuthError::Unauthorized => write!(f, "peer is not authorized for this account"),
            AuthError::Timeout => write!(f, "peer sent no auth frame before the deadline"),
            AuthError::Protocol(m) => write!(f, "sync auth protocol violation: {m}"),
        }
    }
}

impl std::error::Error for AuthError {}

/// Run the mutual auth handshake. On `Ok(())` both sides have verified each other under their own
/// policies and the caller may proceed to [`crate::session::run_session`] over the same stream; on
/// `Err` the caller drops the connection without revealing any inventory.
pub async fn run_auth_phase<W, R, A>(
    send: &mut W,
    recv: &mut R,
    auth: &A,
    cfg: AuthConfig,
) -> Result<(), AuthError>
where
    W: AsyncWrite + Unpin,
    R: AsyncRead + Unpin,
    A: NodeAuth,
{
    match cfg.role {
        // Acceptor verifies the dialer BEFORE revealing its own binding.
        AuthRole::Acceptor => {
            verify_peer(recv, auth, &cfg).await?;
            send_ours(send, auth, &cfg).await?;
        },
        // Dialer presents first (to the node it already authenticated), then verifies the acceptor
        // before proceeding to the data phase.
        AuthRole::Dialer => {
            send_ours(send, auth, &cfg).await?;
            verify_peer(recv, auth, &cfg).await?;
        },
    }
    Ok(())
}

async fn send_ours<W: AsyncWrite + Unpin>(
    send: &mut W,
    auth: &dyn NodeAuth,
    cfg: &AuthConfig,
) -> Result<(), AuthError> {
    // A store that cannot mint a binding (no local account/device) cannot authorize — fail closed.
    let binding =
        auth.local_binding(&cfg.local_node, cfg.now_ms).map_err(|_| AuthError::Unauthorized)?;
    // Bound the WRITE by the same pre-auth deadline as the read: a peer that opens the stream but
    // never grants receive credit would otherwise hang this write forever, blocking the acceptor's
    // single-session accept slot — a pre-auth DoS.
    let frame = Frame::Auth { account_id: cfg.account_id, binding };
    match tokio::time::timeout(cfg.pre_auth_timeout, codec::write_frame(send, &frame)).await {
        Ok(Ok(())) => Ok(()),
        Ok(Err(e)) => Err(AuthError::Codec(e)),
        Err(_elapsed) => Err(AuthError::Timeout),
    }
}

async fn verify_peer<R: AsyncRead + Unpin>(
    recv: &mut R,
    auth: &dyn NodeAuth,
    cfg: &AuthConfig,
) -> Result<(), AuthError> {
    let read = codec::read_frame_within(recv, MAX_AUTH_FRAME_BYTES);
    let frame = match tokio::time::timeout(cfg.pre_auth_timeout, read).await {
        Ok(Ok(frame)) => frame,
        Ok(Err(e)) => return Err(AuthError::Codec(e)),
        Err(_elapsed) => return Err(AuthError::Timeout),
    };
    let Frame::Auth { account_id: peer_account, binding } = frame else {
        return Err(AuthError::Protocol("peer did not open with an auth frame".into()));
    };
    // Account scope is enforced regardless of policy — even an Open endpoint must not proceed to
    // the data phase (whose Hello reveals the hosted account id + inventory) for a peer that
    // named a DIFFERENT account. Open relaxes the BINDING check, never the scope. Uniform error
    // either way.
    if peer_account != cfg.account_id {
        return Err(AuthError::Unauthorized);
    }
    // Every rejection below is the SAME uniform error — no not-on-roster / bad-sig distinction on
    // the wire.
    match cfg.policy {
        AuthPolicy::Open => Ok(()),
        AuthPolicy::Closed => {
            match auth.authorize(&binding, &cfg.remote_node, cfg.now_ms) {
                Ok(true) => Ok(()),
                Ok(false) => Err(AuthError::Unauthorized),
                // A real fault (DB read failed) is not the same as a rejected peer, but it still
                // means we cannot admit — surface it uniformly rather than admitting on error.
                Err(_) => Err(AuthError::Unauthorized),
            }
        },
        AuthPolicy::InviteToken =>
            Err(AuthError::Protocol("invite-token admission is not supported yet".into())),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const ACCT: [u8; 32] = [1u8; 32];
    const D_NODE: [u8; 32] = [2u8; 32];
    const A_NODE: [u8; 32] = [3u8; 32];

    /// A fake authorizer that tests the PROTOCOL (ordering, mutual verification, uniform refusal)
    /// independently of the binding crypto (covered by the oplog `node_binding` tests).
    struct FakeAuth {
        binding: Vec<u8>,
        authorize_ok: bool,
    }

    impl NodeAuth for FakeAuth {
        fn local_binding(&self, _local_node: &[u8; 32], _now_ms: i64) -> anyhow::Result<Vec<u8>> {
            Ok(self.binding.clone())
        }

        fn authorize(
            &self,
            _binding: &[u8],
            _remote_node: &[u8; 32],
            _now_ms: i64,
        ) -> anyhow::Result<bool> {
            Ok(self.authorize_ok)
        }
    }

    async fn run_pair(
        dialer: FakeAuth,
        dialer_account: [u8; 32],
        dialer_policy: AuthPolicy,
        acceptor: FakeAuth,
        acceptor_policy: AuthPolicy,
    ) -> (Result<(), AuthError>, Result<(), AuthError>) {
        let (mut d_send, mut a_recv) = tokio::io::duplex(1 << 16);
        let (mut a_send, mut d_recv) = tokio::io::duplex(1 << 16);
        // Short: the happy path never waits, and a refusal leaves the peer's read pending (the
        // duplex send half is not dropped until this fn returns, unlike a real connection
        // that closes on abort), so a tight timeout keeps the failure-path tests fast.
        let timeout = Duration::from_millis(300);
        let dialer_side = run_auth_phase(&mut d_send, &mut d_recv, &dialer, AuthConfig {
            role: AuthRole::Dialer,
            account_id: dialer_account,
            local_node: D_NODE,
            remote_node: A_NODE,
            policy: dialer_policy,
            now_ms: 1,
            pre_auth_timeout: timeout,
        });
        let acceptor_side = run_auth_phase(&mut a_send, &mut a_recv, &acceptor, AuthConfig {
            role: AuthRole::Acceptor,
            account_id: ACCT,
            local_node: A_NODE,
            remote_node: D_NODE,
            policy: acceptor_policy,
            now_ms: 1,
            pre_auth_timeout: timeout,
        });
        tokio::join!(dialer_side, acceptor_side)
    }

    fn ok_auth() -> FakeAuth {
        FakeAuth { binding: vec![1, 2, 3], authorize_ok: true }
    }

    #[tokio::test]
    async fn mutual_closed_authorization_admits_both() {
        let (d, a) =
            run_pair(ok_auth(), ACCT, AuthPolicy::Closed, ok_auth(), AuthPolicy::Closed).await;
        assert!(d.is_ok() && a.is_ok(), "both sides authorized each other: {d:?} {a:?}");
    }

    #[tokio::test]
    async fn an_unauthorized_dialer_is_refused_before_the_acceptor_reveals_its_binding() {
        // The acceptor rejects the dialer. Because the acceptor verifies BEFORE sending its own
        // binding, it aborts without revealing anything — the dialer's read then fails (no Auth
        // frame arrives). This is the mutual-auth ordering that stops inventory (and here even the
        // acceptor's binding) leaking to an unauthorized peer.
        let acceptor = FakeAuth { binding: vec![9, 9, 9], authorize_ok: false };
        let (dialer, accept) =
            run_pair(ok_auth(), ACCT, AuthPolicy::Closed, acceptor, AuthPolicy::Closed).await;
        assert!(matches!(accept, Err(AuthError::Unauthorized)), "acceptor refused: {accept:?}");
        assert!(dialer.is_err(), "dialer got no acceptor binding — nothing leaked: {dialer:?}");
    }

    #[tokio::test]
    async fn open_policy_admits_a_peer_that_would_fail_closed() {
        // A peer whose binding would never verify is admitted under Open (no verification), so both
        // sides complete.
        let anon_dialer = FakeAuth { binding: vec![], authorize_ok: false };
        let (d, a) =
            run_pair(anon_dialer, ACCT, AuthPolicy::Open, ok_auth(), AuthPolicy::Open).await;
        assert!(d.is_ok() && a.is_ok(), "open admits anyone: {d:?} {a:?}");
    }

    #[tokio::test]
    async fn a_dialer_naming_a_different_account_is_refused_under_closed() {
        // The dialer presents a binding scoped to a different account than the acceptor serves.
        let (_dialer, accept) = run_pair(
            ok_auth(),
            [0xee; 32], // dialer names a different account
            AuthPolicy::Closed,
            ok_auth(),
            AuthPolicy::Closed,
        )
        .await;
        assert!(
            matches!(accept, Err(AuthError::Unauthorized)),
            "the acceptor refuses a cross-account dialer: {accept:?}",
        );
    }

    #[tokio::test]
    async fn even_an_open_endpoint_refuses_a_cross_account_dialer() {
        // Open relaxes the binding check, NOT the account scope: a peer naming a different account
        // than the endpoint serves must be refused before run_session could leak the hosted
        // account's Hello + inventory.
        let (_dialer, accept) = run_pair(
            ok_auth(),
            [0xee; 32], // dialer names a different account
            AuthPolicy::Open,
            ok_auth(),
            AuthPolicy::Open,
        )
        .await;
        assert!(
            matches!(accept, Err(AuthError::Unauthorized)),
            "an open endpoint still enforces the account scope: {accept:?}",
        );
    }

    #[tokio::test]
    async fn invite_token_policy_fails_closed_until_implemented() {
        let (_d, a) =
            run_pair(ok_auth(), ACCT, AuthPolicy::Closed, ok_auth(), AuthPolicy::InviteToken).await;
        assert!(
            matches!(a, Err(AuthError::Protocol(_))),
            "invite-token is not admitted yet: {a:?}"
        );
    }

    #[tokio::test]
    async fn a_peer_that_sends_nothing_times_out() {
        // The write end stays alive but silent, so the acceptor's read pends and hits the deadline
        // (rather than seeing EOF).
        let (_silent_writer, mut recv) = tokio::io::duplex(1 << 10);
        let (mut send, _sink) = tokio::io::duplex(1 << 10);
        let r = run_auth_phase(&mut send, &mut recv, &ok_auth(), AuthConfig {
            role: AuthRole::Acceptor,
            account_id: ACCT,
            local_node: A_NODE,
            remote_node: D_NODE,
            policy: AuthPolicy::Closed,
            now_ms: 1,
            pre_auth_timeout: Duration::from_millis(100),
        })
        .await;
        assert!(matches!(r, Err(AuthError::Timeout)), "a silent peer times out: {r:?}");
    }

    #[tokio::test]
    async fn a_non_auth_first_frame_is_a_protocol_violation() {
        // The peer opens with a data-phase frame instead of an Auth frame.
        let (mut peer_send, mut recv) = tokio::io::duplex(1 << 12);
        let (mut send, _sink) = tokio::io::duplex(1 << 12);
        codec::write_frame(&mut peer_send, &Frame::Done).await.unwrap();
        let r = run_auth_phase(&mut send, &mut recv, &ok_auth(), AuthConfig {
            role: AuthRole::Acceptor,
            account_id: ACCT,
            local_node: A_NODE,
            remote_node: D_NODE,
            policy: AuthPolicy::Open,
            now_ms: 1,
            pre_auth_timeout: Duration::from_millis(200),
        })
        .await;
        assert!(matches!(r, Err(AuthError::Protocol(_))), "a non-auth opener is refused: {r:?}");
    }

    #[test]
    fn auth_errors_render() {
        // Every Display arm is reachable and non-empty.
        for e in [
            AuthError::Unauthorized,
            AuthError::Timeout,
            AuthError::Protocol("x".into()),
            AuthError::Codec(CodecError::Eof),
        ] {
            assert!(!format!("{e}").is_empty());
        }
    }
}