car-server-core 0.55.0

//! Supervised-agent method scopes are enforced by the real WS dispatcher.
//!
//! This uses an in-process loopback listener on port 0 and isolated temporary
//! state. It never starts or talks to an installed `car-server`.

use car_memgine::MemgineEngine;
use car_registry::{
    declarative::DeclRegistry,
    supervisor::{AgentSpec, RestartPolicy, Supervisor},
};
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use futures::{SinkExt, StreamExt};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tokio_tungstenite::{accept_async, connect_async, tungstenite::Message};

type Ws =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

const HOST_TOKEN: &str = "method-scope-host-token-0123456789abcdef";
const SCOPED_TOKEN: &str = "method-scope-agent-token-0123456789abcdef";
const PIPELINED_TOKEN: &str = "pipelined-agent-token-0123456789abcdef";
const UNSCOPED_TOKEN: &str = "unrestricted-agent-token-0123456789abcdef";
const EMPTY_TOKEN: &str = "empty-scope-agent-token-0123456789abcdef";

fn agent_spec(id: &str, token: &str, method_allowlist: Option<Vec<String>>) -> AgentSpec {
    #[cfg(windows)]
    let (command, args) = (
        std::env::var("COMSPEC").unwrap_or_else(|_| r"C:\Windows\System32\cmd.exe".to_string()),
        vec!["/C".to_string(), "exit 0".to_string()],
    );
    #[cfg(unix)]
    let (command, args) = (
        "/bin/sh".to_string(),
        vec!["-c".to_string(), "exit 0".to_string()],
    );
    AgentSpec {
        id: id.to_string(),
        name: id.to_string(),
        command,
        args,
        cwd: None,
        env: Default::default(),
        restart: RestartPolicy::Never,
        max_restarts: 1,
        backoff_secs: 1,
        auto_start: false,
        token: token.to_string(),
        method_allowlist,
        capabilities: Vec::new(),
    }
}

async fn state(root: &std::path::Path) -> Arc<ServerState> {
    let config = ServerStateConfig::new(root.join("journals"))
        .with_shared_memgine(Arc::new(Mutex::new(MemgineEngine::new(None))));
    let state = Arc::new(ServerState::with_config(config));
    state
        .install_host_token(HOST_TOKEN.to_string())
        .expect("install isolated host token");

    let supervisor = Arc::new(
        Supervisor::with_paths(root.join("agents.json"), root.join("logs"))
            .expect("isolated supervisor"),
    );
    supervisor
        .upsert(agent_spec(
            "scoped",
            SCOPED_TOKEN,
            Some(vec!["agents.list".to_string()]),
        ))
        .await
        .unwrap();
    supervisor
        .upsert(agent_spec(
            "pipelined",
            PIPELINED_TOKEN,
            Some(vec!["agents.list".to_string()]),
        ))
        .await
        .unwrap();
    supervisor
        .upsert(agent_spec("unscoped", UNSCOPED_TOKEN, None))
        .await
        .unwrap();
    supervisor
        .upsert(agent_spec("empty", EMPTY_TOKEN, Some(Vec::new())))
        .await
        .unwrap();
    state
        .install_supervisor(supervisor)
        .map_err(|_| ())
        .expect("install isolated supervisor");
    state
        .declagents
        .set(Arc::new(DeclRegistry::at(root.join("declagents.json"))))
        .map_err(|_| ())
        .expect("install isolated declarative-agent registry");
    state
}

async fn spawn_dispatcher(state: Arc<ServerState>) -> SocketAddr {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0)))
        .await
        .expect("bind isolated loopback listener");
    let address = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let Ok((stream, peer)) = listener.accept().await else {
                return;
            };
            let state = state.clone();
            tokio::spawn(async move {
                let Ok(websocket) = accept_async(stream).await else {
                    return;
                };
                let (write, read) = websocket.split();
                let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
            });
        }
    });
    address
}

async fn connect(address: SocketAddr) -> Ws {
    connect_async(format!("ws://{address}"))
        .await
        .expect("connect")
        .0
}

async fn call(ws: &mut Ws, id: &str, method: &str, params: serde_json::Value) -> serde_json::Value {
    ws.send(Message::Text(
        serde_json::json!({"jsonrpc":"2.0","id":id,"method":method,"params":params})
            .to_string()
            .into(),
    ))
    .await
    .expect("send request");
    loop {
        match ws
            .next()
            .await
            .expect("response frame")
            .expect("valid frame")
        {
            Message::Text(text) => {
                let value: serde_json::Value = serde_json::from_str(&text).unwrap();
                if value.get("id").and_then(serde_json::Value::as_str) == Some(id) {
                    return value;
                }
            }
            Message::Ping(_) | Message::Pong(_) => continue,
            other => panic!("unexpected frame: {other:?}"),
        }
    }
}

async fn agent_session(address: SocketAddr, id: &str, token: &str) -> (Ws, serde_json::Value) {
    let mut ws = connect(address).await;
    let auth = call(
        &mut ws,
        "auth",
        "session.auth",
        serde_json::json!({"agent_id": id, "token": token}),
    )
    .await;
    assert!(auth.get("error").is_none(), "agent auth failed: {auth}");
    (ws, auth)
}

#[tokio::test]
async fn scoped_default_and_host_sessions_keep_distinct_dispatch_authority() {
    let root = tempfile::TempDir::new().unwrap();
    let address = spawn_dispatcher(state(root.path()).await).await;

    // On an auth-disabled daemon, put both frames into the socket before
    // reading either response. Supervised auth runs inline, so the second
    // frame observes the newly bound scope rather than the unrestricted
    // pre-auth default.
    let mut pipelined = connect(address).await;
    for request in [
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "pipelined-auth",
            "method": "session.auth",
            "params": {"agent_id": "pipelined", "token": PIPELINED_TOKEN}
        }),
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "pipelined-denied",
            "method": "agents.health",
            "params": {}
        }),
    ] {
        pipelined
            .send(Message::Text(request.to_string().into()))
            .await
            .expect("send pipelined request");
    }
    let mut pipelined_responses = std::collections::HashMap::new();
    while pipelined_responses.len() < 2 {
        let Message::Text(text) = pipelined
            .next()
            .await
            .expect("pipelined response frame")
            .expect("valid pipelined response frame")
        else {
            continue;
        };
        let response: serde_json::Value = serde_json::from_str(&text).unwrap();
        if let Some(id) = response.get("id").and_then(serde_json::Value::as_str) {
            pipelined_responses.insert(id.to_string(), response);
        }
    }
    assert_eq!(
        pipelined_responses["pipelined-auth"]["result"]["method_allowlist"],
        serde_json::json!(["agents.list"])
    );
    assert_eq!(
        pipelined_responses["pipelined-denied"]["error"]["code"], -32601,
        "{}",
        pipelined_responses["pipelined-denied"]
    );
    assert!(
        pipelined_responses["pipelined-denied"]["error"]["message"]
            .as_str()
            .is_some_and(|message| message.starts_with("agent_method_not_allowed:")),
        "{}",
        pipelined_responses["pipelined-denied"]
    );

    let (mut scoped, scoped_auth) = agent_session(address, "scoped", SCOPED_TOKEN).await;
    assert_eq!(
        scoped_auth["result"]["method_allowlist"],
        serde_json::json!(["agents.list"]),
        "{scoped_auth}"
    );
    let denied = call(
        &mut scoped,
        "denied",
        "agents.health",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(denied["error"]["code"], -32601, "{denied}");
    assert_eq!(
        denied["error"]["message"],
        "agent_method_not_allowed: supervised agent token does not allow daemon method `agents.health`",
        "{denied}"
    );
    // Notification interceptors live ahead of the ordinary handler table. An
    // id-bearing probe proves the scope gate is ahead of those too, rather than
    // allowing an excluded producer event to bypass the central admission.
    let denied_interceptor = call(
        &mut scoped,
        "denied-interceptor",
        "agent.chat.event",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(denied_interceptor["error"]["code"], -32601);
    assert!(
        denied_interceptor["error"]["message"]
            .as_str()
            .is_some_and(|message| message.starts_with("agent_method_not_allowed:")),
        "{denied_interceptor}"
    );
    // The denial is request-local, not a disconnect, and the one listed method
    // still reaches its real handler afterward.
    let allowed = call(&mut scoped, "allowed", "agents.list", serde_json::json!({})).await;
    assert!(allowed["result"].as_array().is_some(), "{allowed}");

    // No field is exactly the pre-feature behavior: the same method denied
    // above reaches the handler for an otherwise-identical agent token.
    let (mut unscoped, unscoped_auth) = agent_session(address, "unscoped", UNSCOPED_TOKEN).await;
    assert!(
        unscoped_auth["result"].get("method_allowlist").is_none(),
        "absent scope must preserve the legacy auth response: {unscoped_auth}"
    );
    let default = call(
        &mut unscoped,
        "default",
        "agents.health",
        serde_json::json!({}),
    )
    .await;
    assert!(default["result"].as_array().is_some(), "{default}");

    // A present empty scope is observably different from an absent scope.
    let (mut empty, empty_auth) = agent_session(address, "empty", EMPTY_TOKEN).await;
    assert_eq!(
        empty_auth["result"]["method_allowlist"],
        serde_json::json!([]),
        "{empty_auth}"
    );
    let empty_denial = call(
        &mut empty,
        "empty-denied",
        "agents.list",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(empty_denial["error"]["code"], -32601, "{empty_denial}");

    // Host-role authentication never acquires an agent scope. Host-only
    // registration still runs and can create a scoped token.
    let mut host = connect(address).await;
    let host_auth = call(
        &mut host,
        "host-auth",
        "session.auth",
        serde_json::json!({"host_token": HOST_TOKEN}),
    )
    .await;
    assert_eq!(host_auth["result"]["role"], "host", "{host_auth}");
    let host_upsert = call(
        &mut host,
        "host-upsert",
        "agents.upsert",
        serde_json::to_value(agent_spec(
            "host-created",
            "host-created-token-0123456789abcdef",
            Some(vec!["mail.messages".to_string()]),
        ))
        .unwrap(),
    )
    .await;
    assert_eq!(host_upsert["result"]["id"], "host-created", "{host_upsert}");
    assert_eq!(
        host_upsert["result"]["method_allowlist"],
        serde_json::json!(["mail.messages"]),
        "{host_upsert}"
    );
}

async fn notify(ws: &mut Ws, method: &str, params: serde_json::Value) {
    ws.send(Message::Text(
        serde_json::json!({
            "jsonrpc":"2.0", "method":method, "params":params
        })
        .to_string()
        .into(),
    ))
    .await
    .unwrap();
}

async fn event_matching(ws: &mut Ws, needle: &str) -> serde_json::Value {
    tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            if let Message::Text(text) = ws.next().await.unwrap().unwrap() {
                if text.contains(needle) {
                    return serde_json::from_str(&text).unwrap();
                }
            }
        }
    })
    .await
    .expect("expected routed notification")
}

#[tokio::test]
async fn explicitly_granted_notifications_stream_without_bypassing_producer_ownership() {
    let root = tempfile::TempDir::new().unwrap();
    let state = state(root.path()).await;
    let supervisor = state.supervisor.get().unwrap();
    for (id, grants) in [
        (
            "publisher",
            vec![
                "agents.list",
                "agent.chat.event",
                "browser.producer.register",
                "browser.producer.frame",
                "browser.producer.presentation",
            ],
        ),
        ("muted", vec!["agents.list", "browser.producer.register"]),
    ] {
        supervisor
            .upsert(agent_spec(
                id,
                &format!("{id}-fixture-token-0123456789abcdef"),
                Some(grants.into_iter().map(str::to_string).collect()),
            ))
            .await
            .unwrap();
    }
    let address = spawn_dispatcher(state.clone()).await;
    let mut host = connect(address).await;
    let auth = call(
        &mut host,
        "host",
        "session.auth",
        serde_json::json!({"host_token":HOST_TOKEN}),
    )
    .await;
    assert!(auth.get("error").is_none(), "{auth}");
    let host_id = state
        .sessions
        .lock()
        .await
        .values()
        .find(|s| s.is_host.load(std::sync::atomic::Ordering::Acquire))
        .unwrap()
        .client_id
        .clone();
    let (mut publisher, _) = agent_session(
        address,
        "publisher",
        "publisher-fixture-token-0123456789abcdef",
    )
    .await;
    let (mut muted, _) =
        agent_session(address, "muted", "muted-fixture-token-0123456789abcdef").await;
    for id in ["publisher", "muted"] {
        state.chat_sessions.lock().await.insert(
            id.into(),
            car_server_core::session::ChatSession {
                agent_id: id.into(),
                host_client_id: host_id.clone(),
                created_at: 0,
                local_cancel: None,
            },
        );
    }
    notify(
        &mut publisher,
        "agent.chat.event",
        serde_json::json!({"session_id":"publisher", "delta":"allowed-chat"}),
    )
    .await;
    let chat = event_matching(&mut host, "allowed-chat").await;
    assert_eq!(chat["method"], "agents.chat.event");
    assert_eq!(chat["params"]["agent_id"], "publisher");
    for (id, ws) in [("publisher", &mut publisher), ("muted", &mut muted)] {
        let registered = call(
            ws,
            "register",
            "browser.producer.register",
            serde_json::json!({"conversation_id":id}),
        )
        .await;
        assert_eq!(registered["result"]["ok"], true, "{registered}");
        let subscribed = call(
            &mut host,
            "subscribe",
            "browser.view.subscribe",
            serde_json::json!({"conversation_id":id}),
        )
        .await;
        assert!(subscribed.get("error").is_none(), "{subscribed}");
    }
    let wrong_owner = call(
        &mut publisher,
        "wrong-owner",
        "browser.producer.register",
        serde_json::json!({"conversation_id":"muted"}),
    )
    .await;
    assert!(wrong_owner.get("error").is_some(), "{wrong_owner}");
    let frame = |marker| serde_json::json!({"frame":{"jpeg_base64":marker,"width":1,"height":1,"device_pixel_ratio":1.0,"captured_at":0.0}});
    notify(
        &mut publisher,
        "browser.producer.frame",
        frame("allowed-frame"),
    )
    .await;
    let delivered = event_matching(&mut host, "allowed-frame").await;
    assert_eq!(delivered["method"], "browser.view.event");
    assert_eq!(delivered["params"]["conversation_id"], "publisher");
    let presentation = |marker| serde_json::json!({"presentation":{"revision":1,"owner":"agent","current_action":marker,"pending_signin":null,"blackout_active":false,"tabs":[],"active_tab":null,"url":null,"title":null}});
    notify(
        &mut publisher,
        "browser.producer.presentation",
        presentation("allowed-presentation"),
    )
    .await;
    event_matching(&mut host, "allowed-presentation").await;
    notify(
        &mut muted,
        "agent.chat.event",
        serde_json::json!({"session_id":"muted","delta":"forbidden-chat"}),
    )
    .await;
    notify(
        &mut muted,
        "browser.producer.frame",
        frame("forbidden-frame"),
    )
    .await;
    notify(
        &mut muted,
        "browser.producer.presentation",
        presentation("forbidden-presentation"),
    )
    .await;
    let barrier = call(&mut muted, "barrier", "agents.list", serde_json::json!({})).await;
    assert!(barrier.get("error").is_none());
    let leak = tokio::time::timeout(std::time::Duration::from_millis(150), async {
        loop {
            if let Message::Text(text) = host.next().await.unwrap().unwrap() {
                assert!(
                    !text.contains("forbidden-"),
                    "ungranted notification leaked: {text}"
                );
            }
        }
    })
    .await;
    assert!(leak.is_err());
}