liminal-server 0.25.4

//! ACCESS-004: a worker's registration name is bound to the holder its
//! `Connect` verified. Each pin connects with a registry pass minted here,
//! then registers, through `apply_frame` as a connection process does.

use std::sync::{Arc, Mutex};

use ed25519_dalek::{Signer, SigningKey};
use liminal::protocol::{Frame, ProtocolVersion, WorkerRegisterOutcome, WorkerRegistration};

use super::apply_frame;
use crate::ServerError;
use crate::auth_pass::{PassVerifier, VerifiedHolder, WirePassV1};
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;

const PID: u64 = 7;
const REGISTRY_SEED: [u8; 32] = [42; 32];

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

/// A registration's identity and the holder participant it arrived with.
type HeldRegistration = (String, Option<Vec<u8>>);

/// Every registration the notifier was asked to accept, and the holder each
/// arrived with when the server named one.
#[derive(Debug, Default)]
struct Registrations {
    seen: Mutex<Vec<(u64, String)>>,
    holders: Mutex<Vec<HeldRegistration>>,
}

impl Registrations {
    fn holders(&self) -> Result<Vec<HeldRegistration>, Failure> {
        Ok(self
            .holders
            .lock()
            .map_err(|_| "the holder recorder was poisoned")?
            .clone())
    }

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

impl ConnectionNotifier for Registrations {
    fn on_worker_registered(
        &self,
        pid: u64,
        registration: &WorkerRegistration,
    ) -> Result<(), ServerError> {
        self.seen
            .lock()
            .map_err(|_| ServerError::ConfigValidation {
                message: "the registration recorder was poisoned".to_owned(),
            })?
            .push((pid, registration.identity.clone()));
        Ok(())
    }

    fn on_worker_registered_by(
        &self,
        pid: u64,
        registration: &WorkerRegistration,
        holder: Option<&VerifiedHolder>,
    ) -> Result<(), ServerError> {
        self.holders
            .lock()
            .map_err(|_| ServerError::ConfigValidation {
                message: "the holder recorder was poisoned".to_owned(),
            })?
            .push((
                registration.identity.clone(),
                holder.map(|holder| holder.participant().to_vec()),
            ));
        self.on_worker_registered(pid, registration)
    }

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

/// A pass for `participant`, signed by the registry key the runtime trusts.
fn pass_for(participant: &str) -> Result<Vec<u8>, Failure> {
    let key = SigningKey::from_bytes(&REGISTRY_SEED);
    let mut pass = WirePassV1 {
        participant: participant.as_bytes().to_vec(),
        public_key: key.verifying_key().to_bytes(),
        conversations: Vec::new(),
        live: "watcher.".to_owned(),
        may_enroll: false,
        // A fixed window that holds any clock: the pins read no clock.
        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:?}"))?)
}

const BEARER: &[u8] = b"bearer-secret";

fn verifier() -> Result<PassVerifier, Failure> {
    let registry = SigningKey::from_bytes(&REGISTRY_SEED).verifying_key();
    Ok(PassVerifier::new(registry.as_bytes(), 30)
        .map_err(|error| format!("verifier: {error:?}"))?)
}

/// A runtime trusting the registry key, reporting to `notifier`.
fn runtime(notifier: &Arc<Registrations>) -> Result<ConnectionRuntime, Failure> {
    Ok(ConnectionRuntime::for_tests_with_pass(
        Arc::new(WorkerFrontDoorServices::new()),
        Arc::clone(notifier) as Arc<dyn ConnectionNotifier>,
        None,
        verifier()?,
    ))
}

/// A runtime taking the bearer beside passes, with `auth.require_verified_worker`
/// set as `require` says.
fn bearer_and_pass_runtime(
    notifier: &Arc<Registrations>,
    require: bool,
) -> Result<ConnectionRuntime, Failure> {
    let verifier = if require {
        verifier()?.requiring_verified_worker()
    } else {
        verifier()?
    };
    Ok(ConnectionRuntime::for_tests_with_pass(
        Arc::new(WorkerFrontDoorServices::new()),
        Arc::clone(notifier) as Arc<dyn ConnectionNotifier>,
        Some(BEARER.to_vec()),
        verifier,
    ))
}

fn connect(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    token: Vec<u8>,
) -> Result<(), Failure> {
    let version = ProtocolVersion::new(1, 0);
    let action = apply_frame(
        PID,
        runtime,
        state,
        Frame::Connect {
            flags: 0,
            min_version: version,
            max_version: version,
            auth_token: token,
        },
    );
    match action {
        FrameAction::Respond(Frame::ConnectAck { .. }) => Ok(()),
        other => Err(format!("the Connect was not acknowledged: {other:?}").into()),
    }
}

fn register(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    identity: &str,
) -> Result<WorkerRegisterOutcome, Failure> {
    let registration = WorkerRegistration {
        namespaces: vec!["default".to_owned()],
        task_queue: "findings".to_owned(),
        node: None,
        activity_types: Vec::new(),
        identity: identity.to_owned(),
        activities: Vec::new(),
    };
    match apply_frame(
        PID,
        runtime,
        state,
        Frame::WorkerRegister {
            flags: 0,
            registration,
        },
    ) {
        FrameAction::Respond(Frame::WorkerRegisterAck { outcome, .. }) => Ok(outcome),
        other => Err(format!("the WorkerRegister was not answered: {other:?}").into()),
    }
}

/// R2, red at main: the server takes the registration name on trust.
#[test]
fn a_registration_named_other_than_its_verified_holder_is_refused() -> Result<(), Failure> {
    let notifier = Arc::new(Registrations::default());
    let runtime = runtime(&notifier)?;
    runtime.register_for_tests(PID)?;
    let mut state = ConnectionProcessState::default();
    connect(&runtime, &mut state, pass_for("watcher")?)?;

    let outcome = register(&runtime, &mut state, "archie")?;

    assert_eq!(
        outcome,
        WorkerRegisterOutcome::Rejected {
            reason: "registration identity is not the verified holder".to_owned(),
        },
        "a worker holding a pass for `watcher` registered as `archie`"
    );
    assert_eq!(
        notifier.seen()?,
        Vec::new(),
        "a refused registration must not reach the notifier"
    );
    assert_eq!(
        runtime.stored_registration_for_tests(PID)?,
        None,
        "a refused registration must not be stored on the record, nor emit its topology change"
    );
    Ok(())
}

/// R2's control, green before and after: the holder's own name registers
/// and reaches the notifier once.
#[test]
fn a_registration_named_as_its_verified_holder_is_accepted() -> Result<(), Failure> {
    let notifier = Arc::new(Registrations::default());
    let runtime = runtime(&notifier)?;
    runtime.register_for_tests(PID)?;
    let mut state = ConnectionProcessState::default();
    connect(&runtime, &mut state, pass_for("watcher")?)?;

    let outcome = register(&runtime, &mut state, "watcher")?;

    assert_eq!(outcome, WorkerRegisterOutcome::Accepted);
    assert_eq!(notifier.seen()?, vec![(PID, "watcher".to_owned())]);
    assert_eq!(
        runtime
            .stored_registration_for_tests(PID)?
            .map(|registration| registration.identity),
        Some("watcher".to_owned()),
        "positive control: an accepted registration is stored, so the refused pin's None is seen"
    );
    Ok(())
}

/// R1: a pass `Connect` leaves the connection holding the holder its pass
/// names. A new type, so its red is the build's: at main there is no
/// `verified_holder` to read.
#[test]
fn a_pass_connect_holds_the_holder_its_pass_names() -> Result<(), Failure> {
    let notifier = Arc::new(Registrations::default());
    let runtime = runtime(&notifier)?;
    let mut state = ConnectionProcessState::default();
    connect(&runtime, &mut state, pass_for("watcher")?)?;

    let holder = state
        .verified_holder
        .as_ref()
        .ok_or("a verified pass Connect left no holder")?;
    assert_eq!(holder.participant(), b"watcher");
    assert_eq!(holder.live(), "watcher.");
    Ok(())
}

/// R1: a bearer names no one, so a bearer `Connect` holds no holder, and
/// its registration is taken as today.
#[test]
fn a_bearer_connect_holds_no_holder() -> Result<(), Failure> {
    let runtime = ConnectionRuntime::for_tests_with_auth_token(
        Arc::new(WorkerFrontDoorServices::new()),
        BEARER.to_vec(),
    );
    let mut state = ConnectionProcessState::default();
    connect(&runtime, &mut state, BEARER.to_vec())?;

    assert!(
        state.verified_holder.is_none(),
        "a bearer Connect proved a holder: {:?}",
        state.verified_holder
    );
    assert_eq!(
        register(&runtime, &mut state, "anyone")?,
        WorkerRegisterOutcome::Accepted
    );
    Ok(())
}

/// R3, red at main: with `require_verified_worker` on, a connection that
/// proved no holder (here, the bearer) may not register.
#[test]
fn a_registration_without_a_verified_holder_is_refused_when_required() -> Result<(), Failure> {
    let notifier = Arc::new(Registrations::default());
    let runtime = bearer_and_pass_runtime(&notifier, true)?;
    let mut state = ConnectionProcessState::default();
    connect(&runtime, &mut state, BEARER.to_vec())?;

    assert_eq!(
        register(&runtime, &mut state, "worker-1")?,
        WorkerRegisterOutcome::Rejected {
            reason: "worker registration requires a verified holder".to_owned(),
        },
        "a bearer connection registered although a verified holder is required"
    );
    assert_eq!(notifier.seen()?, Vec::new());
    Ok(())
}

/// R3's controls, green before and after: with the flag off the bearer
/// registers as today, and with it on a pass holder registers as itself.
#[test]
fn the_required_holder_flag_refuses_only_what_proved_no_holder() -> Result<(), Failure> {
    let notifier = Arc::new(Registrations::default());
    let runtime = bearer_and_pass_runtime(&notifier, false)?;
    let mut state = ConnectionProcessState::default();
    connect(&runtime, &mut state, BEARER.to_vec())?;
    assert_eq!(
        register(&runtime, &mut state, "worker-1")?,
        WorkerRegisterOutcome::Accepted
    );

    let runtime = bearer_and_pass_runtime(&notifier, true)?;
    let mut state = ConnectionProcessState::default();
    connect(&runtime, &mut state, pass_for("watcher")?)?;
    assert_eq!(
        register(&runtime, &mut state, "watcher")?,
        WorkerRegisterOutcome::Accepted
    );
    Ok(())
}

/// R4, red at main: the notifier hears a registration with the holder its
/// connection verified, and with none for a bearer.
#[test]
fn a_registration_reaches_the_notifier_with_its_verified_holder() -> Result<(), Failure> {
    let notifier = Arc::new(Registrations::default());
    let runtime = bearer_and_pass_runtime(&notifier, false)?;

    let mut held = ConnectionProcessState::default();
    connect(&runtime, &mut held, pass_for("watcher")?)?;
    register(&runtime, &mut held, "watcher")?;
    let mut bearer = ConnectionProcessState::default();
    connect(&runtime, &mut bearer, BEARER.to_vec())?;
    register(&runtime, &mut bearer, "worker-1")?;

    assert_eq!(
        notifier.holders()?,
        vec![
            ("watcher".to_owned(), Some(b"watcher".to_vec())),
            ("worker-1".to_owned(), None),
        ],
        "each registration must reach the notifier with its connection's holder"
    );
    Ok(())
}