liminal-server 0.24.5

//! Real Lys and real standalone wire. Missing fixture inputs fail, never skip.
//!
//! The socket venue supplies an isolated real Lys install with two signed passes,
//! a mapped conversation and a revocable grant. No HTTP route is simulated here.

use super::*;
use liminal::protocol::{Frame, FrameHeader, ProtocolVersion, codec};
use liminal_protocol::wire::{
    ClientRequest, EnrollmentRequest, EnrollmentToken, ParticipantFrame, ReceiverDirection,
    ServerValue, TransportRejectionReason,
};
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::PathBuf;

type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;

#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct Fixture {
    server_config: PathBuf,
    member_pass: PathBuf,
    nonmember_pass: PathBuf,
    revoker_credential: PathBuf,
    grant: String,
    operation: String,
    conversation: u64,
    denied_name: String,
    closes_connection: bool,
}

impl Fixture {
    fn load() -> Result<Self> {
        let path = std::env::var_os("ACCESS007_LYS_FIXTURE")
            .ok_or("ACCESS007_LYS_FIXTURE is required for the real Lys wire proof")?;
        // The fixture uses the same configured key-cache cap for its small
        // metadata file. Tokens and response payloads are separate bounded files.
        let text = http::read_file(&PathBuf::from(path), configuration()?.max_keyset_bytes)?;
        let fixture: Self =
            toml::from_str(&text).map_err(|_| "real Lys fixture configuration is invalid")?;
        let _: lys_identity::grants::GrantId = fixture
            .grant
            .parse()
            .map_err(|_| "fixture grant id is invalid")?;
        let _: lys_identity::OperationId = fixture
            .operation
            .parse()
            .map_err(|_| "fixture operation id is invalid")?;
        if fixture.denied_name.is_empty() {
            return Err("fixture refusal name is required".into());
        }
        Ok(fixture)
    }
}

async fn wait_for(
    state: &StreamState,
    predicate: impl Fn(&StreamDiagnostics) -> bool,
) -> Result<StreamDiagnostics> {
    let mut changed = state.subscribe();
    loop {
        let status = changed.borrow_and_update().clone()?;
        if predicate(&status) {
            return Ok(status);
        }
        if status.stopped || matches!(status.readiness, "terminal" | "gap" | "reset") {
            return Err(status
                .last_error
                .unwrap_or("grant stream stopped before the required signal")
                .into());
        }
        changed
            .changed()
            .await
            .map_err(|_| "grant stream progress signal closed")?;
    }
}

fn write_frame(stream: &mut TcpStream, frame: &Frame) -> Result<()> {
    let mut bytes = zeroize::Zeroizing::new(vec![0; codec::encoded_len(frame)?]);
    let length = codec::encode(frame, &mut bytes)?;
    stream.write_all(&bytes[..length])?;
    Ok(())
}

fn read_bytes(stream: &mut TcpStream, bound: usize) -> Result<zeroize::Zeroizing<Vec<u8>>> {
    let mut header = [0; FrameHeader::WIRE_LEN];
    stream.read_exact(&mut header)?;
    let length = usize::try_from(u32::from_be_bytes([
        header[6], header[7], header[8], header[9],
    ]))?;
    let total = length
        .checked_add(header.len())
        .ok_or("wire frame length overflow")?;
    if total > bound {
        return Err("wire frame exceeds the configured connection bound".into());
    }
    let mut bytes = zeroize::Zeroizing::new(vec![0; total]);
    bytes[..header.len()].copy_from_slice(&header);
    stream.read_exact(&mut bytes[header.len()..])?;
    Ok(bytes)
}

fn connect(
    address: std::net::SocketAddr,
    pass: &std::path::Path,
    config: &GrantStreamConfig,
    bound: usize,
) -> Result<(TcpStream, Frame)> {
    let mut stream = TcpStream::connect(address)?;
    stream.set_nodelay(true)?;
    let pass = http::read_file(pass, config.max_credential_bytes)?;
    write_frame(
        &mut stream,
        &Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: liminal::protocol::REVOCATION_CLOSE_VERSION,
            auth_token: liminal_protocol::credential::WireCredential::lys_record(pass.as_bytes()),
        },
    )?;
    let bytes = read_bytes(&mut stream, bound)?;
    Ok((stream, codec::decode(&bytes)?.0))
}

fn enrollment(
    stream: &mut TcpStream,
    conversation: u64,
    token: [u8; 16],
    bound: usize,
) -> Result<ParticipantFrame> {
    let request = ParticipantFrame::ClientRequest(ClientRequest::Enrollment(EnrollmentRequest {
        conversation_id: conversation,
        enrollment_token: EnrollmentToken::new(token),
    }));
    let mut bytes = zeroize::Zeroizing::new(vec![
        0;
        liminal_protocol::wire::encoded_len(&request)
            .map_err(|error| format!("{error:?}"))?
    ]);
    let length = liminal_protocol::wire::encode(&request, &mut bytes)
        .map_err(|error| format!("{error:?}"))?;
    stream.write_all(&bytes[..length])?;
    let answer = read_bytes(stream, bound)?;
    Ok(
        liminal_protocol::wire::decode(&answer, ReceiverDirection::Client)
            .map_err(|error| format!("{error:?}"))?,
    )
}

fn health(address: std::net::SocketAddr, bound: usize) -> Result<serde_json::Value> {
    let mut stream = TcpStream::connect(address)?;
    stream.write_all(b"GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n")?;
    let mut bytes = Vec::new();
    stream
        .take(
            u64::try_from(bound)?
                .checked_add(1)
                .ok_or("health bound overflow")?,
        )
        .read_to_end(&mut bytes)?;
    if bytes.len() > bound {
        return Err("health answer over bound".into());
    }
    let separator = bytes
        .windows(4)
        .position(|part| part == b"\r\n\r\n")
        .ok_or("health answer has no body")?;
    Ok(serde_json::from_slice(&bytes[separator + 4..])?)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn real_lys_stream_admits_members_revokes_a_live_right_and_joins_its_waiting_owner()
-> Result<()> {
    let fixture = Fixture::load()?;
    let config = local_config(&fixture)?;
    let lys = config
        .auth
        .as_ref()
        .and_then(|auth| auth.lys.as_ref())
        .ok_or("real Lys config is absent")?;
    let stream_config = lys
        .grant_stream
        .as_ref()
        .ok_or("grant stream config is absent")?;
    if !stream_config.enforce_revocation {
        return Err("the live proof requires revocation enforcement".into());
    }
    let supervisor = crate::server::ConnectionSupervisor::from_config(&config)?;
    let endpoint = crate::health::start_health_server(
        "127.0.0.1:0".parse()?,
        crate::health::SharedReadinessState::new(
            crate::health::ReadinessState::ready_without_cluster(),
        ),
    )?;
    let mut owner = GrantStreamOwner::start_configured(
        lys,
        &supervisor,
        &endpoint,
        tokio::runtime::Handle::current(),
        None,
    )?
    .ok_or("grant stream owner is absent")?;
    let state = owner.state();
    let listener = crate::server::listener::ServerListener::bind(&config, supervisor.clone())?;
    let mut listener = listener;
    let idle = wait_for(&state, |status| {
        status.readiness == "ready" && status.requests == 2
    })
    .await?;
    let body = health(endpoint.local_addr(), stream_config.max_response_bytes)?;
    assert_eq!(body["grant_stream"]["readiness"], "ready");
    assert_eq!(
        body["grant_stream"]["cursor"],
        serde_json::to_value(idle.cursor)?
    );
    assert_eq!(
        state.snapshot()?.requests,
        2,
        "one baseline and one waiting read at idle"
    );
    let bound = config.limits.max_connection_inbox_bytes;
    let (mut member, answer) = connect(
        listener.local_addr(),
        &fixture.member_pass,
        stream_config,
        bound,
    )?;
    assert!(
        matches!(answer, Frame::ConnectAck { .. }),
        "the real member Connect must be admitted"
    );
    let enrolled = enrollment(&mut member, fixture.conversation, [1; 16], bound)?;
    assert!(
        matches!(
            enrolled,
            ParticipantFrame::ServerValue(ServerValue::EnrollBound(_))
        ),
        "the mapped member must enroll on the actual wire"
    );
    let (mut nonmember, answer) = connect(
        listener.local_addr(),
        &fixture.nonmember_pass,
        stream_config,
        bound,
    )?;
    refused_nonmember(answer, &mut nonmember, &fixture, bound)?;
    let revision = revoke(lys, stream_config, &state, &fixture).await?;
    wait_for(&state, |status| {
        status.readiness == "ready" && status.cursor.is_some_and(|cursor| cursor >= revision)
    })
    .await?;
    withdrawn(&mut member, &fixture, bound)?;
    drop(member);
    drop(nonmember);
    owner.finish()?;
    assert!(
        state.snapshot()?.stopped,
        "the worker ended before the server store flush"
    );
    let requests = state.snapshot()?.requests;
    crate::server::shutdown::run_shutdown_sequence(
        &mut listener,
        None,
        &supervisor,
        config.drain_timeout(),
    )?;
    assert_eq!(
        state.snapshot()?.requests,
        requests,
        "nothing outlives the joined owner"
    );
    endpoint.shutdown()?;
    Ok(())
}

/// The nonmember is refused at Connect or at its first enrollment, naming the denied right.
fn refused_nonmember(
    answer: Frame,
    nonmember: &mut TcpStream,
    fixture: &Fixture,
    bound: usize,
) -> Result<()> {
    match answer {
        Frame::ConnectError {
            message: Some(message),
            ..
        } => assert!(message.contains(&fixture.denied_name)),
        Frame::ConnectAck { .. } => {
            let answer = enrollment(nonmember, fixture.conversation, [2; 16], bound)?;
            let ParticipantFrame::ServerValue(ServerValue::ParticipantTransportRejected(rejected)) =
                answer
            else {
                return Err("the real nonmember enrollment was not refused".into());
            };
            let TransportRejectionReason::AccessRefused { refusal } = rejected.reason else {
                return Err("nonmember refusal was not an access decision".into());
            };
            assert!(refusal.contains(&fixture.denied_name));
        }
        _ => return Err("nonmember Connect returned an unrelated answer".into()),
    }
    Ok(())
}

/// Revokes the fixture right through the real Lys route and answers the revision it committed.
async fn revoke(
    lys: &crate::config::types::LysConfig,
    stream_config: &crate::config::types::GrantStreamConfig,
    state: &StreamState,
    fixture: &Fixture,
) -> Result<u64> {
    let credential = http::read_file(
        &fixture.revoker_credential,
        stream_config.max_credential_bytes,
    )?;
    let revoker = http::Http::new(
        lys.api_url.as_deref().unwrap_or(&lys.issuer),
        credential,
        stream_config.max_response_bytes,
        state.cancellation(),
    )?;
    let receipt: serde_json::Value = revoker.post(&format!("grants/{}/revoke", fixture.grant),
        &serde_json::json!({"operation": fixture.operation, "route": "api", "reason": "fixture right withdrawn"})).await?;
    let revision = receipt["receipt"]["index"]
        .as_u64()
        .and_then(|index| index.checked_add(1))
        .ok_or("real revoke has no receipt revision")?;
    assert!(
        receipt.get("degraded").is_none(),
        "a degraded commit is not a revocation proof"
    );
    Ok(revision)
}

/// The withdrawn right either closes the member's connection or refuses its next enrollment.
fn withdrawn(member: &mut TcpStream, fixture: &Fixture, bound: usize) -> Result<()> {
    if fixture.closes_connection {
        let bytes = read_bytes(member, bound)?;
        assert!(matches!(
            codec::decode(&bytes)?.0,
            Frame::DisconnectReason {
                reason: liminal::protocol::DisconnectReason::GrantRevoked,
                ..
            }
        ));
    } else {
        let answer = enrollment(member, fixture.conversation, [3; 16], bound)?;
        let ParticipantFrame::ServerValue(ServerValue::ParticipantTransportRejected(rejected)) =
            answer
        else {
            return Err("withdrawn right still enrolled on the real wire".into());
        };
        let TransportRejectionReason::AccessRefused { refusal } = rejected.reason else {
            return Err("withdrawal lacked its access refusal".into());
        };
        assert!(refusal.contains("grant_revoked"));
    }
    Ok(())
}

/// The fixture's server config bound to loopback alone, with no cluster or WebSocket listener.
fn local_config(fixture: &Fixture) -> Result<crate::config::ServerConfig> {
    let mut config = crate::config::file::load_config(&fixture.server_config)?;
    config.listen_address = "127.0.0.1:0".parse()?;
    config.extra_listen_addresses.clear();
    config.cluster = None;
    config.websocket = None;
    Ok(config)
}