From 0789ab62cbfbc1b037dbc4bb7dd44337a9f6fd91 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 19 Sep 2026 04:58:37 +0000 Subject: [PATCH] feat: the direct op (channels/tunnel/direct, ADR-007) end to end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - params.rs: OP_TUNNEL_DIRECT / TUNNEL_DIRECT_SCOPE constants; SubstrateAddr (untagged {host,port} | {path}; also the ADR-008 §3 peer shape); TunnelDirectParams with render_target(); tunnel_direct_spec() — Sub-typed, External, alk/tunnel ALPN marker, scope ["tunnel:direct"] (never implied by tunnel:open), per-substrate if/then target sub-schema (malformed targets fail the registry's schema gate as INVALID_INPUT before the establisher). 7 unit tests. - producer.rs: direct_establisher (+ witness variant) — parse, render the target to the registry backing-string form, dial via the SAME injected DialFn; no registry lookup (unknown_resource can never fire); typed errors per ADR-007 §4 minus the registry row; CF-006 witness seam. register_tunnel_direct_openable reuses the base pump handler (plan-flow unchanged). - consumer.rs: TunnelSession::open_direct — separate constructor (two scopes are two capabilities), plain open_channel (dial establishers never bind), identical session/data-planes/teardown. - tests/producer_direct_op.rs: 15 integration tests — tcp + udp e2e round-trips, malformed-target rejection (no phantom session), dial-failure pass-through, NOT_FOUND posture, scope separation both ways (FORBIDDEN, raw-call + session-level), CF-006 witness (raw + session paths), parse backstop, target-rendering pin. Harness: Direct registration mode, wire_direct* topologies, direct_identity/both_scopes_identity. - CHANGELOG: Unreleased → Added. Verified: cargo test green (93 native, 94 --all-features); clippy -D warnings clean (native + wasm32); fmt clean; wasm32 check passes (default crate stays wasm-clean); cargo doc clean. Task: tasks/tunnels/direct-op.md (status: completed) --- CHANGELOG.md | 19 ++ src/consumer.rs | 45 ++- src/lib.rs | 7 +- src/params.rs | 341 ++++++++++++++++++++++ src/producer.rs | 88 +++++- tasks/tunnels/direct-op.md | 91 +++++- tests/harness.rs | 173 ++++++++++- tests/producer_direct_op.rs | 558 ++++++++++++++++++++++++++++++++++++ 8 files changed, 1304 insertions(+), 18 deletions(-) create mode 100644 tests/producer_direct_op.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 80a9b25..d1fb2c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,25 @@ is decomposed in `tasks/tunnels/` and lands in the next release. The dependency bump below is the only behavior-affecting change so far (no wire or API change at this level yet). +### Added + +- **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 + substrate-shaped address object, also the forwarded op's `peer` + shape) + `TunnelDirectParams {substrate, target}` with a + per-substrate `input_schema` (malformed targets fail the registry's + schema gate — `INVALID_INPUT` — before the establisher), + `tunnel_direct_spec()`, `direct_establisher` (+ witness variant; + no registry lookup — `unknown_resource` can never fire; targets + render to the same backing-string form the registry backings use + and ride the SAME injected `DialFn`), `register_tunnel_direct_ + openable` (same pump handler + plan-flow as the base op), and + `TunnelSession::open_direct` (separate constructor — `tunnel:direct` + is a separate grant from `tunnel:open`, never implied by it; the + session type, data planes, and teardown API are identical). An old + producer rejects the unknown op loudly (`NOT_FOUND`). + ### Changed - **alkcall 0.7.1 → 0.8.0** (crates.io). The two upstream asks the diff --git a/src/consumer.rs b/src/consumer.rs index 6aa6030..647237c 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -22,7 +22,10 @@ use bytes::Bytes; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf}; use crate::error::{ReverseOpenError, TunnelIoError, TunnelOpenError}; -use crate::params::{Substrate, TunnelParams, OP_TUNNEL_OPEN, TUNNEL_ALPN}; +use crate::params::{ + Substrate, SubstrateAddr, TunnelDirectParams, TunnelParams, OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, + TUNNEL_ALPN, +}; use crate::wire::{frame_datagram, DatagramReader}; /// The typed consumer session (ADR-005): one session = one channel = @@ -104,6 +107,46 @@ impl TunnelSession { )) } + /// Open a tunnel channel toward a CALLER-NAMED target (the direct + /// op — dynamic-target egress, ADR-007; the ssh `direct-tcpip` / + /// SOCKS5 CONNECT shape). A separate constructor from [`open`](Self::open), + /// not a params enum: `tunnel:direct` and `tunnel:open` are two + /// scopes — two capabilities — and the caller states which one it + /// holds. + /// + /// Serializes `{substrate, target}`, calls + /// `open_channel(OP_TUNNEL_DIRECT, …, TUNNEL_ALPN)` (plain + /// `open_channel` — dial establishers never bind, so no reply + /// fields are expected), and presents the identical session type, + /// data planes, and teardown API as [`open`](Self::open). + /// + /// Error surface: the base open errors minus `unknown_resource` + /// (no registry lookup); a malformed target is the schema / + /// `invalid_input`-class failure (registry-side); an old producer + /// rejects the unknown op loudly (`NOT_FOUND` — the SSH posture). + pub async fn open_direct( + client: &ChannelClient, + target: SubstrateAddr, + substrate: Substrate, + ) -> Result { + let input = + serde_json::to_value(&TunnelDirectParams { substrate, target }).map_err(|e| { + TunnelOpenError::Open(ChannelOpenError::CallFailed { + error: CallError::internal(format!("params serialize: {e}")), + }) + })?; + let (channel_id, send, recv) = client + .open_channel(OP_TUNNEL_DIRECT, input, TUNNEL_ALPN) + .await?; + Ok(Self::from_halves( + channel_id, + substrate, + client.manager().clone(), + send, + recv, + )) + } + /// Adopt a producer-allocated channel ID (the reverse path, `-R`): /// install the data plane from the adopted halves. Early arrivals /// are parked by the manager and drained on adopt (the adoption diff --git a/src/lib.rs b/src/lib.rs index bf1e264..063ac63 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -29,9 +29,14 @@ pub mod local; pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession}; pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError}; -pub use params::{establishment_reason, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; +pub use params::{ + establishment_reason, tunnel_direct_spec, SubstrateAddr, TunnelDirectParams, OP_TUNNEL_DIRECT, + TUNNEL_DIRECT_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, }; diff --git a/src/params.rs b/src/params.rs index 5a92bf4..0c9d205 100644 --- a/src/params.rs +++ b/src/params.rs @@ -3,6 +3,13 @@ //! `OperationSpec` builder, the scope/ALPN/op-id constants, and the //! typed establishment-reason helper (ADR-049 §4 surface). //! +//! The direct op's params ([`TunnelDirectParams`] + [`SubstrateAddr`], +//! ADR-007) and its spec builder live here too: `{substrate, target}` +//! — the caller names the dial target at session time (the ssh +//! `direct-tcpip` / SOCKS5 CONNECT shape), scope-gated by +//! `tunnel:direct` (a separate grant from `tunnel:open`, never implied +//! by it). +//! //! 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 @@ -27,6 +34,17 @@ pub const TUNNEL_ALPN: &str = "alk/tunnel"; /// The scope gating tunnel opens (ADR-006). Stable once published. pub const TUNNEL_OPEN_SCOPE: &str = "tunnel:open"; +/// The `channels/tunnel/direct` operation id (the direct op — +/// dynamic-target egress, ADR-007; the flavor-form op-id convention, +/// ADR-002 Amendment 1). +pub const OP_TUNNEL_DIRECT: &str = "channels/tunnel/direct"; + +/// The scope gating the direct op (ADR-007 §3 / ADR-006 Amendment 1). +/// A separate grant from [`TUNNEL_OPEN_SCOPE`] — **never implied by +/// it**: "dial `postgres-primary`" and "dial anything from my +/// network" are different capabilities. Stable once published. +pub const TUNNEL_DIRECT_SCOPE: &str = "tunnel:direct"; + /// 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)] @@ -52,6 +70,50 @@ pub enum Substrate { Unix, } +/// The substrate-shaped address object (ADR-007 §2): keyed by the +/// sibling `substrate` field of the params — `tcp`/`udp` → +/// [`SubstrateAddr::HostPort`] (domain-form hosts stay unresolved on +/// the wire — producer-side resolution, the ssh `-D` semantic), +/// `unix` → [`SubstrateAddr::Path`]. Untagged: the shapes are +/// mutually exclusive in practice and self-describing. Also the +/// `peer` value shape of the forwarded op (ADR-008 §3) — defined +/// once here. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(untagged)] +pub enum SubstrateAddr { + HostPort { host: String, port: u16 }, + Path { path: String }, +} + +/// The direct-op input params (ADR-007 §2). Wire-stable: both fields +/// required; renames are a breaking wire change. The target's +/// shape-vs-substrate correspondence is pinned by the spec's +/// `input_schema` (per-substrate `oneOf`, validated registry-side +/// before the establisher) and re-checked by serde here (the +/// backstop — a mismatch parses as the wrong variant or fails). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TunnelDirectParams { + /// The substrate discriminator; selects the data-plane framing + /// (ADR-003) and the target's expected shape. + pub substrate: Substrate, + /// The dial target, keyed by `substrate` ([`SubstrateAddr`]). + pub target: SubstrateAddr, +} + +impl TunnelDirectParams { + /// Render the target to the backing-string form the `DialFn` / + /// registry backings use: `"host:port"` for tcp/udp, the path for + /// unix. The direct establisher calls this before dialing — the + /// dial closure stays substrate-agnostic (ADR-007 §4). + pub fn render_target(&self) -> String { + match &self.target { + SubstrateAddr::HostPort { host, port } => format!("{host}:{port}"), + SubstrateAddr::Path { path } => path.clone(), + } + } +} + /// The open-op spec (`channels/tunnel/sub`): `Sub`-typed, external, /// channel-open marker on the `alk/tunnel` ALPN, scope-gated by /// `tunnel:open` (ADR-006), input/output schemas per wire.md §The @@ -91,6 +153,93 @@ pub fn tunnel_open_spec() -> OperationSpec { ) } +/// The direct-op spec (`channels/tunnel/direct`, ADR-007): `Sub`-typed, +/// external, channel-open marker on the `alk/tunnel` ALPN, scope-gated +/// by `tunnel:direct` (a separate grant from `tunnel:open` — never +/// implied by it, ADR-006 Amendment 1), input/output schemas per +/// wire.md §The Direct Op. The target sub-schema is per-substrate — +/// `if`/`then` conditioned on the sibling `substrate` value +/// (`tcp`/`udp` → `{host, port}`, `unix` → `{path}`) — so a malformed +/// target fails the registry's schema validation (the typed +/// `INVALID_INPUT` class) before the establisher runs. +pub fn tunnel_direct_spec() -> OperationSpec { + OperationSpec::new( + OP_TUNNEL_DIRECT, + OperationType::Sub, + Visibility::External, + json!({ + "type": "object", + "properties": { + "substrate": { "type": "string", "enum": ["tcp", "udp", "unix"] } + }, + "required": ["substrate", "target"], + "allOf": [ + { + "if": { + "properties": { "substrate": { "enum": ["tcp", "udp"] } }, + "required": ["substrate"] + }, + "then": { + "properties": { + "target": { + "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": { + "target": { + "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_DIRECT_SCOPE.to_string()], + required_scopes_any: None, + resource_type: None, + resource_action: None, + }, + None, + ) + .with_channel_open(ChannelOpenSpec::new(TUNNEL_ALPN)) + .with_description( + "Open a tunnel channel to a caller-named target (params: {substrate, target}; \ + dynamic-target egress)", + ) +} + /// The establishment-failure reason code of a /// `channel:open_failed` error (`details.reason`): `dial_failed`, /// `unknown_resource`, `resource_shortage`, `handler_error`, @@ -216,4 +365,196 @@ mod tests { "tcp" | "udp" | "unix" if allowed.contains(&substrate_str(params.substrate)) ) } + + // --- direct op (ADR-007) ------------------------------------------- + + #[test] + fn direct_params_round_trip_per_substrate() { + let cases: Vec<(Substrate, serde_json::Value, SubstrateAddr)> = vec![ + ( + Substrate::Tcp, + json!({"host": "db.internal", "port": 5432}), + SubstrateAddr::HostPort { + host: "db.internal".to_string(), + port: 5432, + }, + ), + ( + 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, target, addr) in cases { + let params = TunnelDirectParams { + substrate, + target: addr.clone(), + }; + let v = serde_json::to_value(¶ms).unwrap(); + assert_eq!( + v, + json!({"substrate": substrate_str(substrate), "target": target}) + ); + let back: TunnelDirectParams = serde_json::from_value(v).unwrap(); + assert_eq!(back, params); + } + } + + #[test] + fn substrate_addr_untagged_disambiguation() { + let hostport: SubstrateAddr = + serde_json::from_value(json!({"host": "h", "port": 1})).unwrap(); + let path: SubstrateAddr = serde_json::from_value(json!({"path": "/p"})).unwrap(); + assert_eq!( + hostport, + SubstrateAddr::HostPort { + host: "h".to_string(), + port: 1 + } + ); + assert_eq!( + path, + SubstrateAddr::Path { + path: "/p".to_string() + } + ); + // Host/port without the host field fails; an array fails. + assert!(serde_json::from_value::(json!({"port": 1})).is_err()); + assert!(serde_json::from_value::(json!([1, 2])).is_err()); + } + + #[test] + fn direct_params_unknown_fields_rejected() { + let v = json!({ + "substrate": "tcp", + "target": {"host": "h", "port": 1}, + "resource": "sneaky" + }); + assert!(serde_json::from_value::(v).is_err()); + // The untagged target is lenient about extra target fields + // (untagged variants carry no deny_unknown_fields) — a + // cross-shape target parses as whichever variant fits. The + // SCHEMA is the gate that rejects shape-vs-substrate + // mismatches registry-side (additionalProperties: false per + // branch; direct_target_schema_gates_shape_per_substrate). + // The establisher's serde parse is the backstop for the + // shape it can actually use (render_target keys off the + // sibling substrate). + let cross_shape = json!({ + "substrate": "unix", + "target": {"path": "/p", "host": "h"} + }); + let parsed: TunnelDirectParams = serde_json::from_value(cross_shape).unwrap(); + assert_eq!( + parsed.target, + SubstrateAddr::Path { + path: "/p".to_string() + } + ); + } + + #[test] + fn direct_params_missing_fields_rejected() { + let no_target = json!({"substrate": "tcp"}); + let no_substrate = json!({"target": {"host": "h", "port": 1}}); + assert!(serde_json::from_value::(no_target).is_err()); + assert!(serde_json::from_value::(no_substrate).is_err()); + } + + #[test] + fn direct_spec_matches_wire_md() { + let spec = tunnel_direct_spec(); + assert_eq!(spec.name, "channels/tunnel/direct"); + assert_eq!(spec.op_type, OperationType::Sub); + assert_eq!(spec.visibility, Visibility::External); + assert_eq!( + spec.input_schema["properties"]["substrate"]["enum"], + json!(["tcp", "udp", "unix"]) + ); + assert_eq!( + spec.input_schema["required"], + json!(["substrate", "target"]) + ); + assert_eq!(spec.output_schema["properties"]["channel_id"]["minimum"], 1); + assert_eq!( + spec.access_control.required_scopes, + vec![TUNNEL_DIRECT_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 direct_target_schema_gates_shape_per_substrate() { + // The oneOf/if-then structure: per-substrate target shapes are + // enforced by the schema itself (registry-side, before the + // establisher). Structural assertions per tunnels/params' + // precedent (no jsonschema dep here). + let schema = tunnel_direct_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"]["target"]; + let unix_then = &allof[1]["then"]["properties"]["target"]; + 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 target 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 fields fail the schema" + ); + assert_eq!(unix_then["additionalProperties"], json!(false)); + assert_eq!(allof[0]["then"]["properties"]["target"]["type"], "object"); + assert_eq!(allof[1]["then"]["properties"]["target"]["type"], "object"); + } + + #[test] + fn render_target_matches_registry_backing_form() { + let tcp = TunnelDirectParams { + substrate: Substrate::Tcp, + target: SubstrateAddr::HostPort { + host: "db.internal".to_string(), + port: 5432, + }, + }; + assert_eq!(tcp.render_target(), "db.internal:5432"); + let udp = TunnelDirectParams { + substrate: Substrate::Udp, + target: SubstrateAddr::HostPort { + host: "10.0.0.7".to_string(), + port: 53, + }, + }; + assert_eq!(udp.render_target(), "10.0.0.7:53"); + let unix = TunnelDirectParams { + substrate: Substrate::Unix, + target: SubstrateAddr::Path { + path: "/run/postgresql/.s.PGSQL.5432".to_string(), + }, + }; + assert_eq!(unix.render_target(), "/run/postgresql/.s.PGSQL.5432"); + } } diff --git a/src/producer.rs b/src/producer.rs index 2f6e540..5bebdce 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -21,7 +21,9 @@ //! //! The listen establisher variant (the producer-side listener — the //! `-R` far-side listener shape) is [`listen_establisher`] + -//! [`AcceptQueue`] below (the `tunnels/producer-listen` shape). +//! [`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. use std::sync::Arc; @@ -35,9 +37,11 @@ use alkcall::registry::registration::OperationRegistry; use serde_json::Value; use tokio::io::{AsyncRead, AsyncWrite}; -use crate::params::{Substrate, TunnelParams}; +use crate::params::{Substrate, TunnelDirectParams, TunnelParams}; -pub use crate::params::{OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; +pub use crate::params::{ + OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE, TUNNEL_OPEN_SCOPE, +}; /// The dialed target handle — the plan payload (typed-opaque /// `ChannelPlan`; establisher and handler agree on the concrete @@ -189,6 +193,57 @@ fn parse_params(input: &Value) -> Result { .map_err(|e| TunnelEstablishError::HandlerError(format!("invalid params: {e}"))) } +/// The direct establisher (ADR-007 §4 — the dial shape minus the +/// registry lookup): parse `TunnelDirectParams`, render the target to +/// the same backing-string form the registry backings use +/// (`"host:port"` for tcp/udp, the path for unix), and call the +/// injected [`DialFn`] — the SAME dial function the base op uses, so +/// a producer that registered `dial_tcp` for the base op registers +/// the direct op with zero new substrate code (the UDP framed-adapter +/// wrap, ADR-003, happens inside the dial closure — the establisher +/// stays substrate-agnostic). No registry lookup — `unknown_resource` +/// can never fire. +/// +/// The per-call `auth` carries the opener identity (CF-006); record it +/// into `identity_witness` when provided (the direct op is the +/// arbitrary-egress capability — identity visibility matters more, +/// not less). +/// +/// Error mapping (ADR-007 §4's table minus the registry row): the +/// serde-parse failure of a schema-valid-shaped target maps +/// [`TunnelEstablishError::HandlerError`] (the schema gate should +/// have caught it — the parse is the backstop); dial failures pass +/// through the `DialFn`'s own typed errors unchanged (`dial_failed` / +/// `resource_shortage` / `handler_error`). +pub fn direct_establisher(dial: DialFn) -> OpenEstablisher { + direct_establisher_with_witness(dial, None) +} + +/// [`direct_establisher`] with an optional identity witness (the test +/// seam for the CF-006 contract, same as the base op's). +pub fn direct_establisher_with_witness( + dial: DialFn, + identity_witness: Option, +) -> OpenEstablisher { + Arc::new(move |input: Value, auth| { + let dial = Arc::clone(&dial); + let witness = identity_witness.clone(); + Box::pin(async move { + if let Some(w) = &witness { + *w.lock().await = auth.identity.as_ref().map(|i| i.id.clone()); + } + let params: TunnelDirectParams = + serde_json::from_value(input.clone()).map_err(|e| { + TunnelEstablishError::HandlerError(format!("invalid direct params: {e}")) + })?; + let rendered = params.render_target(); + let substrate = params.substrate; + let dialed: TargetHandle = dial(substrate, &rendered).await?; + Ok(Establishment::new(Arc::new(dialed) as ChannelPlan)) + }) + }) +} + /// The listen establisher (shape 2 — producer.md §The Establisher): /// the same open op, same params, same typed errors; the establisher /// pops the next accepted connection from the injected [`AcceptFn`] @@ -381,6 +436,33 @@ pub fn register_tunnel_openable( ) } +/// Register the direct op (`channels/tunnel/direct`, ADR-007) on a +/// `ChannelCore` + the session's dispatch registry: the direct spec + +/// [`direct_establisher_with_witness`] + the SAME pump handler as the +/// base op (the direct op's plan payload is a [`TargetHandle`] exactly +/// like the base op's; plan-flow is unchanged), via +/// `register_openable_with_establisher` (`timeout: None` = the 10s +/// default bound). `dial` is the SAME [`DialFn`] the base op uses — +/// no new substrate code. `identity_witness` is the optional CF-006 +/// probe seam (the establisher records the per-call opener identity +/// it saw). +pub fn register_tunnel_direct_openable( + core: &ChannelCore, + on_registry: &Arc, + auth: AuthContext, + dial: DialFn, + identity_witness: Option, +) -> Result<(), String> { + core.register_openable_with_establisher( + crate::params::tunnel_direct_spec(), + Some(direct_establisher_with_witness(dial, identity_witness)), + make_tunnel_pump_handler(), + on_registry, + auth, + None, + ) +} + /// Register the tunnel open op with a LISTEN establisher (shape 2): /// the same spec + pump handler as [`register_tunnel_openable`]; the /// establisher pops accepted handles from the injected [`AcceptFn`] diff --git a/tasks/tunnels/direct-op.md b/tasks/tunnels/direct-op.md index 8cb9e8a..b07ff96 100644 --- a/tasks/tunnels/direct-op.md +++ b/tasks/tunnels/direct-op.md @@ -1,7 +1,7 @@ --- id: tunnels/direct-op name: Direct op — params, spec, establisher, consumer open_direct (ADR-007) -status: pending +status: completed depends_on: [] scope: broad risk: medium @@ -146,20 +146,20 @@ Integration (extend the duplex harness from `tunnels/producer-open-op`): ## Acceptance Criteria -- [ ] `TunnelDirectParams`/`SubstrateAddr` round-trip; malformed +- [x] `TunnelDirectParams`/`SubstrateAddr` round-trip; malformed targets rejected (serde + schema shape pinned) -- [ ] `tunnel_direct_spec()` matches wire.md §The Direct Op exactly -- [ ] `direct_establisher`: no registry lookup, target rendering into +- [x] `tunnel_direct_spec()` matches wire.md §The Direct Op exactly +- [x] `direct_establisher`: no registry lookup, target rendering into the `DialFn` backing string, typed errors per ADR-007 §4 -- [ ] `register_tunnel_direct_openable` reuses the base pump handler + +- [x] `register_tunnel_direct_openable` reuses the base pump handler + dial injection; witness seam present -- [ ] `TunnelSession::open_direct` (separate ctor; identical +- [x] `TunnelSession::open_direct` (separate ctor; identical session/teardown semantics) -- [ ] Integration tests: e2e round-trips (tcp + udp), malformed-target +- [x] Integration tests: e2e round-trips (tcp + udp), malformed-target rejection, scope separation (FORBIDDEN), NOT_FOUND posture, per-call identity witness -- [ ] Clippy/fmt clean; wasm32 check passes (no substrate code added) -- [ ] `cargo test` green +- [x] Clippy/fmt clean; wasm32 check passes (no substrate code added) +- [x] `cargo test` green ## References @@ -172,8 +172,77 @@ Integration (extend the duplex harness from `tunnels/producer-open-op`): ## Notes -> Agent fills during implementation. +- The `input_schema` pins the per-substrate target sub-schema as two + `if`/`then` branches under `allOf`, keyed on the sibling + `substrate` value (tcp/udp → `{host, port}` with + `additionalProperties: false`; unix → `{path}`, same posture) — + JSON Schema `if/then` composes through alkcall's jsonschema 0.46 + compile, and the integration tests prove the gate fires + (`INVALID_INPUT` before the establisher). Structural assertions + pin the branch shape per `tunnels/params`'s no-jsonschema-dep + precedent. +- Untagged `SubstrateAddr` is deliberately lenient about extra target + fields (untagged variants carry no `deny_unknown_fields`): a + cross-shape target parses as whichever variant fits. The schema is + the shape-vs-substrate gate; the establisher's serde parse is the + backstop only (unknown substrate value → `handler_error`, + `direct_establisher` maps `TunnelEstablishError::HandlerError`). + This asymmetry is documented in both the type docs and the test. +- `TunnelDirectParams::render_target()` centralizes the backing-string + form (`"host:port"` / path) — the establisher injects the SAME + `DialFn` with the rendered string; the render is pinned by test to + the registry-backing form. +- `TunnelSession::open_direct` uses plain `open_channel` (no reply + fields — dial establishers never bind); `SubstrateAddr` re-exports + as the future forwarded-op `peer` shape (ADR-008 §3, defined once). +- Harness: `RegistrationMode::Direct`, `wire_direct` / + `wire_direct_with_identity` (the raw-call topology), + `wire_direct_forward` / `wire_direct_forward_with_identity` (the + session-level forward topology; identity parameterized so the + session-level FORBIDDEN probe runs through the real client API), + `direct_identity` / `both_scopes_identity`. ## Summary -> Agent fills this on completion. \ No newline at end of file +Implemented the direct op end to end per ADR-007: + +- **params.rs** — `OP_TUNNEL_DIRECT` / `TUNNEL_DIRECT_SCOPE` + constants; `SubstrateAddr` (untagged `{host, port}` | `{path}`, + the ADR-008 §3 `peer` shape too); `TunnelDirectParams` + (`deny_unknown_fields`, `render_target()`); `tunnel_direct_spec()` + (`Sub`-typed, External, `ChannelOpenSpec::new(TUNNEL_ALPN)`, scope + `["tunnel:direct"]`, output `{channel_id: integer > 0}`, + per-substrate `if/then` target sub-schema). 7 unit tests: round- + trips per substrate, untagged disambiguation, loud unknown-field + rejection, spec-shape conformance, schema structure. +- **producer.rs** — `direct_establisher` / + `direct_establisher_with_witness` (parse → render → injected + `DialFn`; no registry lookup; typed-error mapping per ADR-007 §4 + minus the registry row; CF-006 witness seam) + + `register_tunnel_direct_openable` (direct spec + establisher + the + SAME `make_tunnel_pump_handler`, plan-flow unchanged). +- **consumer.rs** — `TunnelSession::open_direct(client, target, + substrate)`: separate constructor, serializes `{substrate, target}`, + plain `open_channel(OP_TUNNEL_DIRECT, …, TUNNEL_ALPN)`, + `from_halves` — identical session/data-planes/teardown as `open`. +- **lib.rs** — the new surface re-exported. + +Tests: `tests/producer_direct_op.rs` — 15 integration tests over the +duplex harness: tcp + udp e2e round-trips through `pump_bidi` +(datagram boundary preserved via the framed dial closure), malformed +targets (`unix` without `path`, `tcp` without `port`, missing target) +→ `INVALID_INPUT` with no phantom channel/session, dial-failure +pass-through (`dial_failed`), NOT_FOUND on a producer without the op, +scope separation both ways (`tunnel:open` ↛ `tunnel:direct` and +`tunnel:direct` ↛ `tunnel:open` — FORBIDDEN, raw-call AND session- +level), CF-006 witness on the direct establisher (raw + session +paths), the establisher parse backstop (`handler_error`), and the +target-rendering pin (the dial closure observes the exact +backing-string form). Harness: `Direct` registration mode + +`wire_direct*` topologies + `direct_identity`/`both_scopes_identity`. + +Verified: `cargo test` green (93 native; 94 with `--all-features`), +clippy `-D warnings` clean (native + wasm32), fmt clean, wasm32 +check passes (default crate stays wasm-clean — no substrate code +added), `cargo doc --no-deps` clean. CHANGELOG updated (Unreleased → +Added). \ No newline at end of file diff --git a/tests/harness.rs b/tests/harness.rs index db50646..a26b7ee 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -34,8 +34,10 @@ use alkcall::protocol::connection::{split_single_stream, CallConnection}; use alkcall::protocol::dispatch::Dispatcher; use alkcall::registry::registration::OperationRegistry; -use alktunnels::params::{Substrate, TUNNEL_OPEN_SCOPE}; -use alktunnels::producer::{register_tunnel_openable, ResourceRegistry}; +use alktunnels::params::{Substrate, TUNNEL_DIRECT_SCOPE, TUNNEL_OPEN_SCOPE}; +use alktunnels::producer::{ + register_tunnel_direct_openable, register_tunnel_openable, ResourceRegistry, +}; /// The consumer identity the transport carries (the mTLS/QUIC /// analogue — key-based identity resolved out-of-band, attached to @@ -48,6 +50,29 @@ pub fn consumer_identity() -> Identity { } } +/// The direct-egress identity: `tunnel:direct` ONLY (no +/// `tunnel:open`) — the scope-separation probe's identity. +pub fn direct_identity() -> Identity { + Identity { + id: "consumer-direct".to_string(), + scopes: vec![TUNNEL_DIRECT_SCOPE.to_string()], + resources: std::collections::HashMap::new(), + } +} + +/// Both scopes: `tunnel:open` AND `tunnel:direct` (a deployment +/// granting both grants both explicitly — ADR-006 Amendment 1). +pub fn both_scopes_identity() -> Identity { + Identity { + id: "consumer-both".to_string(), + scopes: vec![ + TUNNEL_OPEN_SCOPE.to_string(), + TUNNEL_DIRECT_SCOPE.to_string(), + ], + resources: std::collections::HashMap::new(), + } +} + /// The scoped token (the hub-forwarding path) — resolves to /// [`consumer_identity`] via [`TestIdProvider`]. pub const TEST_AUTH_TOKEN: &str = "test-token-with-tunnel-open-scope"; @@ -105,6 +130,12 @@ pub enum RegistrationMode { /// The no-witness dial establisher (`tunnel_establisher` — the /// public default shape, no CF-006 probe seam). DialNoWitness(alktunnels::producer::DialFn), + /// The direct op (`channels/tunnel/direct`, ADR-007): the direct + /// spec + the direct establisher (no registry lookup) over the + /// SAME injected dial, with the CF-006 witness seam. The witness + /// rides the topology's `identity_witness` (the direct op's + /// witness and the base op's never mix in one test). + Direct(alktunnels::producer::DialFn), } pub async fn wire_with( @@ -212,6 +243,16 @@ pub async fn wire_with( ) .expect("register tunnel openable"); } + RegistrationMode::Direct(dial) => { + register_tunnel_direct_openable( + &core, + &producer_op_registry, + AuthContext::anonymous(b"alk/tunnel"), + dial, + Some(Arc::clone(&identity_witness)), + ) + .expect("register tunnel direct openable"); + } RegistrationMode::Listen(accept) => { alktunnels::producer::register_tunnel_listen_openable( &core, @@ -308,6 +349,33 @@ pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn .await } +/// The direct-op topology (ADR-007): the direct op registered with +/// the CF-006 witness; the consumer's transport identity is +/// [`both_scopes_identity`] (both grants — the opens authorize). +pub async fn wire_direct(dial: alktunnels::producer::DialFn) -> Topology { + wire_with( + ResourceRegistry::new(), + RegistrationMode::Direct(dial), + Some(both_scopes_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + +/// [`wire_direct`] with an explicit transport identity (the +/// scope-separation probes construct their own identities). +pub async fn wire_direct_with_identity(identity: Identity) -> Topology { + wire_with( + ResourceRegistry::new(), + RegistrationMode::Direct(failing_dial("unused")), + Some(identity), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + /// A topology with the consumer's transport identity REPLACED by the /// serving-side override (`ServingConfig.identity` — CF-005 (a) probe /// shape) or stripped entirely (the fail-closed probe). The transport @@ -472,6 +540,107 @@ pub async fn wire_forward( } } +/// The DIRECT-op forward topology (ADR-007): the producer (accept +/// side, serving) registers the direct op — [`register_tunnel_direct_openable`] +/// over the injected dial, CF-006 witness attached — and the consumer +/// (connect side, pure client) holds the identity the session tests +/// authorize with. This is the topology `TunnelSession::open_direct` +/// is normative for. +pub async fn wire_direct_forward(dial: alktunnels::producer::DialFn) -> ForwardTopology { + wire_direct_forward_with_identity(dial, both_scopes_identity()).await +} + +/// [`wire_direct_forward`] with an explicit transport identity (the +/// session-level scope-separation probes construct their own). +pub async fn wire_direct_forward_with_identity( + dial: alktunnels::producer::DialFn, + identity: Identity, +) -> ForwardTopology { + let producer_op_registry = Arc::new(OperationRegistry::new()); + let (producer_manager_tx, mut producer_manager_rx) = + tokio::sync::mpsc::channel::(1); + let identity_witness = Arc::new(tokio::sync::Mutex::new(None::)); + let dial_witness = Arc::clone(&identity_witness); + let expected_caller = identity.clone(); + let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| { + let registry_arc = Arc::clone(&producer_op_registry); + let dial = Arc::clone(&dial); + let witness = Arc::clone(&dial_witness); + let manager_tx = producer_manager_tx.clone(); + let caller = expected_caller.clone(); + tokio::spawn(async move { + let channel0_bidi = match channel0_conn.accept_bi().await { + Ok(s) => s, + Err(_) => return, + }; + let (writer, reader) = split_single_stream(channel0_bidi); + // The accept-side transport-identity posture: the transport + // authenticated the dialer; attach the RESOLVED identity + // (scopes included — resolved out-of-band) to channel 0. + let _ = channel0_conn.set_identity(caller); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + let core = alkcall::channels::operations::ChannelCore::new( + manager.clone(), + alkcall::channels::policy::default_policy(), + ); + let ops = ChannelOperations::with_default_policy(manager.clone()); + ops.register_on(®istry_arc) + .expect("register channel ops on producer"); + register_tunnel_direct_openable( + &core, + ®istry_arc, + AuthContext::anonymous(b"alk/tunnel"), + dial, + Some(witness), + ) + .expect("register tunnel direct openable"); + let _ = manager_tx.send(manager).await; + let dp = Dispatcher::new( + registry_arc, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }); + + let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024); + let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); + let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + consumer_conn + .set_identity(identity) + .expect("transport identity set once"); + + let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); + let producer_transport_auth = AuthContext::anonymous(b"alk/channels"); + let _adapter_task = tokio::spawn(async move { + let _ = alkcall::core::types::ProtocolHandler::handle( + &adapter, + producer_conn, + &producer_transport_auth, + ) + .await; + }); + + let consumer_client = ChannelClient::from_connection(consumer_conn) + .await + .expect("consumer channel client init"); + let consumer_client = Arc::new(consumer_client); + let producer_manager = producer_manager_rx + .recv() + .await + .expect("producer manager captured"); + + ForwardTopology { + consumer: consumer_client, + producer_manager, + identity_witness, + } +} + /// An echo dial closure (the ADR-004 injection point under test): an /// in-process pipe per dial — the "target" echoes bytes until EOF. /// The consumer side gets split `DuplexStream` halves; a spawned task diff --git a/tests/producer_direct_op.rs b/tests/producer_direct_op.rs new file mode 100644 index 0000000..ff44d84 --- /dev/null +++ b/tests/producer_direct_op.rs @@ -0,0 +1,558 @@ +//! Integration tests for the direct op (`channels/tunnel/direct`, +//! ADR-007) end to end over the duplex harness: the params → spec → +//! establisher → registration → `TunnelSession::open_direct` chain. +//! +//! Covered: e2e round-trips through `pump_bidi` (tcp + udp — the +//! framed adapter wraps at the dial closure, datagram boundary +//! preserved), the schema-class failure for malformed targets (no +//! phantom session), the `NOT_FOUND` posture on a producer without +//! the direct op registered, the structural scope separation +//! (`tunnel:open` does NOT imply `tunnel:direct` — FORBIDDEN, and +//! vice versa), and the CF-006 per-call opener identity on the +//! direct establisher (witness). + +mod harness; + +use std::sync::Arc; + +use alkcall::protocol::wire::CallError; +use alktunnels::params::{Substrate, SubstrateAddr}; +use alktunnels::{TunnelEstablishError, TunnelSession}; + +use harness::{ + all_substrate_dial, consumer_identity, echo_dial, failing_dial, framed_udp_echo_dial, wire, + wire_direct, wire_direct_forward, wire_direct_with_identity, Topology, +}; + +fn target_hostport(host: &str, port: u16) -> SubstrateAddr { + SubstrateAddr::HostPort { + host: host.to_string(), + port, + } +} + +/// Call the producer's direct op on channel 0 (the raw call surface), +/// returning the wire `CallError` on failure. +async fn call_direct( + topo: &Topology, + input: serde_json::Value, +) -> Result { + let payload = serde_json::json!({ + "operationId": "channels/tunnel/direct", + "input": input, + }); + let response = topo.consumer_call.call_with_payload(payload).await; + response.result +} + +/// Adopt the producer-allocated channel and pump it against in-process +/// halves (the consumer-side data plane, driven manually). +async fn adopt_and_pump( + topo: &Topology, + channel_id: u32, +) -> ( + tokio::io::ReadHalf, + tokio::io::WriteHalf, + tokio::task::JoinHandle<(u64, u64)>, +) { + let (send, recv) = topo + .consumer_manager + .adopt_channel(channel_id, "alk/tunnel", None) + .await + .expect("adopt producer-allocated channel"); + let bidi = alkcall::core::types::BiStream::from_joined(recv, send); + let (channel_end, local_end) = tokio::io::duplex(64 * 1024); + let (c_read, c_write) = tokio::io::split(channel_end); + let pump = tokio::spawn(alkcall::channels::pump::pump_bidi(bidi, c_read, c_write)); + let (l_read, l_write) = tokio::io::split(local_end); + (l_read, l_write, pump) +} + +/// Only channel 0 may remain — a failed open leaves no data channel +/// on either side (the phantom-channel property). +fn no_data_channels(topo: &Topology) -> bool { + let consumer = topo.consumer_manager.channel_ids(); + let producer = topo.producer.manager().channel_ids(); + consumer.iter().all(|&id| id == 0) && producer.iter().all(|&id| id == 0) +} + +fn establishment_reason_of(err: &CallError) -> Option<&str> { + err.details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()) +} + +#[tokio::test] +async fn direct_tcp_round_trips_through_pump_bidi() { + let topo = wire_direct(echo_dial()).await; + + let out = call_direct( + &topo, + serde_json::json!({ + "substrate": "tcp", + "target": {"host": "db.internal", "port": 5432} + }), + ) + .await + .expect("direct open must succeed"); + let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32; + assert!(topo.producer.manager().has_channel(channel_id)); + + let (mut read, mut write, pump) = adopt_and_pump(&topo, channel_id).await; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + write.write_all(b"ping-direct").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 11]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip timed out") + .expect("read"); + assert_eq!(&buf, b"ping-direct"); + + // EOF propagates and the wrapper reaps the producer-side channel. + drop(write); + drop(read); + let (c2p, p2c) = tokio::time::timeout(std::time::Duration::from_secs(5), pump) + .await + .expect("pump completion timed out") + .expect("pump task"); + assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!( + !topo.producer.manager().has_channel(channel_id), + "wrapper reaped the producer-side channel after pump completion" + ); +} + +#[tokio::test] +async fn direct_udp_round_trips_framed() { + // The UDP dial closure wraps the framed adapter (ADR-003) at the + // boundary — the establisher stays substrate-agnostic; the + // session's datagram plane sees boundary-preserved frames. + let topo = wire_direct(all_substrate_dial()).await; + + let out = call_direct( + &topo, + serde_json::json!({ + "substrate": "udp", + "target": {"host": "10.0.0.7", "port": 53} + }), + ) + .await + .expect("udp direct open must succeed"); + let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32; + + let mut session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Udp, + "alk/tunnel", + ) + .await + .expect("adopt into a udp session"); + session + .send_datagram(b"datagram-direct") + .await + .expect("send"); + let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("round trip timed out") + .expect("recv") + .expect("datagram"); + assert_eq!(dg.as_ref(), b"datagram-direct"); + session.close().await; +} + +#[tokio::test] +async fn malformed_target_is_schema_class_and_leaves_no_channel() { + // The registry's input-schema gate runs before the establisher: a + // unix target without `path` (and a tcp target without `port`) + // fails as `INVALID_INPUT` — never a phantom session, never a + // establisher dial. + let topo = wire_direct(echo_dial()).await; + + let err = call_direct( + &topo, + serde_json::json!({ + "substrate": "unix", + "target": {"host": "nope", "port": 1} + }), + ) + .await + .expect_err("unix target without path must fail the schema gate"); + assert_eq!(err.code, "INVALID_INPUT"); + assert!(establishment_reason_of(&err).is_none()); + assert!(no_data_channels(&topo)); + + let err = call_direct( + &topo, + serde_json::json!({ + "substrate": "tcp", + "target": {"host": "nope"} + }), + ) + .await + .expect_err("tcp target without port must fail the schema gate"); + assert_eq!(err.code, "INVALID_INPUT"); + assert!(no_data_channels(&topo)); + + // Missing target entirely. + let err = call_direct(&topo, serde_json::json!({"substrate": "tcp"})) + .await + .expect_err("missing target must fail the schema gate"); + assert_eq!(err.code, "INVALID_INPUT"); + assert!(no_data_channels(&topo)); +} + +#[tokio::test] +async fn dial_failure_passes_through_typed() { + // The dial closure's own typed errors pass through unchanged + // (ADR-007 §4's table minus the registry row). + let topo = wire_direct(failing_dial("no route to target")).await; + + let err = call_direct( + &topo, + serde_json::json!({ + "substrate": "tcp", + "target": {"host": "blackhole.internal", "port": 9} + }), + ) + .await + .expect_err("failed dial must fail the open"); + assert_eq!(err.code, "channel:open_failed"); + assert_eq!(establishment_reason_of(&err), Some("dial_failed")); + assert!(no_data_channels(&topo)); +} + +#[tokio::test] +async fn producer_without_direct_op_rejects_loudly() { + // An old producer (only the base op registered) rejects the + // unknown op loudly — NOT_FOUND, the SSH "unknown channel type" + // posture. + let registry = alktunnels::producer::ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + assert!(topo + .producer_registry + .registration("channels/tunnel/direct") + .is_none()); + + let err = call_direct( + &topo, + serde_json::json!({ + "substrate": "tcp", + "target": {"host": "db.internal", "port": 5432} + }), + ) + .await + .expect_err("unknown op must be rejected loudly"); + assert_eq!(err.code, "NOT_FOUND"); + assert!(no_data_channels(&topo)); +} + +#[tokio::test] +async fn tunnel_open_scope_does_not_imply_tunnel_direct() { + // ADR-006 Amendment 1: the scopes are structurally separate + // grants — an identity holding `tunnel:open` (but not + // `tunnel:direct`) is FORBIDDEN on the direct op. + let topo = wire_direct_with_identity(consumer_identity()).await; + + let err = call_direct( + &topo, + serde_json::json!({ + "substrate": "tcp", + "target": {"host": "db.internal", "port": 5432} + }), + ) + .await + .expect_err("tunnel:open must NOT authorize the direct op"); + assert_eq!(err.code, "FORBIDDEN"); + assert!(no_data_channels(&topo)); +} + +#[tokio::test] +async fn direct_scope_does_not_imply_tunnel_open() { + // The separation is symmetric: `tunnel:direct` does not authorize + // the base open op either (two capabilities; a deployment + // granting both grants both explicitly). A tunnel:direct-only + // identity is FORBIDDEN on `channels/tunnel/sub` even though the + // producer registered it. + let registry = alktunnels::producer::ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = harness::wire_with( + registry, + harness::RegistrationMode::Dial(echo_dial()), + Some(harness::direct_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + let payload = serde_json::json!({ + "operationId": "channels/tunnel/sub", + "input": {"resource": "echo", "substrate": "tcp"}, + }); + let response = topo.consumer_call.call_with_payload(payload).await; + let err = response + .result + .expect_err("tunnel:direct must NOT authorize the base op"); + assert_eq!(err.code, "FORBIDDEN"); + assert!(no_data_channels(&topo)); +} + +#[tokio::test] +async fn direct_establisher_witnesses_per_call_identity() { + // CF-006: the direct establisher's per-call auth carried the END + // CALLER's identity (the direct op is the arbitrary-egress + // capability — identity visibility matters more, not less). + let topo = wire_direct(echo_dial()).await; + + let out = call_direct( + &topo, + serde_json::json!({ + "substrate": "tcp", + "target": {"host": "db.internal", "port": 5432} + }), + ) + .await + .expect("direct open authorized (both-scopes identity)"); + assert!(out["channel_id"].is_u64()); + + let witness = topo.identity_witness.lock().await.clone(); + assert_eq!( + witness.as_deref(), + Some("consumer-both"), + "direct establisher saw the per-call opener identity (CF-006)" + ); +} + +#[tokio::test] +async fn establisher_parse_backstop_maps_handler_error() { + // The establisher's serde parse is the backstop behind the schema + // gate: feed an input the schema would have rejected but a direct + // establisher call never schema-checks (the raw-call shape — + // assembly variants, or a registry whose schema gate is looser) + // and observe the typed `handler_error` mapping (ADR-007 §4). + // An unknown substrate value fails the enum parse. + let establisher = alktunnels::producer::direct_establisher(echo_dial()); + let err = establisher( + serde_json::json!({ + "substrate": "sctp", + "target": {"host": "h", "port": 1} + }), + alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), + ) + .await; + match err { + Err(alkcall::channels::operations::EstablishmentError::HandlerError { message }) => { + assert!(message.contains("invalid direct params"), "{message}"); + } + other => panic!("expected HandlerError backstop, got {other:?}"), + } +} + +#[tokio::test] +async fn direct_establisher_renders_target_to_backing_string() { + // The design pin: the establisher renders the target to the SAME + // backing-string form the registry backings use and calls the + // injected DialFn with it — the dial closure observes the exact + // rendered form ("host:port" / path), per ADR-007 §4. + let observed: Arc>> = + Arc::new(tokio::sync::Mutex::new(Vec::new())); + let obs = Arc::clone(&observed); + let dial: alktunnels::producer::DialFn = + Arc::new(move |substrate: Substrate, backing: &str| { + let obs = Arc::clone(&obs); + let backing = backing.to_string(); + Box::pin(async move { + obs.lock().await.push((substrate, backing)); + Err(TunnelEstablishError::DialFailed("stop here".to_string())) + }) + }); + let establisher = alktunnels::producer::direct_establisher(dial); + let auth = alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"); + + let _ = establisher( + serde_json::json!({"substrate": "tcp", "target": {"host": "db.internal", "port": 5432}}), + auth.clone(), + ) + .await; + let _ = establisher( + serde_json::json!({"substrate": "unix", "target": {"path": "/run/pg/.s.PGSQL.5432"}}), + auth, + ) + .await; + + let seen = observed.lock().await.clone(); + assert_eq!( + seen, + vec![ + (Substrate::Tcp, "db.internal:5432".to_string()), + (Substrate::Unix, "/run/pg/.s.PGSQL.5432".to_string()), + ], + "targets rendered to the registry backing-string form" + ); +} + +// --- consumer session level (TunnelSession::open_direct) --------------- + +#[tokio::test] +async fn open_direct_session_round_trip_and_teardown() { + let topo = wire_direct_forward(echo_dial()).await; + + let mut session = TunnelSession::open_direct( + &topo.consumer, + target_hostport("db.internal", 5432), + Substrate::Tcp, + ) + .await + .expect("open_direct"); + assert!(topo.producer_manager.has_channel(session.channel_id)); + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (read, write) = session.stream_halves().expect("stream session has halves"); + write.write_all(b"ping-direct").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 11]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"ping-direct"); + + // Teardown semantics identical to `open` (ADR-007 §5). + let channel_id = session.channel_id; + let reaped = session.close().await; + assert!(reaped, "close reaped the adopted entry"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!( + !topo.producer_manager.has_channel(channel_id), + "producer side reaped after pump completion" + ); +} + +#[tokio::test] +async fn open_direct_session_datagram_boundary_preserved() { + let topo = wire_direct_forward(framed_udp_echo_dial()).await; + + let mut session = TunnelSession::open_direct( + &topo.consumer, + target_hostport("10.0.0.7", 53), + Substrate::Udp, + ) + .await + .expect("udp open_direct"); + session.send_datagram(b"one").await.expect("send"); + session.send_datagram(b"").await.expect("send empty"); + let first = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("round trip") + .expect("recv") + .expect("datagram"); + assert_eq!(first.as_ref(), b"one"); + let second = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("round trip") + .expect("recv") + .expect("datagram"); + assert!( + second.is_empty(), + "the empty datagram crosses framed (F-2 layering)" + ); + session.close().await; +} + +#[tokio::test] +async fn open_direct_malformed_target_no_phantom_session() { + // The session-level failure surface: a malformed target is the + // schema/invalid_input-class failure (registry-side); a failed + // open never yields a session. + let topo = wire_direct_forward(echo_dial()).await; + + let err = match TunnelSession::open_direct( + &topo.consumer, + SubstrateAddr::Path { + path: "/nope".to_string(), + }, + Substrate::Tcp, + ) + .await + { + Err(e) => e, + Ok(_) => panic!("path-shaped target on tcp must fail"), + }; + let call = err.open_ref().call_error().expect("wire error"); + assert_eq!(call.code, "INVALID_INPUT"); + + // Both managers hold only channel 0 (the phantom-channel property). + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0) + && topo + .producer_manager + .channel_ids() + .iter() + .all(|&id| id == 0), + "no phantom session" + ); +} + +#[tokio::test] +async fn open_direct_scope_separated_session_level() { + // Session-level scope separation (ADR-006 Amendment 1): a + // tunnel:open-only consumer client is FORBIDDEN on open_direct — + // the scopes are separate grants, and the session constructor + // states which capability it invokes. + let topo = harness::wire_direct_forward_with_identity(echo_dial(), consumer_identity()).await; + + let err = match TunnelSession::open_direct( + &topo.consumer, + target_hostport("db.internal", 5432), + Substrate::Tcp, + ) + .await + { + Err(e) => e, + Ok(_) => panic!("tunnel:open-only identity must NOT open_direct"), + }; + let call = err.open_ref().call_error().expect("wire error"); + assert_eq!(call.code, "FORBIDDEN"); + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0) + && topo + .producer_manager + .channel_ids() + .iter() + .all(|&id| id == 0), + "no phantom session" + ); +} + +#[tokio::test] +async fn open_direct_witness_carries_session_identity() { + let topo = wire_direct_forward(echo_dial()).await; + let _session = TunnelSession::open_direct( + &topo.consumer, + target_hostport("db.internal", 5432), + Substrate::Tcp, + ) + .await + .expect("open_direct"); + let witness = topo.identity_witness.lock().await.clone(); + assert_eq!( + witness.as_deref(), + Some("consumer-both"), + "the session path's establisher saw the per-call opener identity" + ); +}