alktunnels/tests/producer_forwarded.rs

//! Integration tests for the forwarded op's SERVING side (the
//! listen-opener half: ADR-008 §3, the `tunnels/forwarded-receipt`
//! shape): `register_tunnel_forwarded_openable`, the
//! `register_openable` shape (no establisher; the data plane starts
//! immediately), scope-gated `tunnel:forwarded`, whose handler parses
//! `TunnelForwardedParams`, hands the correlation material (parsed
//! params + `BiStream`) to the assembly, and awaits the assembly's
//! pump inline (R-02).
//!
//! Covered (the task's test plan): the serving-side e2e (an open with
//! valid params + scope → the assembly handler invoked with the parsed
//! params + the stream; the assembly pumps against local duplex halves
//! → byte round-trip), the scope gate (identity without
//! `tunnel:forwarded` → FORBIDDEN, never implied by
//! `tunnel:open`/`tunnel:direct`), malformed params (unknown field,
//! `listen: 0`, missing `peer` → loud rejection; the serde-gated shape
//! tears the channel down at birth (no phantom), and the
//! correlation-miss teardown (the assembly's rejection: drop the
//! stream, the channel reaps).
//!
//! The reply reader's unit tests land in `src/params.rs` (this task's
//! projection logic, unit-testable independent of the session type);
//! the full bind-first flow e2e lives in `tunnels/bindfirst-producer`.
//!
//! Topology (the harness's producer-leg stub, mirrored): the producer
//! leg is the connect side (it holds the `ChannelClient` whose
//! `call_open_op` is the per-accept call plane (the reverse POC's
//! `open_reverse_channel` shape, reduced); the opener leg is the
//! accept side whose install-hook serving registry carries the
//! forwarded op under test. The test plays the producer: per probe it
//! CALLS `channels/tunnel/forwarded` on the opener's channel-0 call
//! plane, adopts the opener-allocated `channel_id` on the
//! producer-side manager, and pumps the adopted channel against a
//! driving duplex (the dispatcher-pump dance, test-driven). The
//! opener-side assembly correlates by the `listen` token and pumps the
//! forwarded channel against the local "BIND client" halves; bytes
//! the test writes on the client end traverse the whole path and echo
//! back.

mod harness;

use std::sync::Arc;

use alkcall::channels::manager::ChannelManager;
use alkcall::core::auth::Identity;
use alkcall::protocol::wire::CallError;
use alktunnels::params::{TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE};
use alktunnels::producer::{register_tunnel_forwarded_openable, TargetHandle};

use harness::{wire_with, RegistrationMode};

/// The opener-side identity holding ONLY `tunnel:forwarded` (the
/// ingress grant, pinned separate from `tunnel:open`/`tunnel:direct`,
/// never implied by either; this identity holds neither).
fn forwarded_only_identity() -> Identity {
    Identity {
        id: "opener-forwarded-only".to_string(),
        scopes: vec![TUNNEL_FORWARDED_SCOPE.to_string()],
        resources: std::collections::HashMap::new(),
    }
}

/// A plain dial-egress identity (`tunnel:open` only, the
/// never-implied probe's identity).
fn open_only_identity() -> Identity {
    Identity {
        id: "opener-open-only".to_string(),
        scopes: vec![TUNNEL_OPEN_SCOPE.to_string()],
        resources: std::collections::HashMap::new(),
    }
}

/// What the assembly's test handler received per forwarded open (the
/// invocation probe): the parsed params as the wrapper handed them
/// over.
#[derive(Default)]
struct Received {
    params: Option<alktunnels::params::TunnelForwardedParams>,
}

type ReceivedWitness = Arc<tokio::sync::Mutex<Received>>;

/// Serve the forwarded op on the opener side with the injected
/// assembly handler: record the parsed params, pop the correlated
/// halves for the token, and `pump_bidi` the channel `BiStream`
/// against them (convention 8: the assembly owns the pump; the
/// protocol-side built-in handler only bridges wrapper→assembly). A
/// correlation miss (no halves for the token) is the assembly's
/// rejection: drop the stream, return; teardown-at-birth reaps the
/// forwarded channel.
fn serve_forwarded_on(
    consumer_registry: &Arc<alkcall::registry::registration::OperationRegistry>,
    manager: ChannelManager,
    received: ReceivedWitness,
    local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>>,
) {
    let handler: alktunnels::producer::ForwardedHandler = Arc::new(move |forwarded, bidi| {
        let received = Arc::clone(&received);
        let local_halves = Arc::clone(&local_halves);
        tokio::spawn(async move {
            *received.lock().await = Received {
                params: Some(forwarded),
            };
            let target = local_halves.lock().await.pop();
            if let Some(target) = target {
                alkcall::channels::pump::pump_bidi(bidi, target.read, target.write).await;
            }
        })
    });
    let core = alkcall::channels::operations::ChannelCore::new(
        manager,
        alkcall::channels::policy::default_policy(),
    );
    register_tunnel_forwarded_openable(
        &core,
        consumer_registry,
        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
        handler,
    )
    .expect("register forwarded openable");
}

/// A topology for the forwarded-op serving probes: the producer leg
/// (connect side) serves only the unused base op (a failing dial, the
/// base op is not what's under test); the opener leg is the accept
/// side whose serving registry carries the forwarded op. The test
/// plays the producer: `topo.producer.call_open_op(OP_TUNNEL_FORWARDED,
/// …)` is the per-accept call; the opener-allocated `channel_id` in
/// the reply is what the producer adopts.
async fn wire_forwarded_serving(identity: Identity) -> harness::Topology {
    wire_with(
        alktunnels::producer::ResourceRegistry::new(),
        RegistrationMode::Dial(harness::failing_dial("base op unused in forwarded probes")),
        Some(identity),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await
}

fn forwarded_input(listen: u32, peer: (u32, u16)) -> serde_json::Value {
    serde_json::json!({
        "listen": listen,
        "substrate": "tcp",
        "peer": {"host": peer.0.to_string(), "port": peer.1},
    })
}

/// Call the forwarded op on the producer's per-accept call plane
/// (`call_open_op`, the producer-side initiator's surface; the
/// opener's wrapper allocates the channel ID, the producer adopts).
async fn call_forwarded(
    producer: &alkcall::channels::client::ChannelClient,
    input: serde_json::Value,
) -> Result<serde_json::Value, CallError> {
    producer
        .call_open_op(alktunnels::params::OP_TUNNEL_FORWARDED, input)
        .await
        .result
}

/// The accepted-side echo pair (the producer-side pump target): a
/// duplex pair; the driving end an echo task (the app server's
/// socket stand-in), the far end the pump's halves.
fn driving_halves() -> TargetHandle {
    let (drive_end, target_end) = tokio::io::duplex(64 * 1024);
    let (mut t_read, mut t_write) = tokio::io::split(target_end);
    tokio::spawn(async move {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let mut buf = vec![0u8; 8192];
        loop {
            match t_read.read(&mut buf).await {
                Ok(0) | Err(_) => return,
                Ok(n) => {
                    if t_write.write_all(&buf[..n]).await.is_err() {
                        return;
                    }
                }
            }
        }
    });
    let (d_read, d_write) = tokio::io::split(drive_end);
    TargetHandle {
        read: Box::new(d_read),
        write: Box::new(d_write),
    }
}

/// The "local BIND client" pair (the opener-side pump target): a raw
/// duplex pair; the map's halves are the assembly pump's target, the
/// far end is the test's client end (driven directly).
fn bind_client_pair() -> (
    tokio::io::ReadHalf<tokio::io::DuplexStream>,
    tokio::io::WriteHalf<tokio::io::DuplexStream>,
    TargetHandle,
) {
    let (a, b) = tokio::io::duplex(64 * 1024);
    let (a_read, a_write) = tokio::io::split(a);
    let (b_read, b_write) = tokio::io::split(b);
    (
        b_read,
        b_write,
        TargetHandle {
            read: Box::new(a_read),
            write: Box::new(a_write),
        },
    )
}

/// THE serving-half e2e (this task's gate): an open with valid params
/// + scope → the assembly handler invoked with the parsed params +
/// the stream → the assembly pumps against local duplex halves → byte
/// round-trip through the forwarded channel. (The full bind-first
/// flow e2e lives in `tunnels/bindfirst-producer`; this task's e2e is
/// the serving half standalone.)
#[tokio::test]
async fn forwarded_serving_round_trip_end_to_end() {
    let topo = wire_forwarded_serving(forwarded_only_identity()).await;
    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
        Arc::new(tokio::sync::Mutex::new(Vec::new()));
    serve_forwarded_on(
        &topo.consumer_registry,
        topo.consumer_manager.clone(),
        Arc::clone(&received),
        Arc::clone(&local_halves),
    );

    // Correlate FIRST (the waiting listener session's bookkeeping:
    // the assembly pre-registered the local halves for the token
    // before the forwarded open arrives; a late push is the
    // correlation-miss shape, probed separately below).
    const TOKEN: u32 = 918273;
    let (mut b_read, mut b_write, local) = bind_client_pair();
    local_halves.lock().await.push(local);

    // The producer calls the forwarded op (valid correlation token +
    // substrate-shaped peer).
    let out = call_forwarded(&topo.producer, forwarded_input(TOKEN, (198, 55231)))
        .await
        .expect("forwarded open");
    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
    assert!(channel_id > 0);

    // The wrapper allocated the channel on the opener's manager (the
    // serving side allocates; the producer adopts) and the assembly
    // pump is live against the correlated halves.
    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
    assert!(
        topo.consumer_manager.has_channel(channel_id),
        "the forwarded channel is live on the opener side"
    );

    // The assembly's test handler saw the parsed params verbatim.
    let got = received.lock().await.params.clone().expect("invoked");
    assert_eq!(got.listen, TOKEN, "the correlation token rides `listen`");
    assert_eq!(got.substrate, alktunnels::params::Substrate::Tcp);
    assert_eq!(
        got.peer,
        alktunnels::params::SubstrateAddr::HostPort {
            host: "198".to_string(),
            port: 55231,
        }
    );

    // The producer-side adopted pump (the test drives the accepted
    // side's echo): adopt the opener-allocated ID on the
    // producer-side manager and pump the adopted channel against a
    // driving duplex (the dispatcher-pump dance, test-driven).
    let (send, recv) = topo
        .producer
        .manager()
        .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("producer adopts the opener-allocated ID");
    let driving = driving_halves();
    let _producer_pump = tokio::spawn(alkcall::channels::pump::pump_bidi(
        alkcall::core::types::BiStream::from_joined(recv, send),
        driving.read,
        driving.write,
    ));

    // THE round-trip: bytes the test writes on the local BIND client's
    // end cross the assembly pump → the forwarded channel → the
    // producer-side adopted pump → the driving echo → and back (the
    // forwarded channel's data path, ADR-008 §3's serving half).
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    b_write.write_all(b"forwarded-ping").await.expect("write");
    b_write.flush().await.expect("flush");
    let mut buf = vec![0u8; 14];
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        b_read.read_exact(&mut buf),
    )
    .await
    .expect("round trip")
    .expect("read");
    assert_eq!(&buf, b"forwarded-ping");
}

/// The scope gate: an identity WITHOUT `tunnel:forwarded` is FORBIDDEN,
/// never implied by `tunnel:open` (this identity holds exactly that
/// grant; ADR-006 Amendment 1's separation).
#[tokio::test]
async fn forwarded_scope_gate_is_independent_grant() {
    let topo = wire_forwarded_serving(open_only_identity()).await;
    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
        Arc::new(tokio::sync::Mutex::new(Vec::new()));
    serve_forwarded_on(
        &topo.consumer_registry,
        topo.consumer_manager.clone(),
        Arc::clone(&received),
        local_halves,
    );

    match call_forwarded(&topo.producer, forwarded_input(7, (10, 1))).await {
        Ok(_) => panic!("a scope-less opener must not receive forwarded opens"),
        Err(e) => {
            assert_eq!(
                e.code, "FORBIDDEN",
                "missing `tunnel:forwarded` → FORBIDDEN"
            );
            assert!(
                e.message.contains("tunnel:forwarded"),
                "the denial names the missing scope: {}",
                e.message
            );
        }
    }
    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
    // No channel was ever allocated (the ACL runs before the wrapper).
    assert!(
        topo.consumer_manager
            .channel_ids()
            .iter()
            .all(|&id| id == 0),
        "a FORBIDDEN open leaves no channel on the opener side"
    );
    // The assembly handler never ran (no invocation witnessed).
    assert!(received.lock().await.params.is_none());
}

/// Malformed params: `listen: 0` and a missing `peer` fail the
/// registry's schema gate LOUD (`INVALID_INPUT`; before the wrapper,
/// no channel ever); an unknown top-level field passes the schema (the
/// additive posture) but fails the serde parse (the loud backstop) →
/// the wrapper logs + returns early → the channel tears down at birth
/// (no phantom channel).
#[tokio::test]
async fn forwarded_malformed_params_tear_down_loud() {
    let topo = wire_forwarded_serving(forwarded_only_identity()).await;
    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
        Arc::new(tokio::sync::Mutex::new(Vec::new()));
    serve_forwarded_on(
        &topo.consumer_registry,
        topo.consumer_manager.clone(),
        Arc::clone(&received),
        local_halves,
    );

    // `listen: 0`: the schema gate (minimum: 1) fires first.
    match call_forwarded(
        &topo.producer,
        serde_json::json!({
            "listen": 0,
            "substrate": "tcp",
            "peer": {"host": "10", "port": 53},
        }),
    )
    .await
    {
        Ok(_) => panic!("`listen: 0` must fail the open"),
        Err(e) => {
            assert_eq!(e.code, "INVALID_INPUT", "schema minimum 1 gate");
        }
    }

    // Missing `peer`, the schema gate again (all three required).
    match call_forwarded(
        &topo.producer,
        serde_json::json!({"listen": 7, "substrate": "tcp"}),
    )
    .await
    {
        Ok(_) => panic!("missing `peer` must fail the open"),
        Err(e) => {
            assert_eq!(e.code, "INVALID_INPUT");
        }
    }

    // Unknown top-level field: passes the schema (additive posture),
    // fails the serde parse (deny-unknown backstop) → the wrapper
    // returns early → teardown-at-birth (the channel was allocated;
    // the reply IS the channel_id, but reaps promptly).
    let out = call_forwarded(
        &topo.producer,
        serde_json::json!({
            "listen": 918273,
            "substrate": "tcp",
            "peer": {"host": "198", "port": 55232},
            "sneaky": true
        }),
    )
    .await
    .expect("the malformed open rides the no-establisher reply shape");
    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    assert!(
        !topo.consumer_manager.has_channel(channel_id),
        "a malformed-params open tears the channel down at birth (no phantom)"
    );
    // The assembly handler never ran (the parse gate is protocol-side).
    assert!(received.lock().await.params.is_none());
}

/// The correlation-miss teardown (the assembly's rejection; it holds
/// the map): a valid-shaped open whose `listen` token matches no
/// waiting session → the assembly's error path drops the stream and
/// returns → teardown-at-birth reaps the forwarded channel. The
/// producer's accept loop drops the accepted connection on the open's
/// failure (the shapes mirror; pinned producer-side in
/// `tunnels/bindfirst-producer`).
#[tokio::test]
async fn forwarded_correlation_miss_tears_channel_down() {
    let topo = wire_forwarded_serving(forwarded_only_identity()).await;
    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
    // NO correlated halves for the token (the map is empty).
    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
        Arc::new(tokio::sync::Mutex::new(Vec::new()));
    serve_forwarded_on(
        &topo.consumer_registry,
        topo.consumer_manager.clone(),
        Arc::clone(&received),
        local_halves,
    );

    let out = call_forwarded(&topo.producer, forwarded_input(42113, (198, 55233)))
        .await
        .expect("the open itself resolves (allocation is wrapper-side)");
    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    assert!(
        !topo.consumer_manager.has_channel(channel_id),
        "a correlation miss tears the forwarded channel down (teardown-at-birth)"
    );
    // The assembly handler RAN (the invocation was witnessed; the
    // rejection is the assembly's) and its map missed.
    let witnessed = received.lock().await.params.clone();
    assert_eq!(witnessed.expect("invoked").listen, 42113);
}