car-server-core 0.55.0

//! Protocol-v3 negotiation and fail-closed host/auth surface tests over a real
//! `run_dispatch` WebSocket connection.

use car_memgine::MemgineEngine;
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use futures::{SinkExt, StreamExt};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
use tempfile::TempDir;
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>>;

fn state(journal_dir: std::path::PathBuf) -> Arc<ServerState> {
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let config = ServerStateConfig::new(journal_dir).with_shared_memgine(engine);
    Arc::new(ServerState::with_config(config))
}

async fn spawn_dispatcher(state: Arc<ServerState>, connections: usize) -> SocketAddr {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0)))
        .await
        .expect("bind loopback");
    let address = listener.local_addr().expect("local address");
    tokio::spawn(async move {
        for _ in 0..connections {
            let (stream, peer) = listener.accept().await.expect("accept");
            let socket = accept_async(stream).await.expect("WebSocket handshake");
            let (write, read) = socket.split();
            let state = state.clone();
            tokio::spawn(async move {
                let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
            });
        }
    });
    address
}

async fn call(
    socket: &mut Ws,
    id: &str,
    method: &str,
    params: serde_json::Value,
) -> serde_json::Value {
    socket
        .send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "method": method,
                "params": params,
            })
            .to_string()
            .into(),
        ))
        .await
        .expect("send request");
    let text = socket
        .next()
        .await
        .expect("response frame")
        .expect("response frame ok")
        .into_text()
        .expect("text response");
    serde_json::from_str(&text).expect("parse response")
}

async fn negotiate(socket: &mut Ws, id: &str) -> serde_json::Value {
    call(
        socket,
        id,
        "server.handshake",
        serde_json::json!({
            "protocol_version": car_proto::PROTOCOL_VERSION,
            "required_capabilities": car_proto::REQUIRED_CLIENT_CAPABILITIES,
            "optional_capabilities": [],
        }),
    )
    .await
}

fn assert_handshake_required(response: &serde_json::Value, method: &str) {
    assert_eq!(
        response["error"]["code"],
        car_proto::PROTOCOL_HANDSHAKE_REQUIRED_ERROR_CODE,
        "{method} should fail with the typed handshake-required code: {response}"
    );
    assert!(
        response["error"]["message"]
            .as_str()
            .unwrap_or_default()
            .starts_with(car_proto::PROTOCOL_HANDSHAKE_REQUIRED_MESSAGE_PREFIX),
        "{method} should carry the stable handshake-required prefix: {response}"
    );
    assert!(
        response.get("result").is_none(),
        "{method} must not dispatch before negotiation: {response}"
    );
}

#[tokio::test]
async fn auth_and_host_surfaces_require_exact_v3_before_dispatch() {
    let journal = TempDir::new().expect("journal tempdir");
    let address = spawn_dispatcher(state(journal.path().to_path_buf()), 2).await;

    let (mut legacy, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect legacy client");

    for (index, (method, params)) in [
        ("auth.start", serde_json::json!({})),
        (
            "auth.complete",
            serde_json::json!({
                "redirect_uri": "http://127.0.0.1/callback",
                "code": "must-not-be-consumed",
                "verifier": "legacy-verifier",
                "attempt_id": "legacy-attempt",
            }),
        ),
        (
            "auth.completion_status",
            serde_json::json!({ "attempt_id": "legacy-attempt" }),
        ),
        ("auth.status", serde_json::json!({})),
        ("host.subscribe", serde_json::json!({})),
    ]
    .into_iter()
    .enumerate()
    {
        let response = call(&mut legacy, &format!("legacy-{index}"), method, params).await;
        assert_handshake_required(&response, method);
    }

    for (id, params) in [
        ("missing-version", serde_json::json!({})),
        (
            "string-version",
            serde_json::json!({ "protocol_version": "2" }),
        ),
        (
            "mismatch",
            serde_json::json!({ "protocol_version": car_proto::PROTOCOL_VERSION - 1 }),
        ),
    ] {
        let mismatch = call(&mut legacy, id, "server.handshake", params).await;
        assert_eq!(
            mismatch["error"]["code"],
            car_proto::PROTOCOL_VERSION_MISMATCH_ERROR_CODE
        );
        assert!(mismatch["error"]["message"]
            .as_str()
            .unwrap_or_default()
            .starts_with(car_proto::PROTOCOL_VERSION_MISMATCH_MESSAGE_PREFIX));
    }

    // A mismatch must not partially negotiate the session or allow a browser
    // flow to start.
    let after_mismatch = call(
        &mut legacy,
        "after-mismatch",
        "auth.start",
        serde_json::json!({}),
    )
    .await;
    assert_handshake_required(&after_mismatch, "auth.start");

    let unsupported = call(
        &mut legacy,
        "unsupported-mandatory",
        "server.handshake",
        serde_json::json!({
            "protocol_version": car_proto::PROTOCOL_VERSION,
            "required_capabilities": ["future.mandatory.v1"],
        }),
    )
    .await;
    assert_eq!(
        unsupported["error"]["code"],
        car_proto::PROTOCOL_CAPABILITY_MISMATCH_ERROR_CODE
    );
    assert!(unsupported["error"]["message"]
        .as_str()
        .unwrap_or_default()
        .starts_with(car_proto::PROTOCOL_CAPABILITY_MISMATCH_MESSAGE_PREFIX));
    eprintln!("C1_WS_UNSUPPORTED_MANDATORY_CAPABILITY={unsupported}");

    let malformed = call(
        &mut legacy,
        "malformed-capabilities",
        "server.handshake",
        serde_json::json!({
            "protocol_version": car_proto::PROTOCOL_VERSION,
            "required_capabilities": "models.catalog-identity.v1",
        }),
    )
    .await;
    assert_eq!(
        malformed["error"]["code"],
        car_proto::PROTOCOL_CAPABILITY_MISMATCH_ERROR_CODE
    );

    let version_only = call(
        &mut legacy,
        "version-only",
        "server.handshake",
        serde_json::json!({ "protocol_version": car_proto::PROTOCOL_VERSION }),
    )
    .await;
    assert_eq!(
        version_only["result"]["negotiated_capabilities"],
        serde_json::json!([])
    );
    let snapshot_without_capability = call(
        &mut legacy,
        "snapshot-without-capability",
        "models.catalog_snapshot",
        serde_json::json!({}),
    )
    .await;
    assert!(snapshot_without_capability["error"]["message"]
        .as_str()
        .unwrap_or_default()
        .starts_with(car_proto::PROTOCOL_CAPABILITY_MISMATCH_MESSAGE_PREFIX));

    let (mut compatible, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect v3 client");
    let handshake = negotiate(&mut compatible, "v3").await;
    assert_eq!(
        handshake["result"]["protocol_version"],
        car_proto::PROTOCOL_VERSION
    );
    assert_eq!(
        handshake["result"]["client_protocol_version"],
        car_proto::PROTOCOL_VERSION
    );
    assert_eq!(
        handshake["result"]["negotiated_capabilities"],
        serde_json::json!(car_proto::REQUIRED_CLIENT_CAPABILITIES)
    );
    assert!(
        handshake["result"]["assistant_name"]
            .as_str()
            .is_some_and(|name| !name.is_empty()),
        "the handshake must carry the configured assistant name: {handshake}"
    );
    assert!(
        handshake["result"]["assistant_aliases"]
            .as_array()
            .is_some_and(|aliases| aliases.iter().all(serde_json::Value::is_string)),
        "the handshake must carry assistant aliases as strings: {handshake}"
    );
    assert_eq!(
        handshake["result"]["assistant_brand"],
        car_identity::BRAND_NAME,
        "the handshake must carry the stable assistant brand"
    );
    assert!(
        handshake["result"].get("user_name").is_none(),
        "the public handshake must not expose the user's personal name"
    );
    eprintln!("C1_WS_NEGOTIATED_HANDSHAKE={handshake}");

    // Same-version re-handshake is idempotent.
    let repeated = negotiate(&mut compatible, "v3-again").await;
    assert_eq!(
        repeated["result"]["protocol_version"],
        car_proto::PROTOCOL_VERSION
    );

    let subscribed = call(
        &mut compatible,
        "subscribe",
        "host.subscribe",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(subscribed["result"]["subscribed"], true, "{subscribed}");
}

#[tokio::test]
async fn reconnect_starts_unnegotiated_and_must_handshake_again() {
    let journal = TempDir::new().expect("journal tempdir");
    let address = spawn_dispatcher(state(journal.path().to_path_buf()), 2).await;

    let (mut first, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect first session");
    assert!(negotiate(&mut first, "first-handshake")
        .await
        .get("error")
        .is_none());
    let first_subscribe = call(
        &mut first,
        "first-subscribe",
        "host.subscribe",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(first_subscribe["result"]["subscribed"], true);
    first.close(None).await.expect("close first session");

    let (mut second, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect second session");
    let before_handshake = call(
        &mut second,
        "second-subscribe-early",
        "host.subscribe",
        serde_json::json!({}),
    )
    .await;
    assert_handshake_required(&before_handshake, "host.subscribe");

    assert!(negotiate(&mut second, "second-handshake")
        .await
        .get("error")
        .is_none());
    let after_handshake = call(
        &mut second,
        "second-subscribe",
        "host.subscribe",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(after_handshake["result"]["subscribed"], true);
}

/// car#1050: a client's own build was unobservable. `car --version` answers for
/// the bundled CLI — on macOS `/usr/local/bin/car` is a symlink into
/// `CarHost.app` — which is a different component from the `car-runtime`
/// npm/PyPI package that made the call, and the only place that package's
/// version ever appeared was inside the skew warning's prose. The daemon
/// received `client_version` on every handshake and threw it away; it now
/// echoes it, so the number is readable off the reply every host already gets.
#[tokio::test]
async fn handshake_echoes_the_client_version_it_was_told() {
    let journal = TempDir::new().expect("journal tempdir");
    let address = spawn_dispatcher(state(journal.path().to_path_buf()), 3).await;

    let (mut socket, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect client");
    let reply = call(
        &mut socket,
        "hs-echo",
        "server.handshake",
        serde_json::json!({
            "protocol_version": car_proto::PROTOCOL_VERSION,
            "client_version": "0.46.1",
        }),
    )
    .await;
    assert_eq!(reply["result"]["client_version"], "0.46.1", "{reply}");
    assert_eq!(
        reply["result"]["server_version"],
        env!("CARGO_PKG_VERSION"),
        "echoing the client's version must not displace the daemon's own: {reply}"
    );

    // Client-controlled and otherwise unbounded, so it is truncated before it
    // reaches the reply or the log.
    let (mut shouty, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect verbose client");
    let reply = call(
        &mut shouty,
        "hs-long",
        "server.handshake",
        serde_json::json!({
            "protocol_version": car_proto::PROTOCOL_VERSION,
            "client_version": "9".repeat(4096),
        }),
    )
    .await;
    assert_eq!(
        reply["result"]["client_version"]
            .as_str()
            .expect("echoed as a string")
            .len(),
        64,
        "an over-long report is clamped, not echoed whole: {reply}"
    );

    // A client that reports nothing must not be silently read as agreeing with
    // the daemon — that is the conflation the issue reported, one layer down.
    let (mut silent, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect silent client");
    let reply = call(
        &mut silent,
        "hs-silent",
        "server.handshake",
        serde_json::json!({ "protocol_version": car_proto::PROTOCOL_VERSION }),
    )
    .await;
    assert_eq!(reply["result"]["client_version"], "unknown", "{reply}");
}

#[tokio::test]
async fn capability_negotiation_occurs_only_after_transport_auth() {
    const TOKEN: &str = "protocol-v3-authenticated-capability-token";
    let journal = TempDir::new().expect("journal tempdir");
    let state = state(journal.path().to_path_buf());
    state
        .install_auth_token(TOKEN.to_string())
        .expect("install auth token");
    let address = spawn_dispatcher(state, 2).await;

    let (mut unauthenticated, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect unauthenticated client");
    let rejected = negotiate(&mut unauthenticated, "pre-auth").await;
    assert_eq!(rejected["error"]["code"], -32001, "{rejected}");

    let (mut authenticated, _) = connect_async(format!("ws://{address}"))
        .await
        .expect("connect authenticated client");
    let auth = call(
        &mut authenticated,
        "auth",
        "session.auth",
        serde_json::json!({"token": TOKEN}),
    )
    .await;
    assert_eq!(auth["result"]["ok"], true, "{auth}");
    let handshake = negotiate(&mut authenticated, "post-auth").await;
    assert_eq!(
        handshake["result"]["negotiated_capabilities"],
        serde_json::json!(car_proto::REQUIRED_CLIENT_CAPABILITIES)
    );
}