diff --git a/CHANGELOG.md b/CHANGELOG.md index d1fb2c1..2395597 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,24 @@ dependency bump below is the only behavior-affecting change so far ### Added +- **The forwarded op + the bind-first params surface (ADR-008 as + amended)** — the wire-stable types pin first (the ordering + `tunnels/params` used for the base op): `OP_TUNNEL_FORWARDED` / + `TUNNEL_FORWARDED_SCOPE` constants; `TunnelForwardedParams + {listen, substrate, peer}` (deny-unknown; `listen` is the + producer-minted correlation token, positive by wire contract — + schema `minimum: 1` + a serde-level pin) with a per-substrate + `peer` sub-schema (malformed peers fail the registry's schema gate + before the opener-side handler); `tunnel_forwarded_spec()` + (`Sub`-typed, External, `alk/tunnel` ALPN marker, no establisher — + the forwarded data plane starts immediately); `TunnelListenReply` + + `parse_listen_reply` (the typed projection of the bind-first + listen open's additive `bound`/`listen` reply fields — unknown + reply fields tolerated, malformed ones fail loud via the new + `TunnelError::MalformedReply`); `BindFirstPlan` (the opaque + assembly-typed listener + the token — the plan payload the + bind-first establisher/accept-dispatcher task consumes). + - **The direct op (`channels/tunnel/direct`, ADR-007)** — dynamic- target egress (the ssh `direct-tcpip` / SOCKS5 CONNECT shape), riding the same `alk/tunnel` ALPN: `SubstrateAddr` (the untagged diff --git a/src/error.rs b/src/error.rs index 65505b9..89d91a9 100644 --- a/src/error.rs +++ b/src/error.rs @@ -20,6 +20,13 @@ pub enum TunnelError { /// An I/O failure on the session's data plane. #[error("io: {0}")] Io(#[from] std::io::Error), + /// A reply field failed its typed projection — the wire carried a + /// malformed shape (e.g. [`crate::params::parse_listen_reply`] + /// found `bound` not substrate-shaped, a non-positive `listen`, + /// or a reply with no `channel_id` at all). A misbehaving + /// producer is never silently coerced. + #[error("malformed reply: {0}")] + MalformedReply(String), } /// A failed tunnel open or adopt. A failure never yields a session diff --git a/src/lib.rs b/src/lib.rs index 063ac63..08eade4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -30,14 +30,15 @@ pub mod local; pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession}; pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError}; pub use params::{ - establishment_reason, tunnel_direct_spec, SubstrateAddr, TunnelDirectParams, OP_TUNNEL_DIRECT, - TUNNEL_DIRECT_SCOPE, + establishment_reason, parse_listen_reply, tunnel_direct_spec, tunnel_forwarded_spec, + SubstrateAddr, TunnelDirectParams, TunnelForwardedParams, TunnelListenReply, OP_TUNNEL_DIRECT, + OP_TUNNEL_FORWARDED, TUNNEL_DIRECT_SCOPE, TUNNEL_FORWARDED_SCOPE, }; pub use params::{tunnel_open_spec, ChannelOpenError, Substrate, TunnelParams, OP_TUNNEL_OPEN}; pub use params::{TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; pub use producer::{ direct_establisher, direct_establisher_with_witness, register_tunnel_direct_openable, - register_tunnel_listen_openable, AcceptFn, AcceptQueue, DialFn, ResourceRegistry, TargetHandle, - TunnelEstablishError, + register_tunnel_listen_openable, AcceptFn, AcceptQueue, BindFirstPlan, DialFn, + ResourceRegistry, TargetHandle, TunnelEstablishError, }; pub use wire::MAX_DATAGRAM_LEN; diff --git a/src/params.rs b/src/params.rs index 0c9d205..dc48e63 100644 --- a/src/params.rs +++ b/src/params.rs @@ -10,6 +10,16 @@ //! `tunnel:direct` (a separate grant from `tunnel:open`, never implied //! by it). //! +//! The forwarded op's params ([`TunnelForwardedParams`], ADR-008 §3 +//! as amended) and its spec builder live here as well: +//! `{listen, substrate, peer}` — the accept-as-open ingress op, keyed +//! by the producer-minted correlation token (Amendment 1), scope-gated +//! by `tunnel:forwarded` (a separate grant again). The bind-first +//! listen open's reply projection ([`TunnelListenReply`] + +//! [`parse_listen_reply`]) is the typed view of the two additive +//! reply fields (`bound`, `listen`) the bind-first establisher +//! contributes. +//! //! The params layout is a one-way door once a consumer exists //! (ADR-001): `{resource, substrate}` only — the producer owns the //! backing; no URL-style addressing. `deny_unknown_fields` is the @@ -22,7 +32,9 @@ use alkcall::registry::spec::{ AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility, }; use serde::{Deserialize, Serialize}; -use serde_json::json; +use serde_json::{json, Value}; + +use crate::error::TunnelError; /// The `channels/tunnel/sub` operation id (the open op). pub const OP_TUNNEL_OPEN: &str = "channels/tunnel/sub"; @@ -45,6 +57,18 @@ pub const OP_TUNNEL_DIRECT: &str = "channels/tunnel/direct"; /// network" are different capabilities. Stable once published. pub const TUNNEL_DIRECT_SCOPE: &str = "tunnel:direct"; +/// The `channels/tunnel/forwarded` operation id (the forwarded op — +/// accept-as-open ingress, ADR-008 §3; the flavor-form op-id +/// convention, ADR-002 Amendment 1). +pub const OP_TUNNEL_FORWARDED: &str = "channels/tunnel/forwarded"; + +/// The scope gating the forwarded op (ADR-008 §3 / ADR-006 Amendment +/// 1). A separate grant from [`TUNNEL_OPEN_SCOPE`] and +/// [`TUNNEL_DIRECT_SCOPE`] — **never implied by either**: receiving +/// producer-initiated forwarded connections (ingress) is a different +/// capability from dialing egress. Stable once published. +pub const TUNNEL_FORWARDED_SCOPE: &str = "tunnel:forwarded"; + /// The open-op input params (ADR-001). Wire-stable: both fields /// required; renames are a breaking wire change. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -240,6 +264,218 @@ pub fn tunnel_direct_spec() -> OperationSpec { ) } +/// The bind-first listen open's reply projection (ADR-008 §2 as +/// amended): the two additive fields the bind-first establisher +/// contributes, extracted from the full success reply. +/// +/// **The struct is the projected pair, not a deny-unknown wire gate.** +/// The reply arrives as `{channel_id, bound?, listen?}` (the +/// wrapper-reserved `channel_id` key plus the establisher's additive +/// fields); unknown extra reply fields must be tolerated (the additive +/// posture — old consumers ignore unknown fields). Parse the full +/// reply with [`parse_listen_reply`], never with `deny_unknown_fields` +/// deserialization of the whole reply. Both fields are `None` for +/// accept-first/dial opens (the establisher supplied nothing). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TunnelListenReply { + /// The real OS-chosen bound address (port `0` resolved), + /// substrate-shaped — present only when the establisher supplied + /// it. Informational surface, NOT addressing: a consumer MUST NOT + /// treat `bound` as a dial target (the producer owns the backing — + /// ADR-001's invariant); it exists for RFC-1928-reply-class + /// fidelity (BND.ADDR) and observability. + #[serde(skip_serializing_if = "Option::is_none")] + pub bound: Option, + /// The correlation token (ADR-008 Amendment 1): the u32 the + /// establisher minted — the key the forwarded op's `listen` param + /// carries. Present only when the establisher supplied it. + #[serde(skip_serializing_if = "Option::is_none")] + pub listen: Option, +} + +/// Project the bind-first listen open's reply (ADR-008 §2 as amended) +/// out of the FULL success output [`Value`] — the shape +/// `ChannelClient::open_channel_with_reply` returns. Extracts the +/// optional `bound` + `listen` fields into [`TunnelListenReply`]. +/// +/// - Unknown extra reply fields are **tolerated** — the reply viewer +/// must not deny unknown fields (the additive posture; the wire is +/// plain JSON reply fields, and the projection is a typed view). +/// - `channel_id` must be present and a positive integer (the +/// wrapper-owned key — its absence means the caller is not looking +/// at an open reply at all; fail loud). +/// - A present-but-malformed `bound` (not substrate-shaped) or +/// `listen` (not a positive u32) is a loud error — a misbehaving +/// producer is never silently coerced ([`TunnelError:: +/// MalformedReply`]). +pub fn parse_listen_reply(reply: &Value) -> Result { + match reply.get("channel_id").and_then(Value::as_u64) { + Some(id) if id > 0 => {} + _ => { + return Err(TunnelError::MalformedReply( + "listen reply missing channel_id".to_string(), + )) + } + } + let bound = match reply.get("bound") { + None | Some(Value::Null) => None, + Some(v) => Some( + serde_json::from_value::(v.clone()).map_err(|e| { + TunnelError::MalformedReply(format!("bound not substrate-shaped: {e}")) + })?, + ), + }; + let listen = match reply.get("listen") { + None | Some(Value::Null) => None, + Some(v) => Some( + v.as_u64() + .filter(|n| *n > 0 && *n <= u32::MAX as u64) + .map(|n| n as u32) + .ok_or_else(|| { + TunnelError::MalformedReply("listen not a positive u32".to_string()) + })?, + ), + }; + Ok(TunnelListenReply { bound, listen }) +} + +/// The forwarded-op input params (ADR-008 §3 as amended). Wire-stable: +/// all three fields required; renames are a breaking wire change. +/// `deny_unknown_fields` is the loud-rejection posture (ADR-001's +/// precedent — unknown fields fail, never silently ignored). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TunnelForwardedParams { + /// The correlation token (ADR-008 Amendment 1): the u32 the + /// bind-first establisher minted and returned in the listen + /// open's `listen` reply field — NOT the listen channel's ID + /// (that key is superseded: the serving-side handler never sees + /// its own channel ID, and generic relays don't rewrite params). + /// Positive by wire contract (schema `minimum: 1`; the serde pin + /// is the backstop). + #[serde(deserialize_with = "positive_u32")] + pub listen: u32, + /// The listen tunnel's substrate; shapes the forwarded channel's + /// data-plane framing (ADR-003). + pub substrate: Substrate, + /// The accepted connection's remote address, keyed by the sibling + /// `substrate` field ([`SubstrateAddr`]). Required in v1 — a + /// producer that cannot determine the peer does not serve a + /// bind-first listener over that substrate (ADR-008 §3). + pub peer: SubstrateAddr, +} + +/// The serde-level pin behind `listen`'s schema `minimum: 1`: the +/// correlation token is positive by contract (the establisher seeds +/// the mint counter non-zero), so a `listen: 0` on the wire is a +/// malformed open — rejected at parse time, never silently accepted. +fn positive_u32<'de, D>(deserializer: D) -> Result +where + D: serde::Deserializer<'de>, +{ + let v = u32::deserialize(deserializer)?; + if v == 0 { + return Err(serde::de::Error::custom("listen must be > 0")); + } + Ok(v) +} + +/// The forwarded-op spec (`channels/tunnel/forwarded`, ADR-008 §3 as +/// amended): `Sub`-typed, external, channel-open marker on the +/// `alk/tunnel` ALPN, scope-gated by `tunnel:forwarded` (ingress — a +/// separate grant from `tunnel:open` and `tunnel:direct`, ADR-006 +/// Amendment 1), input/output schemas per wire.md §The Forwarded Op. +/// All three params required; the `peer` sub-schema is per-substrate — +/// `if`/`then` conditioned on the sibling `substrate` value (`tcp`/ +/// `udp` → `{host, port}`, `unix` → `{path}`), so a malformed peer +/// fails the registry's schema validation (the typed +/// `INVALID_INPUT` class) before the opener-side handler runs. +/// +/// **No establisher** on the forwarded op — the data plane starts +/// immediately (the opener-side serving registration is +/// `register_openable`, handler = correlate by `listen` + accept the +/// `BiStream` + pump; the correlation registry is assembly-side +/// bookkeeping, ADR-008 §3). +pub fn tunnel_forwarded_spec() -> OperationSpec { + OperationSpec::new( + OP_TUNNEL_FORWARDED, + OperationType::Sub, + Visibility::External, + json!({ + "type": "object", + "properties": { + "listen": { "type": "integer", "minimum": 1 }, + "substrate": { "type": "string", "enum": ["tcp", "udp", "unix"] } + }, + "required": ["listen", "substrate", "peer"], + "allOf": [ + { + "if": { + "properties": { "substrate": { "enum": ["tcp", "udp"] } }, + "required": ["substrate"] + }, + "then": { + "properties": { + "peer": { + "type": "object", + "properties": { + "host": { "type": "string" }, + "port": { + "type": "integer", + "minimum": 0, + "maximum": 65535 + } + }, + "required": ["host", "port"], + "additionalProperties": false + } + } + } + }, + { + "if": { + "properties": { "substrate": { "const": "unix" } }, + "required": ["substrate"] + }, + "then": { + "properties": { + "peer": { + "type": "object", + "properties": { + "path": { "type": "string" } + }, + "required": ["path"], + "additionalProperties": false + } + } + } + } + ] + }), + json!({ + "type": "object", + "properties": { + "channel_id": { "type": "integer", "minimum": 1 } + }, + "required": ["channel_id"] + }), + vec![], + AccessControl { + required_scopes: vec![TUNNEL_FORWARDED_SCOPE.to_string()], + required_scopes_any: None, + resource_type: None, + resource_action: None, + }, + None, + ) + .with_channel_open(ChannelOpenSpec::new(TUNNEL_ALPN)) + .with_description( + "Receive a producer-initiated forwarded connection (accept-as-open; \ + params: {listen, substrate, peer}; correlate by the `listen` token)", + ) +} + /// The establishment-failure reason code of a /// `channel:open_failed` error (`details.reason`): `dial_failed`, /// `unknown_resource`, `resource_shortage`, `handler_error`, @@ -557,4 +793,251 @@ mod tests { }; assert_eq!(unix.render_target(), "/run/postgresql/.s.PGSQL.5432"); } + + // --- forwarded op + the listen-reply projection (ADR-008, amended) -- + + #[test] + fn forwarded_params_round_trip_per_substrate() { + let cases: Vec<(Substrate, serde_json::Value, SubstrateAddr)> = vec![ + ( + Substrate::Tcp, + json!({"host": "198.51.100.4", "port": 55231}), + SubstrateAddr::HostPort { + host: "198.51.100.4".to_string(), + port: 55231, + }, + ), + ( + Substrate::Udp, + json!({"host": "10.0.0.7", "port": 53}), + SubstrateAddr::HostPort { + host: "10.0.0.7".to_string(), + port: 53, + }, + ), + ( + Substrate::Unix, + json!({"path": "/run/postgresql/.s.PGSQL.5432"}), + SubstrateAddr::Path { + path: "/run/postgresql/.s.PGSQL.5432".to_string(), + }, + ), + ]; + for (substrate, peer, addr) in cases { + let params = TunnelForwardedParams { + listen: 918273, + substrate, + peer: addr, + }; + let v = serde_json::to_value(¶ms).unwrap(); + assert_eq!( + v, + json!({ + "listen": 918273, + "substrate": substrate_str(substrate), + "peer": peer, + }) + ); + let back: TunnelForwardedParams = serde_json::from_value(v).unwrap(); + assert_eq!(back, params); + } + } + + #[test] + fn forwarded_params_unknown_fields_rejected() { + let v = json!({ + "listen": 918273, + "substrate": "tcp", + "peer": {"host": "h", "port": 1}, + "channel": 7 + }); + assert!(serde_json::from_value::(v).is_err()); + } + + #[test] + fn forwarded_params_missing_fields_rejected() { + let no_listen = json!({"substrate": "tcp", "peer": {"host": "h", "port": 1}}); + let no_substrate = json!({"listen": 1, "peer": {"host": "h", "port": 1}}); + let no_peer = json!({"listen": 1, "substrate": "tcp"}); + assert!(serde_json::from_value::(no_listen).is_err()); + assert!(serde_json::from_value::(no_substrate).is_err()); + assert!(serde_json::from_value::(no_peer).is_err()); + } + + #[test] + fn forwarded_params_listen_zero_rejected() { + let v = json!({ + "listen": 0, + "substrate": "tcp", + "peer": {"host": "h", "port": 1} + }); + assert!( + serde_json::from_value::(v).is_err(), + "the correlation token is positive by contract (schema minimum 1; \ + the serde pin is the backstop)" + ); + } + + #[test] + fn listen_reply_full_projection() { + let reply = json!({ + "channel_id": 7, + "bound": {"host": "203.0.113.9", "port": 42113}, + "listen": 918273 + }); + let projected = parse_listen_reply(&reply).unwrap(); + assert_eq!( + projected, + TunnelListenReply { + bound: Some(SubstrateAddr::HostPort { + host: "203.0.113.9".to_string(), + port: 42113, + }), + listen: Some(918273), + } + ); + } + + #[test] + fn listen_reply_bare_projection() { + // The accept-first/dial shape: the establisher contributed + // nothing — both fields None. + let reply = json!({"channel_id": 7}); + let projected = parse_listen_reply(&reply).unwrap(); + assert_eq!( + projected, + TunnelListenReply { + bound: None, + listen: None, + } + ); + } + + #[test] + fn listen_reply_tolerates_unknown_extra_fields() { + // The additive posture — the reply viewer must NOT deny + // unknown fields (future additive reply fields keep old + // consumers working). + let reply = json!({ + "channel_id": 7, + "bound": {"path": "/run/x.sock"}, + "listen": 5, + "future_field": {"nested": true} + }); + let projected = parse_listen_reply(&reply).unwrap(); + assert_eq!( + projected, + TunnelListenReply { + bound: Some(SubstrateAddr::Path { + path: "/run/x.sock".to_string() + }), + listen: Some(5), + } + ); + } + + #[test] + fn listen_reply_rejects_malformed_shapes() { + // No channel_id at all — not an open reply. + assert!(parse_listen_reply(&json!({"bound": {"path": "/p"}})).is_err()); + // channel_id = 0 (the wrapper never allocates 0 — channel 0 is + // the call plane). + assert!(parse_listen_reply(&json!({"channel_id": 0})).is_err()); + // Malformed bound (not substrate-shaped). + let bad_bound = json!({"channel_id": 7, "bound": {"port": 1}}); + assert!(parse_listen_reply(&bad_bound).is_err()); + // Malformed listen (negative-shaped / not a u32 / zero). + let bad_listen = json!({"channel_id": 7, "listen": 0}); + assert!(parse_listen_reply(&bad_listen).is_err()); + let bad_listen_type = json!({"channel_id": 7, "listen": "token"}); + assert!(parse_listen_reply(&bad_listen_type).is_err()); + } + + #[test] + fn forwarded_spec_matches_wire_md() { + let spec = tunnel_forwarded_spec(); + assert_eq!(spec.name, "channels/tunnel/forwarded"); + assert_eq!(spec.op_type, OperationType::Sub); + assert_eq!(spec.visibility, Visibility::External); + assert_eq!( + spec.input_schema["properties"]["listen"]["minimum"], + json!(1) + ); + assert_eq!( + spec.input_schema["properties"]["substrate"]["enum"], + json!(["tcp", "udp", "unix"]) + ); + assert_eq!( + spec.input_schema["required"], + json!(["listen", "substrate", "peer"]) + ); + assert_eq!(spec.output_schema["properties"]["channel_id"]["minimum"], 1); + assert_eq!( + spec.access_control.required_scopes, + vec![TUNNEL_FORWARDED_SCOPE.to_string()] + ); + let channel_open = spec.channel_open.as_ref().expect("channel-open marker"); + assert_eq!(channel_open.alpn, TUNNEL_ALPN); + assert!(spec.description.is_some()); + } + + #[test] + fn forwarded_spec_peer_schema_gates_shape_per_substrate() { + // The per-substrate peer sub-schema: the same if/then + // structure the direct op pins its target with — structural + // assertions per tunnels/params' precedent (no jsonschema dep + // here). + let schema = tunnel_forwarded_spec().input_schema; + let allof = schema["allOf"].as_array().expect("allOf branches"); + assert_eq!(allof.len(), 2, "tcp/udp branch + unix branch"); + let tcpudp_then = &allof[0]["then"]["properties"]["peer"]; + let unix_then = &allof[1]["then"]["properties"]["peer"]; + assert_eq!( + allof[0]["if"]["properties"]["substrate"]["enum"], + json!(["tcp", "udp"]) + ); + assert_eq!( + allof[1]["if"]["properties"]["substrate"]["const"], + json!("unix") + ); + assert_eq!( + tcpudp_then["required"], + json!(["host", "port"]), + "tcp/udp peer requires host+port" + ); + assert_eq!(tcpudp_then["properties"]["port"]["maximum"], json!(65535)); + assert_eq!(unix_then["required"], json!(["path"])); + assert_eq!( + tcpudp_then["additionalProperties"], + json!(false), + "cross-shape peer fields fail the schema" + ); + assert_eq!(unix_then["additionalProperties"], json!(false)); + assert_eq!(allof[0]["then"]["properties"]["peer"]["type"], "object"); + assert_eq!(allof[1]["then"]["properties"]["peer"]["type"], "object"); + } + + #[test] + fn forwarded_params_parse_backstops_the_schema() { + // The serde parse is the backstop for the shape it can use: + // a peer that schema-gates passed but parse-time mismatches + // (or a `listen: 0` riding a registry without the schema gate) + // fails loud at the deserializer. `deny_unknown_fields` holds: + // unknown top-level params reject; the untagged peer is + // lenient about extra peer fields (same documented asymmetry + // as the direct op's target — the SCHEMA is the shape-vs- + // substrate gate). + let cross_shape = json!({ + "listen": 1, + "substrate": "unix", + "peer": {"path": "/p", "host": "h"} + }); + let parsed: TunnelForwardedParams = serde_json::from_value(cross_shape).unwrap(); + assert_eq!( + parsed.peer, + SubstrateAddr::Path { + path: "/p".to_string() + } + ); + } } diff --git a/src/producer.rs b/src/producer.rs index 5bebdce..aaade0e 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -24,7 +24,14 @@ //! [`AcceptQueue`] below (the `tunnels/producer-listen` shape). The //! direct op's establisher ([`direct_establisher`] — dynamic-target //! egress, ADR-007) is the dial shape minus the registry lookup. +//! +//! The bind-first shape's plan payload ([`BindFirstPlan`] — the opaque +//! listener + the correlation token, ADR-008 as amended) is defined +//! here; the establisher and accept dispatcher that consume it land +//! with `tunnels/bindfirst-producer` (the wire-stable types pin +//! first, same ordering `tunnels/params` used for the base op). +use std::any::Any; use std::sync::Arc; use alkcall::channels::operations::{ @@ -40,7 +47,8 @@ use tokio::io::{AsyncRead, AsyncWrite}; use crate::params::{Substrate, TunnelDirectParams, TunnelParams}; pub use crate::params::{ - OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE, TUNNEL_OPEN_SCOPE, + OP_TUNNEL_DIRECT, OP_TUNNEL_FORWARDED, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE, + TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE, }; /// The dialed target handle — the plan payload (typed-opaque @@ -61,6 +69,49 @@ impl std::fmt::Debug for TargetHandle { } } +/// The bind-first establisher's plan payload (ADR-008 §1 as amended): +/// the assembly's listener handle + the correlation token. +/// +/// **The listener is NOT `TargetHandle` halves** — a listener is +/// accepted FROM, not pumped. The plan carries whatever the assembly's +/// `BindFn` produced, typed-opaque ([`Any`]; the same +/// injection pattern as `DialFn`/`AcceptFn` — the accept dispatcher +/// downcasts it to the assembly's listener type via a type-erased +/// accept closure; the ADR's "listener handle in the plan" is +/// assembly-typed, and the protocol sees only this opaque pair). +/// +/// **The correlation token (ADR-008 Amendment 1):** minted by the +/// bind-first establisher — a fresh u32 per open, unique per producer +/// (an `AtomicU32` counter seeded non-zero, e.g. 1, inside the +/// establisher's closure state; wraps are acceptable — u32, and a wrap +/// collision requires 2^32 concurrent opens on one producer). Carried +/// twice: in the plan (the accept dispatcher stamps it into every +/// forwarded open it opens for this listener) and as the additive +/// `listen` reply field (the opener correlates its waiting session on +/// it). It is NOT the listen channel's ID — the serving-side handler +/// never sees its own channel ID, and generic hub relays forward +/// params untouched (Amendment 1's rationale). Per-open fresh, +/// per-producer unique; retired with the listener session. +pub struct BindFirstPlan { + /// The assembly-typed listener handle (e.g. a `TcpListener` + /// wrapper the `BindFn` produced). Opaque to the protocol — the + /// accept dispatcher downcasts via a type-erased accept closure. + pub listener: Arc, + /// The correlation token (see above): minted fresh per open, + /// unique per producer, stamped into each forwarded open's + /// `listen` param. + pub token: u32, +} + +impl std::fmt::Debug for BindFirstPlan { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BindFirstPlan") + .field("listener", &"") + .field("token", &self.token) + .finish() + } +} + /// The typed establishment-failure vocabulary (ADR-049 §3's reason /// codes as this crate's establisher errors; mapped 1:1 onto /// `EstablishmentError`). @@ -487,3 +538,38 @@ pub fn register_tunnel_listen_openable( None, ) } + +#[cfg(test)] +mod tests { + use super::*; + + /// The plan-payload contract: `BindFirstPlan` rides the + /// typed-opaque `ChannelPlan` (`Arc`) and + /// the accept dispatcher recovers it by downcast — the same + /// establisher/handler agreement `TargetHandle` pins. The opaque + /// listener half downcasts to the assembly's concrete type. + #[test] + fn bindfirst_plan_downcasts_through_channel_plan() { + struct TestListener(u32); + let plan: ChannelPlan = Arc::new(BindFirstPlan { + listener: Arc::new(TestListener(42)) as Arc, + token: 918273, + }); + let unwrapped = plan + .downcast::() + .expect("BindFirstPlan plan"); + assert_eq!(unwrapped.token, 918273); + let listener = unwrapped + .listener + .downcast_ref::() + .expect("assembly-typed listener"); + assert_eq!(listener.0, 42); + // A wrong-type downcast fails (the opaque pair is not a + // TargetHandle). + let other: ChannelPlan = Arc::new(TargetHandle { + read: Box::new(tokio::io::empty()), + write: Box::new(tokio::io::sink()), + }); + assert!(other.downcast_ref::().is_none()); + } +} diff --git a/tasks/tunnels/bindfirst-params.md b/tasks/tunnels/bindfirst-params.md index 224ce8c..cb4f03f 100644 --- a/tasks/tunnels/bindfirst-params.md +++ b/tasks/tunnels/bindfirst-params.md @@ -1,7 +1,7 @@ --- id: tunnels/bindfirst-params name: Bind-first params — ListenParams, the forwarded-op spec, the bind-first plan type (ADR-008 as amended) -status: pending +status: completed depends_on: [] scope: narrow risk: medium @@ -129,14 +129,14 @@ pub struct BindFirstPlan { ## Acceptance Criteria -- [ ] `OP_TUNNEL_FORWARDED`/`TUNNEL_FORWARDED_SCOPE` constants; +- [x] `OP_TUNNEL_FORWARDED`/`TUNNEL_FORWARDED_SCOPE` constants; `tunnel_forwarded_spec()` matches wire.md §The Forwarded Op -- [ ] `TunnelForwardedParams` (deny-unknown, `listen: u32 > 0`) + the +- [x] `TunnelForwardedParams` (deny-unknown, `listen: u32 > 0`) + the reply-projection viewer round-trip the pinned shapes -- [ ] `BindFirstPlan` (opaque listener + token) defined; the token +- [x] `BindFirstPlan` (opaque listener + token) defined; the token mint documented (per-producer unique, per-open fresh) -- [ ] Unit tests green; wasm32 check passes (pure protocol types) -- [ ] Clippy/fmt clean +- [x] Unit tests green; wasm32 check passes (pure protocol types) +- [x] Clippy/fmt clean ## References @@ -150,8 +150,69 @@ pub struct BindFirstPlan { ## Notes -> Agent fills during implementation. +- `SubstrateAddr` had already landed (via `tunnels/direct-op`) — + consumed, not redefined (the "either order works" branch resolved + to the consuming side). +- The reply viewer is `parse_listen_reply(&Value) -> + Result`: parses the FULL success + output (the `open_channel_with_reply` shape), tolerates unknown + extra reply fields (the additive posture — no deny-unknown gate on + the reply), requires a positive `channel_id` (its absence/zero = + not an open reply — loud), and fails loud on malformed `bound` + (not substrate-shaped) or `listen` (not a positive u32) via the + new `TunnelError::MalformedReply` variant (error.rs). `null` + fields parse as `None` (defensive; a well-formed reply omits them). +- `TunnelForwardedParams.listen` carries a serde-level + `positive_u32` pin behind the schema's `minimum: 1` (the task's + "listen: 0 rejected (schema minimum 1 + serde-level pin)" — both + gates, test-pinned). +- The `peer` sub-schema follows the direct op's if/then precedent + (tcp/udp → `{host, port}` with `additionalProperties: false` + + `maximum: 65535`; unix → `{path}`), structural assertions per the + no-jsonschema-dep precedent. The untagged-peer lenient-extra-fields + asymmetry carries over from the direct op (schema = the shape gate; + serde = the backstop) — documented in the test. +- `BindFirstPlan` lives in `producer.rs` (the plan-payload family, + next to `TargetHandle`): `{ listener: Arc, + token: u32 }` per the honest shape — the listener is NOT + `TargetHandle` halves (a listener is accepted FROM, not pumped); + the accept dispatcher downcasts via a type-erased accept closure. + The token-mint discipline (AtomicU32 seeded non-zero, per-open + fresh, per-producer unique, wraps acceptable) is pinned in the + type's doc — the establisher that implements it lands with + `tunnels/bindfirst-producer`. +- Custom `Debug` impls (matching `TargetHandle`'s precedent): the + listener renders ``, the token renders plainly. +- Tests: 10 new unit tests in params.rs (forwarded round-trips per + substrate, unknown/missing-field rejection, `listen: 0` serde + rejection, full/bare/extra-field reply projection, malformed-reply + rejection, spec conformance, peer-schema structure, the + parse-backstop asymmetry) + the `BindFirstPlan` + downcast-through-`ChannelPlan` test in producer.rs (incl. the + wrong-type downcast failure). ## Summary -> Agent fills this on completion. \ No newline at end of file +Implemented the bind-first wire surface's params half per ADR-008 as +amended: `params.rs` — `OP_TUNNEL_FORWARDED` / `TUNNEL_FORWARDED_ +SCOPE` constants; `TunnelForwardedParams {listen, substrate, peer}` +(deny-unknown, `listen` positive via schema `minimum: 1` + the serde +pin); `tunnel_forwarded_spec()` (`Sub`-typed, External, +`ChannelOpenSpec::new(TUNNEL_ALPN)`, scope `["tunnel:forwarded"]`, +per-substrate if/then `peer` sub-schema, output +`{channel_id: integer > 0}`, description, no establisher); +`TunnelListenReply` + `parse_listen_reply` (the projected reply +viewer — unknown extra fields tolerated, `channel_id` presence +required, malformed fields fail loud via `TunnelError:: +MalformedReply`). `producer.rs` — `BindFirstPlan {listener: +Arc, token: u32}` with the mint discipline +pinned in docs (non-zero-seeded AtomicU32, per-open fresh, +per-producer unique, wraps acceptable). Re-exported through +`lib.rs`; the establisher/dispatcher/registration that consume these +types land with `tunnels/bindfirst-producer`. + +Verified: `cargo test` green (105 default; 125 with `--all- +features`), clippy `--all-targets -D warnings` clean (native + +wasm32), fmt clean, `wasm32-unknown-unknown` check passes (pure +protocol types — the default crate stays wasm-clean), `cargo doc +--no-deps` clean. CHANGELOG updated (Unreleased → Added). \ No newline at end of file