From ff81f2d27b1f43756220da4fbfeb980680fc8b33 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Thu, 24 Sep 2026 18:04:55 +0000 Subject: [PATCH] producer: the explicit bind-policy gate + the builder-register pin (review-002 C-3, D-2, S-1, N-2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - C-3: Socks5ProducerBuilder.bind_policy(bool) — the substrate-free entry's explicit BIND gate, defaulting to the accept seam's presence; seam presence is never the sole gate - D-2: RegisterMode::ProducerBuilder + an integration test wiring the full builder through .register(...) serving all three commands - S-1: run_datagram_stage's doc names the hot-Ok(None) residual - N-2: wire.md §UDP ASSOCIATE carries the inline-send backpressure line Verified: cargo test (352 green), clippy/fmt/doc clean, wasm checks clean. --- docs/architecture/wire.md | 4 + src/producer.rs | 240 +++++++++++++++++- .../fix-producer-gate-policy.md | 78 +++++- tests/core-integration.rs | 46 ++++ tests/harness.rs | 42 +++ 5 files changed, 393 insertions(+), 17 deletions(-) diff --git a/docs/architecture/wire.md b/docs/architecture/wire.md index be1eba7..9cec204 100644 --- a/docs/architecture/wire.md +++ b/docs/architecture/wire.md @@ -165,6 +165,10 @@ the same channel: sender's address per the RFC's per-datagram source semantics) or composed (a lazy destination→direct-op-tunnel table, LRU-evicted; open cost ~0.8 ms measured, per-target ACL for free). + **Per-datagram `send` is awaited inline** (the stage interleaves + sends with reply drains); a slow egress applies backpressure to + the association, not to other channels — the batch-2 + composed-egress author inherits this bound knowingly. ### BIND: the phase model extended (ADR-007) diff --git a/src/producer.rs b/src/producer.rs index 646a194..a1a9960 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -80,7 +80,10 @@ //! boolean). The channel `BiStream` and a `TcpStream` are equally //! `T` — this is what makes the producer testable without channels //! (the differential-suite pattern) and reusable by the batch-2 -//! local backend (feature `local`, ADR-009). +//! local backend (feature `local`, ADR-009). The builder's shape +//! resolves that gate from its explicit `bind_policy` (defaulting to +//! the accept seam's presence — never the sole gate; ADR-006's +//! "listener configured" ≠ "listener permitted"). //! //! ## Registration //! @@ -346,7 +349,19 @@ where /// flow table); one relay loop, no spawn-per-datagram. A frame's /// egress `send` is awaited inline before the next select iteration — /// the stage interleaves egress sends with reply drains (the POC's -/// measured shape; sufficient for relay postures). +/// measured shape; sufficient for relay postures; a slow egress +/// applies backpressure to the association, never to other channels — +/// wire.md §UDP ASSOCIATE). +/// +/// Residual posture (review 002 S-1, named decision): the `Ok(None)` +/// tolerance's defense is bounded by `yield_now` only — a *misbehaving* +/// egress returning `Ok(None)` at full speed degrades the stage into +/// a yielding busy-wait (making no progress, unbounded by rate; the +/// runtime survives, the stage never parks). A compliant egress parks +/// (the readiness contract); a backoff/budget in the stage would +/// penalize the compliant path, so the residual is bounded by the +/// egress's own correctness and accepted-uncovered — named here, not +/// accidental. async fn run_datagram_stage(mut inner: T, egress: &mut (dyn Egress + Send)) -> StageExit where T: AsyncRead + AsyncWrite + Unpin, @@ -615,11 +630,19 @@ pub fn register_socks5_openable( /// closes over substrate code behind the callbacks; the protocol /// layer never sees a socket type (ADR-004/ADR-008). Missing /// callbacks = those commands refuse RFC-faithfully (ADR-008). +/// +/// The BIND command gate on the substrate-free entry +/// ([`Socks5Producer::serve_socks5_over`]) is the builder's +/// `bind_policy` — explicit via `.bind_policy(bool)`, defaulting to +/// the accept seam's presence for backward-compat. The channels +/// path's gate is the pump handler's, resolved from the per-call +/// opener identity (ADR-006) — this policy never touches it. #[derive(Clone, Default)] pub struct Socks5Producer { dial: Option, egress: Option, accept: Option, + bind_policy: bool, } impl Socks5Producer { @@ -629,10 +652,12 @@ impl Socks5Producer { /// The substrate-free protocol entry (the same shape /// [`serve_socks5_over`] pins) over the builder's seams. The - /// command gate is resolved from the builder's `accept` presence: - /// a producer that wired the accept seam serves BIND to whatever - /// calls this entry directly (the per-caller gate is the pump - /// handler's, resolved from the open-op identity — ADR-006). + /// command gate is the builder's `bind_policy` — an explicit + /// `.bind_policy(bool)` whenever the local policy must speak + /// ("factory wired but gate closed" is expressible here); + /// unwired, it defaults to the accept seam's presence for + /// backward-compat. The per-caller gate is the pump handler's, + /// resolved from the open-op identity — ADR-006. pub async fn serve_socks5_over(&self, stream: T) -> Result<(), SocksError> where T: AsyncRead + AsyncWrite + Unpin, @@ -642,7 +667,7 @@ impl Socks5Producer { self.dial.clone(), self.egress.clone(), self.accept.clone(), - self.accept.is_some(), + self.bind_policy, ) .await } @@ -675,6 +700,7 @@ pub struct Socks5ProducerBuilder { dial: Option, egress: Option, accept: Option, + bind_policy: Option, } impl Socks5ProducerBuilder { @@ -696,8 +722,21 @@ impl Socks5ProducerBuilder { self } + /// The substrate-free entry's BIND command gate (ADR-006's + /// posture — "listener configured" ≠ "listener permitted"): the + /// resolved `can_bind` for [`Socks5Producer::serve_socks5_over`]. + /// Set it when the local policy must close the gate while the + /// accept seam stays wired. Unwired, the gate defaults to the + /// accept seam's presence (the backward-compat reading — the + /// pump handler's identity-derived gate is unaffected). + pub fn bind_policy(mut self, can_bind: bool) -> Self { + self.bind_policy = Some(can_bind); + self + } + pub fn build(self) -> Socks5Producer { Socks5Producer { + bind_policy: self.bind_policy.unwrap_or(self.accept.is_some()), dial: self.dial, egress: self.egress, accept: self.accept, @@ -2249,6 +2288,193 @@ mod tests { result.expect("clean"); } + /// The gate policy (review 002 C-3): the builder's `bind_policy` + /// is the substrate-free entry's BIND gate — "factory wired but + /// gate closed" is expressible (`bind_policy(false)` with the + /// accept seam still wired refuses BIND 0x02, no factory call — + /// the gate precedes the seam) while CONNECT and ASSOCIATE still + /// serve the same producer. Seam presence is never the sole gate. + #[tokio::test] + async fn builder_bind_policy_false_refuses_bind_while_connect_associate_serve() { + let requested = Arc::new(tokio::sync::Mutex::new(None)); + let producer = Socks5Producer::builder() + .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))) + .egress(echo_egress_factory()) + .accept(bind_accept_factory(Arc::clone(&requested))) + .bind_policy(false) + .build(); + + // BIND with the factory wired and the gate closed: 0x02, and + // the accept seam is never called (no phantom listener — the + // gate precedes the seam, the ADR-006 shape). + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await }); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPBind, + &TargetAddr::Domain("policy.closed".into(), 2121), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!( + code, 0x02, + "the explicit policy refuses ConnectionNotAllowed" + ); + assert!( + requested.lock().await.is_none(), + "the closed gate precedes the accept seam — no phantom listener" + ); + drop(client); + let err = tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("session ends") + .expect("join") + .expect_err("the refusal surfaces in-stream"); + assert!(matches!( + err, + SocksError::ServerError(SocksServerError::AuthenticationRejected) + )); + + // CONNECT still serves the same producer (egress unaffected). + let producer = Socks5Producer::builder() + .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))) + .egress(echo_egress_factory()) + .accept(bind_accept_factory(Arc::new(tokio::sync::Mutex::new(None)))) + .bind_policy(false) + .build(); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await }); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("policy.connect".into(), 443), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00, "CONNECT is unaffected by the bind policy"); + client.write_all(b"ping").await.unwrap(); + let mut buf = [0u8; 4]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"ping"); + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("session ends") + .expect("join"); + result.expect("clean"); + + // ASSOCIATE still serves the same producer. + let producer = Socks5Producer::builder() + .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))) + .egress(echo_egress_factory()) + .accept(bind_accept_factory(Arc::new(tokio::sync::Mutex::new(None)))) + .bind_policy(false) + .build(); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await }); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::UDPAssociate, + &TargetAddr::Domain("policy.udp".into(), 1080), + ) + .await; + let (code, addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00, "ASSOCIATE is unaffected by the bind policy"); + assert_eq!( + (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]), + (127, 0, 0, 1, 0, 0), + "the reply carries the sentinel BND.ADDR" + ); + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 8], 53))); + client + .write_all(&datagram_frame(&target, b"policy")) + .await + .unwrap(); + let mut buf = vec![0u8; 2 + 7 + 6]; + client.read_exact(&mut buf).await.expect("echo"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"policy"); + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("session ends") + .expect("join"); + result.expect("clean"); + } + + /// The backward-compat default (review 002 C-3): an unwired + /// `bind_policy` resolves the gate from the accept seam's + /// presence — a factory-wired producer serves BIND on the + /// substrate-free entry (the landed tests' reading preserved), + /// while a producer without the factory still refuses. The + /// explicit policy always wins when wired. + #[tokio::test] + async fn builder_default_bind_policy_follows_accept_presence() { + // Accept wired, policy unwired → BIND serves (the default). + let requested = Arc::new(tokio::sync::Mutex::new(None)); + let producer = Socks5Producer::builder() + .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))) + .accept(bind_accept_factory(Arc::clone(&requested))) + .build(); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await }); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPBind, + &TargetAddr::Domain("default.open".into(), 2121), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00, "the presence default serves BIND"); + assert!( + requested.lock().await.is_some(), + "the accept seam ran (the factory is wired and the gate open)" + ); + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("session ends") + .expect("join"); + result.expect("clean"); + + // Accept unwired, policy unwired → BIND refuses (the gate's + // no-grant answer — no seam to run even if the gate were + // open). + let producer = Socks5Producer::builder() + .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))) + .build(); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await }); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPBind, + &TargetAddr::Domain("default.closed".into(), 2121), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x02, "no factory + no policy = the closed default"); + drop(client); + let err = tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("session ends") + .expect("join") + .expect_err("the refusal surfaces in-stream"); + assert!(matches!( + err, + SocksError::ServerError(SocksServerError::AuthenticationRejected) + )); + } + /// The sentinel reply address (the in-band BND.ADDR posture, /// ADR-003): `127.0.0.1:0` — the RFC-compatibility surface. #[test] diff --git a/tasks/implementation/fix-producer-gate-policy.md b/tasks/implementation/fix-producer-gate-policy.md index a359f02..857ae6e 100644 --- a/tasks/implementation/fix-producer-gate-policy.md +++ b/tasks/implementation/fix-producer-gate-policy.md @@ -1,7 +1,7 @@ --- id: implementation/fix-producer-gate-policy name: "Fix — producer BIND gate policy + register coverage + stage doc lines (review-002 C-3, D-2, S-1, N-2)" -status: pending +status: completed depends_on: [] scope: narrow risk: medium @@ -66,21 +66,21 @@ ride the same files ## Acceptance Criteria -- [ ] The substrate-free entry's BIND gate is an explicit policy +- [x] The substrate-free entry's BIND gate is an explicit policy input (with-gate entry or builder field); `accept.is_some()` remains only as the backward-compat default, never the sole gate -- [ ] Test: accept factory wired + explicit `can_bind=false` ⇒ BIND +- [x] Test: accept factory wired + explicit `can_bind=false` ⇒ BIND refused 0x02; CONNECT/ASSOCIATE still serve -- [ ] The pump handler's identity-derived gate is unchanged (the +- [x] The pump handler's identity-derived gate is unchanged (the existing identity matrix stays green) -- [ ] Integration test: full builder (`dial` + `egress` + `accept`) +- [x] Integration test: full builder (`dial` + `egress` + `accept`) through `.register(...)` serving all three commands over the harness topology -- [ ] `run_datagram_stage`'s doc names the hot-`Ok(None)` residual -- [ ] wire.md §UDP ASSOCIATE carries the inline-send backpressure +- [x] `run_datagram_stage`'s doc names the hot-`Ok(None)` residual +- [x] wire.md §UDP ASSOCIATE carries the inline-send backpressure line -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References @@ -94,8 +94,66 @@ ride the same files ## Notes -> Agent fills this during implementation. +- **C-3 shape chosen: the builder field** (`.bind_policy(bool)`), + not a separate `serve_socks5_over_with_gate` entry — the review's + sketch allowed either; a builder field keeps the entry's signature + untouched (one `serve_socks5_over` shape), rides the existing + builder chain, and the policy is resolved once at `build()` + (`self.bind_policy.unwrap_or(self.accept.is_some())` — + `Option` on the builder, `bool` on the built producer). The + free function `serve_socks5_over` (already gate-explicit) and the + pump handler (identity-derived, ADR-006) are untouched. +- **D-2 vehicle: a new `RegisterMode::ProducerBuilder`** in + `tests/harness.rs` wiring the full builder chain + (`.dial().egress().accept()` + the optional `.bind_policy()`) + through `Socks5Producer::register` — the harness's registration + block gains a third arm (same `ChannelCore` + + `register_openable_with_establisher` shape as the others, via the + delegated method). The integration test + (`producer_builder_registers_and_serves_all_three_commands` in + `tests/core-integration.rs`) serves CONNECT + ASSOCIATE + BIND + over the topology and asserts the identity witness (proving the + builder-delegated establisher ran). C-3's gate tests live in the + producer unit suite (`builder_bind_policy_false_refuses_bind_ + while_connect_associate_serve`, `builder_default_bind_policy_ + follows_accept_presence`) — the substrate-free entry is a unit + shape; the builder-through-register surface is the integration + one. +- **The `bind_policy(false)` test** pins: 0x02 on BIND with the + accept factory wired, the factory never called (no phantom + listener), the in-stream `AuthenticationRejected` session error, + and CONNECT + ASSOCIATE still serving the same producer. + **The default test** pins the backward-compat reading: accept + wired + policy unwired ⇒ BIND serves (factory ran); accept + unwired + policy unwired ⇒ 0x02. +- **S-1/N-2 are doc lines only** (per the review's recommendation — + posture notes, not backoff/budget code): a residual paragraph on + `run_datagram_stage` (hot-`Ok(None)` = a yielding busy-wait, + bounded by the egress's own correctness, accepted-uncovered by + design) and a bolded inline-send sentence on the wire.md §UDP + ASSOCIATE producer-egress bullet. ## Summary -> Agent fills this on completion. \ No newline at end of file +All four items landed (C-3, D-2, S-1, N-2). `Socks5ProducerBuilder` +gained `.bind_policy(bool)` — the substrate-free entry's explicit +BIND gate, defaulting to the accept seam's presence (the built +producer resolves it to a plain `bool`); `Socks5Producer:: +serve_socks5_over` now consults the policy instead of +`accept.is_some()`. The pump handler's identity-derived gate is +byte-identical (the identity matrix stayed green untouched). Two +unit tests pin the policy (explicit-closed refuses 0x02 with the +factory un-called while CONNECT/ASSOCIATE serve; the +presence-default both ways). A new harness mode +(`RegisterMode::ProducerBuilder`) plus one integration test wires +the full builder through `.register(...)` and serves all three +commands end to end — the spec'd assembly shape is no longer +unexercised. Doc lines: `run_datagram_stage` names the hot-`Ok(None)` +residual (S-1); wire.md §UDP ASSOCIATE carries the inline-send +backpressure bound (N-2). Verification: `cargo test` 352 green (125 +lib + 38 associate + 41 bind + 40 connect + 41 integration + 30 +harness + 37 differential + 3 doctests, 2 ignored — the two new +unit tests and the one new integration test added to the review's +348 baseline), `cargo clippy --all-targets -- -D warnings` clean, +`cargo fmt --check` clean, `cargo doc --no-deps` clean, both wasm +checks (`cargo check` + `clippy -D warnings`, wasm32) clean. \ No newline at end of file diff --git a/tests/core-integration.rs b/tests/core-integration.rs index f068fde..b5e56d9 100644 --- a/tests/core-integration.rs +++ b/tests/core-integration.rs @@ -1011,3 +1011,49 @@ async fn readiness_contract_holds_under_concurrent_load() { .await .expect("fresh associate after the hot-egress load"); } + +/// The spec'd assembly shape end to end (review 002 D-2): the full +/// `Socks5Producer::builder()` chain — `dial` + `egress` + `accept` — +/// through `.register(...)` (the builder's delegated registration, not +/// the free function), serving all three commands over the harness +/// topology. The one assembly path producer.md §Wiring documents is +/// the one this test pins: a signature drift or a delegation bug in +/// `register` can no longer pass the suite. +#[tokio::test] +async fn producer_builder_registers_and_serves_all_three_commands() { + let topo = wire_with( + RegisterMode::ProducerBuilder { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: Some(echo_accept()), + bind_policy: None, + }, + Some(both_scopes_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + // CONNECT over the builder's dial seam. + connect_round_trip(&topo, "builder.test".into(), "tcp".into()) + .await + .expect("CONNECT serves through the builder registration"); + + // ASSOCIATE over the builder's egress seam. + associate_round_trip(&topo, 60, b"udp") + .await + .expect("ASSOCIATE serves through the builder registration"); + + // BIND over the builder's accept seam (reply#1 + reply#2 + + // pass-through — the full phase model). + bind_round_trip(&topo, "builder.test", b"bind") + .await + .expect("BIND serves through the builder registration"); + + // The registration is the spec'd one: the establisher rode it + // (the witness is the builder-delegated registration's probe). + assert_eq!( + topo.identity_witness.lock().await.as_deref(), + Some("consumer-both") + ); +} diff --git a/tests/harness.rs b/tests/harness.rs index 22d54b0..e8d4288 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -442,6 +442,17 @@ pub enum RegisterMode { egress: Option, accept: Option, }, + /// The spec'd assembly shape (producer.md §Wiring): the + /// registration rides [`alksocks::producer::Socks5Producer:: + /// builder`] + [`alksocks::producer::Socks5Producer::register`] + /// (the builder's `dial`/`egress`/`accept` setters and the + /// delegated registration — the surface D-2 guards). + ProducerBuilder { + dial: DialFn, + egress: Option, + accept: Option, + bind_policy: Option, + }, /// No socks5 openable — only the generic channel ops (the /// unknown-op / registry-gate probe's shape). NoSocks5, @@ -660,6 +671,37 @@ pub async fn wire_with( ) .expect("register socks5 openable (the real producer)"); } + if let RegisterMode::ProducerBuilder { + dial, + egress, + accept, + bind_policy, + } = &mode + { + let core = ChannelCore::new( + producer_client.manager().clone(), + alkcall::channels::policy::default_policy(), + ); + let mut producer = alksocks::producer::Socks5Producer::builder().dial(Arc::clone(dial)); + if let Some(egress) = egress { + producer = producer.egress(egress.clone()); + } + if let Some(accept) = accept { + producer = producer.accept(accept.clone()); + } + if let Some(can_bind) = bind_policy { + producer = producer.bind_policy(*can_bind); + } + producer + .build() + .register( + &core, + &producer_op_registry, + AuthContext::anonymous(b"alk/socks5"), + Some(Arc::clone(&identity_witness)), + ) + .expect("register socks5 openable (the spec'd builder shape)"); + } // --- consumer pieces (captured from the hook) ----------------------- let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn");