liminal-server 0.28.2

#[path = "support/agent_pass.rs"]
mod agent_pass;
use liminal_server::auth_lys::session::{LysAdmissionRefusal, LysAdmitter, LysSession};
use liminal_server::auth_lys::{AccessResource, RefusalCause, ResourceKind, SubjectBinding};
use liminal_server::config::types::{AuthConfig, ServerConfig};
use liminal_server::config::validation;
use liminal_server::link::LinkAction;
use liminal_server::link_dial::{LinkDialOutcome, LinkDialer};
use liminal_server::server::connection::LiminalConnectionServices;
use liminal_server::server::listener::ServerListener;
use liminal_server::services::production::{ServiceBindings, machine_bearer};
#[path = "link_pass/after_bind.rs"]
mod after_bind;
#[path = "link_pass/boot.rs"]
mod boot;
#[path = "link_pass/fixture.rs"]
mod fixture;
#[path = "link_pass/signature_tests.rs"]
mod signature_tests;
#[path = "link_pass/unavailable_tests.rs"]
mod unavailable_tests;
use fixture::{Fixture, encoded};
use liminal_server::server::connection::{ConnectionSupervisor, WorkerFrontDoorServices};
use lys_core::keys::Ed25519Identity;
use std::io::Write;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use zeroize::Zeroizing;
type TestError = Box<dyn std::error::Error + Send + Sync>;
type TestResult = Result<(), TestError>;
#[derive(Clone)]
struct LogWriter(Arc<Mutex<Vec<u8>>>);
impl Write for LogWriter {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.0
            .lock()
            .map_err(|error| std::io::Error::other(error.to_string()))?
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}
#[test]
fn an_agent_pass_is_asked_at_boot_with_the_signature_lys_verifies() -> TestResult {
    let fixture = Fixture::start(None)?;
    let auth = fixture.auth("")?;
    started(&auth, |server| {
        assert_eq!(fixture.posts.load(Ordering::Acquire), 1);
        let host = server
            .link_host_for_tests()
            .ok_or("machine pass did not install a host")?;
        let machine = Ed25519Identity::from_seed(&Zeroizing::new([1; 32]));
        assert_eq!(
            host.local().machine().ed25519_public(),
            &machine.public_key_bytes()
        );
        let subject =
            SubjectBinding::link("network".into(), "orders-link".into(), "orders".into())?;
        host.rights()
            .holds(LinkAction::Import, &subject, fixture.expiry - 1)?;
        Ok(())
    })?;
    fixture.finish()
}
fn refused_boot(fixture: Fixture, name: &str, expected_posts: usize) -> TestResult {
    let logs = Arc::new(Mutex::new(Vec::new()));
    let writer = LogWriter(Arc::clone(&logs));
    let subscriber = tracing_subscriber::fmt()
        .without_time()
        .with_ansi(false)
        .with_writer(move || writer.clone())
        .finish();
    let _logging = tracing::subscriber::set_default(subscriber);
    liminal_server::metrics::init();
    let before = refusal_count(name)?;
    let auth = fixture.auth("[lys.services]\nmachine = 'computer'\nhost = 'fixture'")?;
    started(&auth, |server| {
        // The configured link retry owner may already have asked again, so
        // the count is a floor (ACCESS-009 R1).
        assert!(
            fixture.posts.load(Ordering::Acquire) >= expected_posts,
            "the boot did not ask"
        );
        // ACCESS-009 R1: the host is installed with no pass held, so an
        // open asks again; until Lys issues, every right is unavailable.
        let host = server
            .link_host_for_tests()
            .ok_or("refused pass installed no host")?;
        let subject =
            SubjectBinding::link("network".into(), "orders-link".into(), "orders".into())?;
        assert!(
            host.rights()
                .holds(LinkAction::Import, &subject, fixture.expiry - 1)
                .is_err(),
            "a right was held without a pass"
        );
        let bindings = ServiceBindings::configure(
            auth.lys.as_ref().ok_or("no Lys config")?,
            Arc::new(LiminalConnectionServices::empty()?),
            tokio::runtime::Handle::current(),
            fixture.expiry - 1,
        )?;
        assert!(bindings.is_none(), "refused pass configured services");
        let output = String::from_utf8(logs.lock().map_err(|error| error.to_string())?.clone())?;
        assert!(output.contains(name), "unnamed refusal: {output}");
        assert!(!output.contains(&fixture.peer_pass), "pass leaked");
        assert!(
            refusal_count(name)? > before,
            "the boot refusal was not counted"
        );
        Ok(())
    })?;
    fixture.finish()
}
fn refusal_count(name: &str) -> Result<u64, TestError> {
    let registry = liminal::metrics::global_registry().ok_or("metrics registry absent")?;
    Ok(registry
        .snapshot()
        .metrics
        .iter()
        .find_map(|metric| {
            if metric.name == "liminal_machine_pass_refusals_total"
                && metric
                    .labels
                    .iter()
                    .any(|(key, value)| key == "reason" && value == name)
                && let liminal::metrics::MetricValue::Counter(value) = metric.value
            {
                return Some(value);
            }
            None
        })
        .unwrap_or(0))
}
#[test]
fn a_refused_dial_is_named_at_boot_and_the_server_still_starts() -> TestResult {
    for (server_name, log_name) in [
        ("agent_dial_refused", "agent_dial_refused"),
        ("HolderRetired", "HolderRetired"),
        ("AgentSignatureRefused", "agent_signature_refused"),
        ("ProviderUnavailable", "ProviderUnavailable"),
    ] {
        refused_boot(Fixture::start(Some(server_name))?, log_name, 1)?;
    }
    Ok(())
}
#[test]
fn a_refused_boot_pass_names_itself_and_the_server_starts() -> TestResult {
    boot::refused_boot()
}
#[test]
fn the_server_listens_before_lys_answers_its_machine_pass() -> TestResult {
    after_bind::listens_before_lys_answers()
}
#[test]
fn an_unreachable_lys_is_named_without_stopping_the_server() -> TestResult {
    let fixture = Fixture::start(Some("disconnect"))?;
    refused_boot(fixture, "lys_could_not_be_asked", 0)
}
#[test]
fn a_pass_signature_refusal_is_named_without_stopping_the_server() -> TestResult {
    let fixture = Fixture::start(None)?;
    let wrong = Ed25519Identity::from_seed(&Zeroizing::new([8; 32]));
    let keys = serde_json::json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"federation",
        "x":encoded(&wrong.public_key_bytes()),"alg":"EdDSA","use":"sig"}]});
    std::fs::write(fixture.root.path().join("keys"), serde_json::to_vec(&keys)?)?;
    refused_boot(fixture, "signature_refused", 1)
}
#[test]
fn pass_path_and_dial_together_are_refused_at_config() -> TestResult {
    let fixture = Fixture::start(None)?;
    let mut config: ServerConfig = toml::from_str(
        "listen_address = '127.0.0.1:4000'\nhealth_listen_address = '127.0.0.1:4001'\ndrain_timeout_ms = 0\nchannels = []\nrouting_rules = []",
    )?;
    config.persistence_path = Some(fixture.root.path().to_path_buf());
    config.auth = Some(fixture.auth("pass_path = 'machine.pass'")?);
    let refusal = validation::validate(&mut config, None)
        .err()
        .ok_or("two pass sources were admitted")?
        .to_string();
    assert!(
        refusal.contains("pass_path") && refusal.contains("agent_id"),
        "{refusal}"
    );
    let valid = fixture.auth("")?;
    for (field, mutation) in [
        ("agent_id", 0),
        ("audience", 1),
        ("audience", 2),
        ("api_url", 3),
        ("pass_path", 4),
        ("api_url", 5),
    ] {
        let mut auth = valid.clone();
        let lys = auth.lys.as_mut().ok_or("no Lys config")?;
        let machine = lys.machine.as_mut().ok_or("no machine config")?;
        match mutation {
            0 => machine.agent_id = "../computer".into(),
            1 => machine.audience.clear(),
            2 => machine.audience = " ".into(),
            3 => lys.api_url = None,
            5 => lys.api_url = Some("http://issuer.example".into()),
            _ => {
                machine.agent_id.clear();
                machine.audience.clear();
            }
        }
        config.auth = Some(auth);
        let refusal = validation::validate(&mut config, None)
            .err()
            .ok_or("invalid source admitted")?
            .to_string();
        assert!(refusal.contains(field), "{refusal}");
    }
    let mut auth = valid;
    let lys = auth.lys.as_mut().ok_or("no Lys config")?;
    let machine = lys.machine.as_mut().ok_or("no machine")?;
    machine.agent_id.clear();
    machine.audience.clear();
    machine.pass_path = fixture.root.path().join("pass");
    std::fs::write(&machine.pass_path, &fixture.peer_pass)?;
    config.auth = Some(auth);
    validation::validate(&mut config, None)?;
    started(config.auth.as_ref().ok_or("no auth config")?, |server| {
        assert!(server.link_host_for_tests().is_some());
        let lys = config
            .auth
            .as_ref()
            .and_then(|auth| auth.lys.as_ref())
            .ok_or("no Lys config")?;
        std::fs::remove_file(&lys.machine.as_ref().ok_or("no machine")?.pass_path)?;
        assert!(
            machine_bearer(lys, fixture.expiry)? == fixture.peer_pass,
            "file source holder changed"
        );
        assert_eq!(
            fixture.posts.load(Ordering::Acquire),
            0,
            "file source asked Lys"
        );
        Ok(())
    })?;
    fixture.finish()
}
#[test]
fn an_expired_pass_is_asked_again_only_at_the_next_refused_open() -> TestResult {
    let fixture = Fixture::start(None)?;
    let auth = fixture.auth("")?;
    started(&auth, |server| {
        let host = server.link_host_for_tests().ok_or("no link host")?;
        let session = server
            .lys_admitter_for_tests()
            .ok_or("no Lys admitter")?
            .admit(fixture.peer_pass.as_bytes(), fixture.expiry)?;
        let subject =
            SubjectBinding::link("network".into(), "orders-link".into(), "orders".into())?;
        let link = AccessResource {
            kind: ResourceKind::Link,
            id: "orders-link".into(),
        };
        assert_eq!(fixture.posts.load(Ordering::Acquire), 1);
        host.rights()
            .holds(LinkAction::Import, &subject, fixture.expiry - 1)?;
        refused(
            session.open_link(&link, LinkAction::Export, fixture.expiry - 31),
            RefusalCause::DeliberateUnavailable,
        )?;
        assert_eq!(fixture.posts.load(Ordering::Acquire), 1);
        refused(
            host.rights()
                .holds(LinkAction::Import, &subject, fixture.expiry),
            RefusalCause::Expired,
        )?;
        assert_eq!(
            fixture.posts.load(Ordering::Acquire),
            1,
            "crossing refreshed"
        );
        refused(
            session.open_link(&link, LinkAction::Export, fixture.expiry),
            RefusalCause::DeliberateUnavailable,
        )?;
        assert_eq!(fixture.posts.load(Ordering::Acquire), 2);
        host.rights()
            .holds(LinkAction::Import, &subject, fixture.expiry)?;
        refused(
            session.open_link(&link, LinkAction::Export, fixture.expiry),
            RefusalCause::DeliberateUnavailable,
        )?;
        assert_eq!(fixture.posts.load(Ordering::Acquire), 2);
        Ok(())
    })?;
    fixture.finish()
}
fn refused<T>(
    answer: Result<T, liminal_server::auth_lys::AccessDecisionError>,
    cause: RefusalCause,
) -> TestResult {
    let refusal = answer.err().ok_or("refused operation admitted")?;
    assert_eq!(refusal.cause, cause);
    Ok(())
}
#[test]
fn an_agent_pass_is_renewed_before_expiry_on_an_open() -> TestResult {
    let fixture = Fixture::start(None)?;
    let auth = fixture.auth("")?;
    started(&auth, |server| {
        let session = server
            .lys_admitter_for_tests()
            .ok_or("no admitter")?
            .admit(fixture.peer_pass.as_bytes(), fixture.expiry - 1)?;
        let link = AccessResource {
            kind: ResourceKind::Link,
            id: "orders-link".into(),
        };
        refused(
            session.open_link(&link, LinkAction::Export, fixture.expiry - 1),
            RefusalCause::DeliberateUnavailable,
        )?;
        assert_eq!(fixture.posts.load(Ordering::Acquire), 2);
        let lys = auth.lys.as_ref().ok_or("no Lys config")?;
        assert!(
            machine_bearer(lys, fixture.expiry - 1)? == fixture.peer_pass,
            "renewal did not update the held pass"
        );
        server
            .link_host_for_tests()
            .ok_or("no host")?
            .rights()
            .holds(
                LinkAction::Import,
                &SubjectBinding::link("network".into(), "orders-link".into(), "orders".into())?,
                fixture.expiry,
            )?;
        Ok(())
    })?;
    fixture.finish()
}
#[test]
fn an_outbound_open_presents_the_dial_sourced_pass() -> TestResult {
    let fixture = Fixture::start(None)?;
    let (importing_host, admitting) = fixture.importer()?;
    let mut auth = fixture.auth("")?;
    let lys = auth.lys.as_mut().ok_or("no Lys config")?;
    let machine = lys.machine.as_mut().ok_or("no machine config")?;
    machine.machines.insert(
        "importer".into(),
        hex::encode(importing_host.local().machine().ed25519_public()),
    );
    let observed = Arc::new(Mutex::new(Vec::new()));
    let importer =
        ConnectionSupervisor::builder(Arc::new(WorkerFrontDoorServices::new())).build()?;
    importer.install_lys_for_tests(
        Arc::new(ObservedAdmitter {
            inner: admitting,
            passes: Arc::clone(&observed),
        }),
        "network".into(),
        Arc::clone(&importing_host),
    )?;
    let config: ServerConfig = toml::from_str(
        "listen_address = '127.0.0.1:0'\nhealth_listen_address = '127.0.0.1:0'\ndrain_timeout_ms = 0\nchannels = []\nrouting_rules = []",
    )?;
    let listener = ServerListener::bind(&config, importer.clone())?;
    let entry = liminal_server::config::types::OutboundLinkConfig {
        link_id: "orders-link".into(),
        peer_holder: "importer".into(),
        peer_address: Some(listener.local_addr().to_string()),
        exports: vec!["orders".into()],
    };
    started(&auth, |server| {
        let host = server.link_host_for_tests().ok_or("no link host")?;
        let lys = auth.lys.as_ref().ok_or("no Lys config")?;
        let machine = lys.machine.as_ref().ok_or("no machine")?;
        assert!(machine.pass_path.as_os_str().is_empty());
        let dialer = LinkDialer::new(host, machine);
        for renewed in [false, true] {
            if renewed {
                expire(&fixture, server)?;
            }
            let LinkDialOutcome::Opened(link) = dialer.open(&entry)? else {
                return Err("dial-sourced open refused".into());
            };
            assert_eq!(link.transport.peer(), &importing_host.local().machine());
            drop(link);
        }
        let presented = observed.lock().map_err(|error| error.to_string())?;
        assert_eq!(presented.len(), 2);
        assert!(
            presented[0] == fixture.first_pass.as_bytes(),
            "first open presented another pass"
        );
        assert!(
            presented[1] == fixture.peer_pass.as_bytes(),
            "next open did not present renewed pass"
        );
        drop(presented);
        assert_eq!(fixture.posts.load(Ordering::Acquire), 2);
        Ok(())
    })?;
    listener.shutdown()?;
    importer.shutdown();
    fixture.finish()
}
#[test]
fn the_services_hold_the_dial_sourced_pass() -> TestResult {
    let fixture = Fixture::start(None)?;
    let auth = fixture.auth("[lys.services]\nmachine = 'computer'\nhost = 'fixture'")?;
    started(&auth, |server| {
        let lys = auth.lys.as_ref().ok_or("no Lys config")?;
        assert!(server.link_host_for_tests().is_some());
        let bearer = machine_bearer(lys, fixture.expiry - 1)?;
        assert!(bearer == fixture.first_pass, "services held another pass");
        let bindings = ServiceBindings::configure(
            lys,
            Arc::new(LiminalConnectionServices::empty()?),
            tokio::runtime::Handle::current(),
            fixture.expiry - 1,
        )?;
        let bindings = bindings.ok_or("services did not bind the dial pass")?;
        let reading = br#"{"agent":"seat","session":"session","context_percent":30,"at_ms":10}"#;
        bindings.context_reporter().report(reading, "reading")?;
        assert_eq!(fixture.posts.load(Ordering::Acquire), 1);
        expire(&fixture, server)?;
        bindings.context_reporter().report(reading, "reading")?;
        assert_eq!(fixture.usages.load(Ordering::Acquire), 2);
        assert_eq!(fixture.posts.load(Ordering::Acquire), 2);
        Ok(())
    })?;
    fixture.finish()
}
fn started(auth: &AuthConfig, run: impl FnOnce(&ConnectionSupervisor) -> TestResult) -> TestResult {
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()?;
    let _guard = runtime.enter();
    let server = ConnectionSupervisor::builder(Arc::new(WorkerFrontDoorServices::new()))
        .auth(auth)?
        .build()?;
    let answer = run(&server);
    server.shutdown();
    answer
}

fn expire(fixture: &Fixture, server: &ConnectionSupervisor) -> TestResult {
    let session = server
        .lys_admitter_for_tests()
        .ok_or("no admitter")?
        .admit(fixture.peer_pass.as_bytes(), fixture.expiry)?;
    let link = AccessResource {
        kind: ResourceKind::Link,
        id: "orders-link".into(),
    };
    refused(
        session.open_link(&link, LinkAction::Export, fixture.expiry),
        RefusalCause::DeliberateUnavailable,
    )
}
struct ObservedAdmitter {
    inner: Arc<dyn LysAdmitter>,
    passes: Arc<Mutex<Vec<Vec<u8>>>>,
}
impl std::fmt::Debug for ObservedAdmitter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ObservedAdmitter").finish_non_exhaustive()
    }
}
impl LysAdmitter for ObservedAdmitter {
    fn admit(&self, pass: &[u8], now: u64) -> Result<Box<dyn LysSession>, LysAdmissionRefusal> {
        self.passes
            .lock()
            .map_err(|_| LysAdmissionRefusal {
                reason: "fixture pass observation refused".into(),
            })?
            .push(pass.to_vec());
        self.inner.admit(pass, now)
    }
}