liminal-server 0.25.4

//! ACCESS-001 R2 at every hot wire route: a connection admitted on a Lys pass
//! record has each `Call` and `Serve` route decided from that pass, offline,
//! before the route's effect, and refused by name on the route's own error
//! frame; the routes without a refusal frame (`PushReply`, the backpressure
//! signals) refuse by having no effect, the pusher reading the refusal by
//! type. The routes' classes are the ones `classify_wire_route` pins.

use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use ed25519_dalek::SigningKey;
use liminal::protocol::{
    CausalContext, Frame, MessageEnvelope, MessageId, ProtocolVersion, SchemaId,
    WorkerRegisterOutcome, WorkerRegistration,
};
use liminal_protocol::credential::WireCredential;

use liminal_protocol::wire::{
    ClientRequest, Generation, ObserverRecoveryHandshake, ObserverRefusal, RecordAdmission,
    RecordAdmissionAttemptToken, ServerValue,
};

use super::{apply_frame, participant_access_refusal};
use crate::ServerError;
use crate::auth_lys::session::ConsumerAdmitter;
use crate::auth_lys::{
    AccessAction, AccessDecisionError, AccessRefusal, AccessResource, DeliberateDecision,
    GrantReceipt, HotAction, LysDeliberateConsumer, LysPassConsumer, RefusalCause, ResourceKind,
    SubjectBinding,
};
use crate::auth_pass::{PassVerifier, VerifiedHolder};
use crate::link::{LinkAction, LinkHost, LinkRights};
use crate::link_handshake::{LinkHandshake, LocalMachine, NOISE_OVERHEAD};
use crate::server::connection::conversation::{ConnectionConversation, ConversationResource};
use crate::server::connection::notifier::ConnectionNotifier;
use crate::server::connection::services::{
    ConnectionServices, ConnectionSubscription, PublishOutcome, SubscriptionResource,
};
use crate::server::connection::state::{ConnectionProcessState, FrameAction};
use crate::server::connection::supervisor::runtime::ConnectionRuntime;
use crate::server::participant::{
    ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantSemanticError,
    ParticipantSemanticHandler,
};
use liminal::protocol::{LinkEnd, LinkOffer, LinkOpenOutcome};
use lys_core::keys::Ed25519Identity;
use zeroize::Zeroizing;

const PID: u64 = 11;
const STREAM: u32 = 3;
const CONVERSATION: u64 = 9;
const SUBSCRIPTION: u64 = 5;
const ISSUED: u64 = 7;
const REGISTRY_SEED: [u8; 32] = [42; 32];

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

/// A consumer standing where `lys-pass` will: one fixture pass, and a grant
/// table the test moves between routes so each decision is shown to be its own.
#[derive(Debug, Default)]
struct Grants {
    held: Mutex<Vec<(String, HotAction)>>,
    /// Link grants by subject (ACCESS-002 R1), decided offline like the hot
    /// table.
    links: Mutex<Vec<(String, LinkAction)>>,
    /// The link ends Lys answers outright, live; every other open is refused.
    opens: Mutex<Vec<(String, AccessAction)>>,
    /// When the fixture pass was issued, in Unix seconds, when it says.
    issued_at: Option<u64>,
}

impl Grants {
    fn grant_link(&self, subject: &str, action: LinkAction) -> Result<(), Failure> {
        self.links
            .lock()
            .map_err(|_| "the link table was poisoned")?
            .push((subject.to_owned(), action));
        Ok(())
    }

    fn allow_open(&self, link: &str, end: AccessAction) -> Result<(), Failure> {
        self.opens
            .lock()
            .map_err(|_| "the open table was poisoned")?
            .push((link.to_owned(), end));
        Ok(())
    }

    fn grant(&self, subject: &str, action: HotAction) -> Result<(), Failure> {
        self.held
            .lock()
            .map_err(|_| "the grant table was poisoned")?
            .push((subject.to_owned(), action));
        Ok(())
    }

    fn revoke(&self, subject: &str, action: HotAction) -> Result<(), Failure> {
        self.held
            .lock()
            .map_err(|_| "the grant table was poisoned")?
            .retain(|(held, held_action)| !(held == subject && *held_action == action));
        Ok(())
    }
}

impl LysPassConsumer for Grants {
    type Pass = ();
    type Holder = str;
    type AdmissionError = std::io::Error;

    fn verify_offline(
        &self,
        bytes: &[u8],
        audience: &str,
        _now: u64,
    ) -> Result<Self::Pass, Self::AdmissionError> {
        if bytes != b"fixture-pass" || audience != "liminal" {
            return Err(std::io::Error::other("signature"));
        }
        Ok(())
    }

    fn verified_holder<'p>(&self, pass: &'p Self::Pass) -> &'p Self::Holder {
        let () = pass;
        "fixture-holder"
    }

    fn verified_issued_at(&self, pass: &Self::Pass) -> Option<u64> {
        let () = pass;
        self.issued_at
    }

    fn decide_hot(
        &self,
        pass: &Self::Pass,
        subject: &SubjectBinding,
        action: HotAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        let () = pass;
        let held = self.held.lock().is_ok_and(|held| {
            held.iter()
                .any(|(id, held_action)| id == &subject.resource().id && *held_action == action)
        });
        if held {
            return Ok(GrantReceipt {
                grant: format!("grant-{}-{action:?}", subject.resource().id),
            });
        }
        Err(AccessDecisionError {
            cause: RefusalCause::MissingGrant,
            refusal: AccessRefusal::new(
                AccessResource {
                    kind: ResourceKind::Subject,
                    id: subject.resource().id.clone(),
                },
                action.into(),
                format!("https://lys.example/ask/{}", subject.subject()),
            ),
        })
    }

    fn decide_link(
        &self,
        pass: &Self::Pass,
        subject: &SubjectBinding,
        action: LinkAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        let () = pass;
        let held = self.links.lock().is_ok_and(|held| {
            held.iter()
                .any(|(id, held_action)| id == &subject.resource().id && *held_action == action)
        });
        if held {
            return Ok(GrantReceipt {
                grant: format!("link-{}-{action:?}", subject.resource().id),
            });
        }
        Err(AccessDecisionError {
            cause: RefusalCause::MissingGrant,
            refusal: AccessRefusal::new(
                AccessResource {
                    kind: ResourceKind::Link,
                    id: subject.resource().id.clone(),
                },
                action.access_action(),
                format!("https://lys.example/ask/{}", subject.subject()),
            ),
        })
    }
}

impl LysDeliberateConsumer for Grants {
    fn decide_live(
        &self,
        pass: &Self::Pass,
        resource: &AccessResource,
        action: AccessAction,
        _now: u64,
    ) -> Result<DeliberateDecision, AccessDecisionError> {
        let () = pass;
        let allowed = self.opens.lock().is_ok_and(|opens| {
            opens
                .iter()
                .any(|(id, end)| id == &resource.id && *end == action)
        });
        if allowed {
            return Ok(DeliberateDecision::Outright(GrantReceipt {
                grant: format!("open-{}-{action:?}", resource.id),
            }));
        }
        Err(AccessDecisionError {
            cause: RefusalCause::MissingGrant,
            refusal: AccessRefusal::new(
                resource.clone(),
                action,
                format!("https://lys.example/ask/{}", resource.id),
            ),
        })
    }
}

/// A conversation that answers nothing: the routes under test stop before it.
#[derive(Debug)]
struct Quiet;

impl ConversationResource for Quiet {
    fn message(&self, _envelope: &MessageEnvelope, _op_id: Option<u64>) -> Result<(), ServerError> {
        Ok(())
    }

    fn participant_pids(&self) -> Vec<u64> {
        Vec::new()
    }

    fn has_detected_crash(&self) -> bool {
        false
    }

    fn await_crash(&self, _timeout: Duration) -> Option<Instant> {
        None
    }

    fn receive_reply(&self, _timeout: Duration) -> Result<MessageEnvelope, ServerError> {
        Err(ServerError::ConfigValidation {
            message: "the quiet conversation never replies".to_owned(),
        })
    }

    fn close(self: Box<Self>) -> Result<(), ServerError> {
        Ok(())
    }

    fn finalize(self: Box<Self>) {}
}

/// Services recording every effect that reached them: a refused route leaves
/// no trace here.
#[derive(Debug, Default)]
struct Door {
    opens: Mutex<Vec<(u64, String)>>,
    messages: Mutex<usize>,
}

impl Door {
    fn opens(&self) -> Result<Vec<(u64, String)>, Failure> {
        Ok(self
            .opens
            .lock()
            .map_err(|_| "the open recorder was poisoned")?
            .clone())
    }

    fn messages(&self) -> Result<usize, Failure> {
        Ok(*self
            .messages
            .lock()
            .map_err(|_| "the message recorder was poisoned")?)
    }
}

fn unreached(route: &str) -> ServerError {
    ServerError::ConfigValidation {
        message: format!("{route} is not a route under test"),
    }
}

impl ConnectionServices for Door {
    fn publish(
        &self,
        _connection: u64,
        _channel: &str,
        _envelope: &MessageEnvelope,
        _idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        Err(unreached("publish"))
    }

    fn subscribe(
        &self,
        _channel: &str,
        _accepted_schemas: &[SchemaId],
        _install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        Err(unreached("subscribe"))
    }

    fn unsubscribe(&self, _subscription: ConnectionSubscription) -> Result<(), ServerError> {
        Ok(())
    }

    fn open_conversation(
        &self,
        conversation_id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        self.opens
            .lock()
            .map_err(|_| unreached("a poisoned open recorder"))?
            .push((conversation_id, subject.to_owned()));
        Ok(ConnectionConversation::new(
            Box::new(Quiet),
            subject.to_owned(),
        ))
    }

    fn conversation_message(
        &self,
        _conversation: &ConnectionConversation,
        _envelope: &MessageEnvelope,
        _op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        *self
            .messages
            .lock()
            .map_err(|_| unreached("a poisoned message recorder"))? += 1;
        Ok(())
    }

    fn close_conversation(&self, _conversation: ConnectionConversation) -> Result<(), ServerError> {
        Ok(())
    }

    fn flush_durable_state(&self) -> Result<(), ServerError> {
        Ok(())
    }
}

#[derive(Debug, Default)]
struct Silent;

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

    fn on_worker_registered_by(
        &self,
        _pid: u64,
        _registration: &WorkerRegistration,
        _holder: Option<&VerifiedHolder>,
    ) -> Result<(), ServerError> {
        Ok(())
    }

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

/// A connected process whose pass is `grants`, over `door`.
fn admitted(
    grants: &Arc<Grants>,
    door: &Arc<Door>,
) -> Result<(ConnectionRuntime, ConnectionProcessState), Failure> {
    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::clone(door) as Arc<dyn ConnectionServices>,
        Arc::new(Silent) as Arc<dyn ConnectionNotifier>,
        None,
        verifier,
    );
    runtime
        .install_lys_admitter(
            Arc::new(ConsumerAdmitter::new(Arc::clone(grants))),
            "network-a".to_owned(),
        )
        .map_err(|_| "a lys admitter was already installed")?;
    runtime.register_for_tests(PID)?;
    let mut state = ConnectionProcessState::default();
    let version = ProtocolVersion::new(1, 0);
    let action = apply_frame(
        PID,
        &runtime,
        &mut state,
        Frame::Connect {
            flags: 0,
            min_version: version,
            max_version: version,
            auth_token: WireCredential::lys_record(b"fixture-pass"),
        },
    );
    match action {
        FrameAction::Respond(Frame::ConnectAck { .. }) => Ok((runtime, state)),
        other => Err(format!("the Lys record was not admitted: {other:?}").into()),
    }
}

fn open(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    subject: &str,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::ConversationOpen {
            flags: 0,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            subject: subject.to_owned(),
        },
    )
}

fn message(runtime: &ConnectionRuntime, state: &mut ConnectionProcessState) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::ConversationMessage {
            flags: 0,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            envelope: MessageEnvelope::new(
                SchemaId::new([1; SchemaId::WIRE_LEN]),
                CausalContext::independent(),
                b"hello".to_vec(),
            ),
        },
    )
}

/// The refusal a conversation route answered on its own error frame, naming
/// the subject and the action it lacked.
fn conversation_refusal(action: FrameAction) -> Result<String, Failure> {
    match action {
        FrameAction::Respond(Frame::ConversationError {
            conversation_id: CONVERSATION,
            stream_id: STREAM,
            message: Some(message),
            ..
        }) => Ok(message),
        other => Err(format!("the route was not refused on its conversation: {other:?}").into()),
    }
}

#[path = "apply_lys_routes_tests/hot.rs"]
mod hot;
#[path = "apply_lys_routes_tests/links.rs"]
mod links;
#[path = "apply_lys_routes_tests/open_mode.rs"]
mod open_mode;
#[path = "apply_lys_routes_tests/services.rs"]
mod services;