liminal-server 0.23.1

//! ACCESS-005 at the connection: the revocation fence at `Connect` (R3), the
//! named refusals (R5) and the live withdrawal of rights (R4), through
//! `apply_frame` as a connection process applies frames, against a real
//! `RevocationAuthority` whose Lys seams are fixtures standing in for the
//! unpublished DIRECTORY-089 producer.

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use ed25519_dalek::{Signer, SigningKey};
use liminal::protocol::{
    CausalContext, Frame, MessageEnvelope, ProtocolError, ProtocolVersion,
    REVOCATION_CLOSE_VERSION, SchemaId,
};

use super::apply_frame;
use crate::ServerError;
use crate::auth_lys::HotAction;
use crate::auth_pass::{PassPrincipal, PassVerifier, WirePassV1};
use crate::auth_revocation::{
    Applied, AuthenticatedBinding, BoundRightClaim, Digest32, GrantBindingSource, GrantFeedChange,
    GrantFeedEvent, GrantId, GrantLogId, Readiness, Revision, RevocationAuthority,
    RevocationEffect, RevocationPolicy, RevocationRefusal, RightAction, UnpublishedGrantDecisions,
};
use crate::server::connection::notifier::ConnectionNotifier;
use crate::server::connection::state::{ConnectionProcessState, FrameAction};
use crate::server::connection::supervisor::runtime::ConnectionRuntime;
use crate::server::connection::worker_front_door::WorkerFrontDoorServices;

type Failure = Box<dyn std::error::Error>;

const PID: u64 = 11;
const REGISTRY_SEED: [u8; 32] = [42; 32];
const HOLDER: &str = "watcher";
const LIVE: &str = "watcher.";
const BEARER: &[u8] = b"bearer-secret";
const ISSUER: Digest32 = Digest32([7; 32]);
const AUDIENCE: &str = "liminal";

/// Counts the pass attaches the runtime announces.
#[derive(Debug, Default)]
struct Attaches(AtomicU64);

impl ConnectionNotifier for Attaches {
    fn on_worker_registered(
        &self,
        _pid: u64,
        _registration: &liminal::protocol::WorkerRegistration,
    ) -> Result<(), ServerError> {
        Ok(())
    }

    fn on_worker_unregistered(&self, _pid: u64) {}

    fn on_pass_attached(&self, _pid: u64, _principal: &PassPrincipal) {
        self.0.fetch_add(1, Ordering::SeqCst);
    }
}

/// The producer fixture: the binding Lys would have produced for each
/// credential, and the contract digest (BLAKE3 here, standing in for the
/// unpublished canonical digest).
#[derive(Debug, Default)]
struct Producer {
    bindings: Mutex<BTreeMap<Vec<u8>, AuthenticatedBinding>>,
}

impl Producer {
    fn produce(&self, credential: &[u8], binding: AuthenticatedBinding) -> Result<(), Failure> {
        self.bindings
            .lock()
            .map_err(|_| "producer poisoned")?
            .insert(credential.to_vec(), binding);
        Ok(())
    }
}

impl GrantBindingSource for Producer {
    fn pass_digest(&self, credential: &[u8]) -> Result<Digest32, RevocationRefusal> {
        Ok(Digest32(*blake3::hash(credential).as_bytes()))
    }

    fn binding_for(&self, credential: &[u8]) -> Result<AuthenticatedBinding, RevocationRefusal> {
        self.bindings
            .lock()
            .map_err(|_| RevocationRefusal::BindingMissing)?
            .get(credential)
            .cloned()
            .ok_or(RevocationRefusal::BindingMissing)
    }
}

fn log() -> Result<GrantLogId, Failure> {
    Ok(GrantLogId::new("grants-L", 0)?)
}

fn grant(id: &str) -> Result<GrantId, Failure> {
    Ok(GrantId::new(id)?)
}

fn evidence(revision: u64) -> Digest32 {
    let mut bytes = [0; 32];
    for (slot, byte) in bytes.iter_mut().zip(revision.to_be_bytes()) {
        *slot = byte;
    }
    Digest32(bytes)
}

fn revoke(revision: u64, id: &str) -> Result<GrantFeedEvent, Failure> {
    Ok(GrantFeedEvent::Change {
        log: log()?,
        revision: Revision::new(revision),
        change: GrantFeedChange::Revoke(grant(id)?),
        evidence: evidence(revision),
    })
}

/// A pass for the holder, signed by the registry key the runtime trusts.
fn pass() -> Result<Vec<u8>, Failure> {
    let key = SigningKey::from_bytes(&REGISTRY_SEED);
    let mut pass = WirePassV1 {
        participant: HOLDER.as_bytes().to_vec(),
        public_key: key.verifying_key().to_bytes(),
        conversations: Vec::new(),
        live: LIVE.to_owned(),
        may_enroll: false,
        issued_at: 0,
        expires_at: u64::MAX,
        signature: [0; 64],
    };
    let unsigned = pass
        .canonical_unsigned_bytes()
        .map_err(|error| format!("unsigned pass encoding failed: {error:?}"))?;
    pass.signature = key.sign(&unsigned).to_bytes();
    Ok(pass
        .canonical_bytes()
        .map_err(|error| format!("signed pass encoding failed: {error:?}"))?)
}

/// The binding for `credential`: Connect on `conn`, publish under
/// `watcher.orders.` on `publish`, subscribe under `watcher.` on `subscribe`.
fn binding(credential: &[u8], required: u64) -> Result<AuthenticatedBinding, Failure> {
    Ok(AuthenticatedBinding {
        issuer: ISSUER,
        audience: AUDIENCE.to_owned(),
        holder: HOLDER.as_bytes().to_vec(),
        pass_digest: Digest32(*blake3::hash(credential).as_bytes()),
        log: log()?,
        required: Revision::new(required),
        rights: vec![
            BoundRightClaim {
                action: RightAction::Connect,
                prefix: String::new(),
                chains: vec![vec![grant("conn")?, grant("root")?]],
            },
            BoundRightClaim {
                action: RightAction::Hot(HotAction::Publish),
                prefix: "watcher.orders.".to_owned(),
                chains: vec![vec![grant("publish")?, grant("root")?]],
            },
            BoundRightClaim {
                action: RightAction::Hot(HotAction::Subscribe),
                prefix: LIVE.to_owned(),
                chains: vec![vec![grant("subscribe")?, grant("root")?]],
            },
        ],
    })
}

struct Fixture {
    runtime: ConnectionRuntime,
    authority: Arc<RevocationAuthority>,
    attaches: Arc<Attaches>,
    credential: Vec<u8>,
}

/// A runtime trusting the registry key (and `bearer` when given), under a
/// revocation authority whose stream is `Ready` at 40 when `ready` is set.
fn fixture(bearer: Option<&[u8]>, ready: bool) -> Result<Fixture, Failure> {
    let attaches = Arc::new(Attaches::default());
    let registry = SigningKey::from_bytes(&REGISTRY_SEED).verifying_key();
    let verifier = PassVerifier::new(registry.as_bytes(), 30)
        .map_err(|error| format!("verifier: {error:?}"))?;
    let runtime = ConnectionRuntime::for_tests_with_pass(
        Arc::new(WorkerFrontDoorServices::new()),
        Arc::clone(&attaches) as Arc<dyn ConnectionNotifier>,
        bearer.map(<[u8]>::to_vec),
        verifier,
    );
    runtime.register_for_tests(PID)?;
    let credential = pass()?;
    let producer = Producer::default();
    producer.produce(&credential, binding(&credential, 40)?)?;
    let authority = Arc::new(RevocationAuthority::new(
        RevocationPolicy {
            log: log()?,
            audience: AUDIENCE.to_owned(),
            issuer: ISSUER,
        },
        Arc::new(producer),
        Arc::new(UnpublishedGrantDecisions),
    ));
    if ready {
        let baseline = GrantFeedEvent::Baseline {
            log: log()?,
            through: Revision::new(40),
            decisions: ["conn", "publish", "subscribe", "root"]
                .into_iter()
                .map(|id| grant(id).map(|grant| (grant, false)))
                .collect::<Result<_, _>>()?,
        };
        assert_eq!(authority.apply(&baseline).applied, Applied::Baseline);
        let ready_event = GrantFeedEvent::Ready {
            log: log()?,
            through: Revision::new(40),
        };
        assert_eq!(authority.apply(&ready_event).applied, Applied::Ready);
    }
    runtime
        .install_revocation(Arc::clone(&authority))
        .map_err(|_| "the authority was already installed")?;
    Ok(Fixture {
        runtime,
        authority,
        attaches,
        credential,
    })
}

fn connect(
    fixture: &Fixture,
    state: &mut ConnectionProcessState,
    token: Vec<u8>,
    max_version: ProtocolVersion,
) -> FrameAction {
    apply_frame(
        PID,
        &fixture.runtime,
        state,
        Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version,
            auth_token: token,
        },
    )
}

/// The `ConnectError` a revocation refusal must be: reason `0x0006` and the
/// stable name as the whole message.
fn refused_with(refusal: RevocationRefusal) -> FrameAction {
    FrameAction::RespondThenClose(Frame::ConnectError {
        flags: 0,
        reason_code: ProtocolError::AUTHENTICATION_FAILURE_CODE,
        message: Some(refusal.name().to_owned()),
    })
}

fn publish(fixture: &Fixture, state: &mut ConnectionProcessState, channel: &str) -> FrameAction {
    apply_frame(
        PID,
        &fixture.runtime,
        state,
        Frame::Publish {
            flags: 0,
            stream_id: 1,
            channel: channel.to_owned(),
            envelope: MessageEnvelope::new(
                SchemaId::new([0x55; 32]),
                CausalContext::independent(),
                vec![1],
            ),
            idempotency_key: None,
        },
    )
}

fn publish_refusal(action: &FrameAction) -> Option<&str> {
    match action {
        FrameAction::Respond(Frame::PublishError { message, .. }) => message.as_deref(),
        _ => None,
    }
}

#[test]
fn a_bound_pass_at_the_close_version_is_admitted_once() -> Result<(), Failure> {
    let fixture = fixture(None, true)?;
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert!(matches!(
        action,
        FrameAction::Respond(Frame::ConnectAck { selected_version, .. })
            if selected_version == REVOCATION_CLOSE_VERSION
    ));
    assert!(state.revocation.is_some());
    assert_eq!(fixture.attaches.0.load(Ordering::SeqCst), 1);
    let diagnostics = fixture.authority.diagnostics()?;
    assert_eq!(diagnostics.bound_connections, 1);
    assert_eq!(diagnostics.counts.admissions_committed, 1);
    Ok(())
}

#[test]
fn a_legacy_only_peer_is_refused_before_any_effect() -> Result<(), Failure> {
    let fixture = fixture(None, true)?;
    let mut state = ConnectionProcessState::default();
    for legacy in [
        ProtocolVersion::new(1, 0),
        ProtocolVersion::new(1, 1),
        ProtocolVersion::new(1, 2),
    ] {
        let action = connect(&fixture, &mut state, fixture.credential.clone(), legacy);
        assert_eq!(action, refused_with(RevocationRefusal::ProtocolUnsupported));
    }
    assert!(!state.authenticated);
    assert!(state.revocation.is_none());
    assert_eq!(fixture.attaches.0.load(Ordering::SeqCst), 0);
    assert_eq!(fixture.authority.diagnostics()?.bound_connections, 0);
    Ok(())
}

#[test]
fn a_bearer_connect_cannot_bypass_the_revocation_policy() -> Result<(), Failure> {
    let fixture = fixture(Some(BEARER), true)?;
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        BEARER.to_vec(),
        REVOCATION_CLOSE_VERSION,
    );
    assert_eq!(action, refused_with(RevocationRefusal::BindingMissing));
    assert!(!state.authenticated);
    Ok(())
}

#[test]
fn a_connect_before_the_stream_is_ready_is_refused_by_name() -> Result<(), Failure> {
    let fixture = fixture(None, false)?;
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert_eq!(action, refused_with(RevocationRefusal::StreamUnavailable));
    assert_eq!(fixture.attaches.0.load(Ordering::SeqCst), 0);
    Ok(())
}

#[test]
fn a_revoked_admitting_grant_refuses_connect_with_no_attach() -> Result<(), Failure> {
    let fixture = fixture(None, true)?;
    let applied = fixture.authority.apply(&revoke(41, "conn")?);
    assert_eq!(applied.applied, Applied::Change);
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert_eq!(action, refused_with(RevocationRefusal::GrantRevoked));
    assert_eq!(fixture.attaches.0.load(Ordering::SeqCst), 0);
    assert_eq!(fixture.authority.diagnostics()?.bound_connections, 0);
    Ok(())
}

#[test]
fn a_refusal_message_is_the_name_and_carries_no_credential_bytes() -> Result<(), Failure> {
    let fixture = fixture(None, false)?;
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    let FrameAction::RespondThenClose(Frame::ConnectError { message, .. }) = action else {
        return Err("the Connect was not refused".into());
    };
    let message = message.ok_or("the refusal named nothing")?;
    assert!(
        RevocationRefusal::ALL
            .iter()
            .any(|refusal| refusal.name() == message)
    );
    assert!(!message.contains(HOLDER));
    assert!(!message.contains(LIVE));
    Ok(())
}

#[test]
fn a_lost_publish_right_refuses_publish_by_name_and_keeps_the_connection() -> Result<(), Failure> {
    let fixture = fixture(None, true)?;
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert!(matches!(
        action,
        FrameAction::Respond(Frame::ConnectAck { .. })
    ));
    let outcome = fixture.authority.apply(&revoke(41, "publish")?);
    assert_eq!(
        outcome.effects,
        vec![RevocationEffect::Withdraw { pid: PID, right: 1 }]
    );
    let refused = publish(&fixture, &mut state, "watcher.orders.eu");
    assert_eq!(publish_refusal(&refused), Some("grant_revoked"));
    let uncovered = publish(&fixture, &mut state, "watcher.billing");
    assert_eq!(
        publish_refusal(&uncovered),
        Some("revocation_binding_missing")
    );
    let grants = state.revocation.as_ref().ok_or("no binding")?;
    assert_eq!(grants.close_pending(), None);
    assert!(grants.permits(HotAction::Subscribe, "watcher.orders.eu"));
    Ok(())
}

#[test]
fn losing_the_admitting_right_closes_and_refuses_every_later_operation() -> Result<(), Failure> {
    let fixture = fixture(None, true)?;
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert!(matches!(
        action,
        FrameAction::Respond(Frame::ConnectAck { .. })
    ));
    // The shared root ancestor: every right derives from it.
    let outcome = fixture.authority.apply(&revoke(41, "root")?);
    assert_eq!(
        outcome.effects,
        vec![RevocationEffect::Close {
            pid: PID,
            reason: RevocationRefusal::GrantRevoked
        }]
    );
    let grants = state.revocation.as_ref().ok_or("no binding")?;
    assert_eq!(
        grants.close_pending(),
        Some(RevocationRefusal::GrantRevoked)
    );
    let refused = publish(&fixture, &mut state, "watcher.orders.eu");
    assert_eq!(publish_refusal(&refused), Some("grant_revoked"));
    // A duplicate of the change adds no close.
    let duplicate = fixture.authority.apply(&revoke(41, "root")?);
    assert_eq!(duplicate.applied, Applied::Duplicate);
    assert!(duplicate.effects.is_empty());
    assert_eq!(fixture.authority.diagnostics()?.counts.closes_decided, 1);
    Ok(())
}

#[test]
fn a_second_connect_never_replaces_the_binding() -> Result<(), Failure> {
    let fixture = fixture(None, true)?;
    let mut state = ConnectionProcessState::default();
    let first = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert!(matches!(
        first,
        FrameAction::Respond(Frame::ConnectAck { .. })
    ));
    let required = state
        .revocation
        .as_ref()
        .map(|grants| grants.binding().required())
        .ok_or("no binding")?;
    let second = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert_eq!(second, FrameAction::Close);
    assert_eq!(
        state
            .revocation
            .as_ref()
            .map(|grants| grants.binding().required()),
        Some(required)
    );
    assert_eq!(fixture.authority.diagnostics()?.bound_connections, 1);
    Ok(())
}

#[test]
fn the_connection_binding_debug_shows_no_holder_or_prefix_bytes() -> Result<(), Failure> {
    let fixture = fixture(None, true)?;
    let mut state = ConnectionProcessState::default();
    let action = connect(
        &fixture,
        &mut state,
        fixture.credential.clone(),
        REVOCATION_CLOSE_VERSION,
    );
    assert!(matches!(
        action,
        FrameAction::Respond(Frame::ConnectAck { .. })
    ));
    let shown = format!("{:?}", state.revocation);
    assert!(!shown.contains(HOLDER));
    assert!(!shown.contains("orders"));
    assert!(!shown.contains("root"));
    assert_eq!(fixture.authority.diagnostics()?.readiness, Readiness::Ready);
    Ok(())
}

#[test]
fn an_open_access_server_under_the_policy_gates_frames_before_connect() -> Result<(), Failure> {
    let runtime = ConnectionRuntime::for_tests(Arc::new(WorkerFrontDoorServices::new()));
    let authority = Arc::new(RevocationAuthority::unpublished(RevocationPolicy {
        log: log()?,
        audience: AUDIENCE.to_owned(),
        issuer: ISSUER,
    }));
    runtime
        .install_revocation(authority)
        .map_err(|_| "the authority was already installed")?;
    let mut state = ConnectionProcessState::default();
    let skipped = apply_frame(
        PID,
        &runtime,
        &mut state,
        Frame::Publish {
            flags: 0,
            stream_id: 1,
            channel: "watcher.orders.eu".to_owned(),
            envelope: MessageEnvelope::new(
                SchemaId::new([0x55; 32]),
                CausalContext::independent(),
                vec![1],
            ),
            idempotency_key: None,
        },
    );
    assert_eq!(skipped, FrameAction::Close);
    let connected = apply_frame(
        PID,
        &runtime,
        &mut state,
        Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: REVOCATION_CLOSE_VERSION,
            auth_token: Vec::new(),
        },
    );
    assert_eq!(connected, refused_with(RevocationRefusal::BindingMissing));
    Ok(())
}