liminal-server 0.24.0

use super::*;
use std::io::{Read, Write};

use crate::auth_revocation::{
    Applied, AuthenticatedBinding, BoundRightClaim, Digest32, GrantBindingSource, GrantFeedChange,
    GrantFeedEvent, GrantId, GrantLogId, Revision, RevocationAuthority, RevocationEffect,
    RevocationPolicy, RevocationRefusal, RightAction, UnpublishedGrantDecisions,
};
use crate::server::connection::loopback::{LoopbackClientEnd, LoopbackDuplex};
use crate::server::connection::supervisor::ConnectionSupervisor;
use crate::server::connection::supervisor::builder::LysInstall;
use crate::server::connection::supervisor::handle::PushReplyFailure;
use crate::server::connection::supervisor::push_types::PushBinding;
use crate::services::lifecycle::ProcessGrant;
use liminal::protocol::{
    CONVERSATION_REPLY_REQUESTED_FLAG, FrameHeader, PUSH_REPLY_REFUSED_FLAG,
    REVOCATION_CLOSE_VERSION, WorkerRefusal, WorkerRefusalCode, decode, encode, encoded_len,
};

fn registration(task_queue: &str, namespaces: &[&str]) -> WorkerRegistration {
    WorkerRegistration {
        namespaces: namespaces
            .iter()
            .map(|namespace| (*namespace).to_owned())
            .collect(),
        task_queue: task_queue.to_owned(),
        node: None,
        activity_types: Vec::new(),
        identity: "fixture-holder".to_owned(),
        activities: Vec::new(),
    }
}

fn register(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    registration: &WorkerRegistration,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::WorkerRegister {
            flags: 0,
            registration: registration.clone(),
        },
    )
}

fn rejected(frame: Frame) -> Result<String, Failure> {
    match frame {
        Frame::WorkerRegisterAck {
            outcome: WorkerRegisterOutcome::Rejected { reason },
            ..
        } => Ok(reason),
        _ => Err("the registration was not rejected".into()),
    }
}

fn response(action: FrameAction) -> Result<Frame, Failure> {
    match action {
        FrameAction::Respond(frame) => Ok(frame),
        _ => Err("the registration supplied no response".into()),
    }
}

fn accepted(frame: &Frame) -> Result<(), Failure> {
    if matches!(
        frame,
        Frame::WorkerRegisterAck {
            outcome: WorkerRegisterOutcome::Accepted,
            ..
        }
    ) {
        Ok(())
    } else {
        Err("the registration was not accepted".into())
    }
}

fn missing_serve(reason: &str, subject: &str) -> Result<(), Failure> {
    let refusal: AccessRefusal = serde_json::from_str(reason)?;
    assert_eq!(
        refusal.refused.resource,
        format!("liminal.subject:{subject}")
    );
    assert_eq!(refusal.refused.action, "serve");
    assert_eq!(refusal.refused.needed.id, subject);
    assert_eq!(refusal.refused.needed.action, "serve");
    Ok(())
}

#[test]
fn an_empty_namespace_registration_requires_only_serve_on_its_task_queue() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let registration = registration("findings", &[]);
    missing_serve(
        &rejected(response(register(&runtime, &mut state, &registration))?)?,
        "findings",
    )?;
    assert_eq!(runtime.stored_registration_for_tests(PID)?, None);
    grants.grant("findings", HotAction::Serve)?;
    accepted(&response(register(&runtime, &mut state, &registration))?)?;
    assert_eq!(
        runtime.stored_registration_for_tests(PID)?,
        Some(registration)
    );
    Ok(())
}

#[test]
fn a_task_queue_outside_the_service_process_scope_is_refused_before_any_grant()
-> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    state.service_process = Some(ProcessGrant {
        service: "orders-service".to_owned(),
        subjects: vec!["orders".to_owned()],
    });
    let registration = registration("findings", &["orders"]);
    let expected =
        "service_process_scope: a service process serves its own subjects and nothing else";
    assert_eq!(
        rejected(response(register(&runtime, &mut state, &registration))?)?,
        expected
    );
    assert_eq!(runtime.stored_registration_for_tests(PID)?, None);
    grants.grant("orders", HotAction::Serve)?;
    grants.grant("findings", HotAction::Serve)?;
    assert_eq!(
        rejected(response(register(&runtime, &mut state, &registration))?)?,
        expected
    );
    assert_eq!(runtime.serving("findings"), None);
    Ok(())
}

fn write(client: &mut LoopbackClientEnd, frame: &Frame) -> Result<(), Failure> {
    let mut bytes = vec![0; encoded_len(frame)?];
    let written = encode(frame, &mut bytes)?;
    client.write_all(bytes.get(..written).ok_or("invalid encoded length")?)?;
    Ok(())
}

fn read(client: &mut LoopbackClientEnd) -> Result<Frame, Failure> {
    let mut header = [0; FrameHeader::WIRE_LEN];
    client.read_exact(&mut header)?;
    let length = u32::from_be_bytes(
        header
            .get(6..10)
            .ok_or("missing frame length")?
            .try_into()?,
    );
    let length = usize::try_from(length)?;
    if length > 64 * 1024 {
        return Err("the fixture frame exceeded its ring capacity".into());
    }
    let total = FrameHeader::WIRE_LEN
        .checked_add(length)
        .ok_or("frame length overflow")?;
    let mut bytes = header.to_vec();
    bytes.resize(total, 0);
    client.read_exact(
        bytes
            .get_mut(FrameHeader::WIRE_LEN..)
            .ok_or("missing frame body")?,
    )?;
    let (frame, consumed) = decode(&bytes)?;
    assert_eq!(consumed, total);
    Ok(frame)
}

fn connect(client: &mut LoopbackClientEnd) -> Result<(), Failure> {
    let version = ProtocolVersion::new(1, 0);
    write(
        client,
        &Frame::Connect {
            flags: 0,
            min_version: version,
            max_version: version,
            auth_token: WireCredential::lys_record(b"fixture-pass"),
        },
    )?;
    if matches!(read(client)?, Frame::ConnectAck { .. }) {
        Ok(())
    } else {
        Err("the fixture connection was not admitted".into())
    }
}

fn request(client: &mut LoopbackClientEnd, subject: &str) -> Result<(), Failure> {
    write(
        client,
        &Frame::ConversationOpen {
            flags: 0,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            subject: subject.to_owned(),
        },
    )?;
    write(
        client,
        &Frame::ConversationMessage {
            flags: CONVERSATION_REPLY_REQUESTED_FLAG,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            envelope: MessageEnvelope::new(
                SchemaId::new([1; SchemaId::WIRE_LEN]),
                CausalContext::independent(),
                b"request".to_vec(),
            ),
        },
    )
}

struct Route(ConnectionSupervisor);

impl Drop for Route {
    fn drop(&mut self) {
        self.0.shutdown();
    }
}

#[test]
fn a_routed_push_binds_the_admitted_task_queue_and_relays_refusal_words() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let route = Route(
        ConnectionSupervisor::builder(door)
            .notifier(Arc::new(Silent))
            .lys_for_tests(LysInstall {
                admitter: Arc::new(ConsumerAdmitter::new(Arc::clone(&grants))),
                network_id: "network-a".to_owned(),
                link_host: None,
                services: None,
                membership: None,
            })
            .build()?,
    );
    let runtime = route.0.runtime_for_test();
    let (mut worker, server) = LoopbackDuplex::bounded(64 * 1024);
    let handle = route.0.spawn_loopback_connection(server)?;
    connect(&mut worker)?;
    grants.grant("refusal", HotAction::Serve)?;
    let registration = registration("refusal.request", &["refusal"]);
    let frame = Frame::WorkerRegister {
        flags: 0,
        registration: registration.clone(),
    };
    write(&mut worker, &frame)?;
    missing_serve(&rejected(read(&mut worker)?)?, &registration.task_queue)?;
    assert_eq!(runtime.stored_registration_for_tests(handle.pid())?, None);
    grants.grant(&registration.task_queue, HotAction::Serve)?;
    write(&mut worker, &frame)?;
    accepted(&read(&mut worker)?)?;
    assert_eq!(
        runtime.stored_registration_for_tests(handle.pid())?,
        Some(registration.clone())
    );

    let (mut caller, server) = LoopbackDuplex::bounded(64 * 1024);
    route.0.spawn_loopback_connection(server)?;
    connect(&mut caller)?;
    grants.grant(&registration.task_queue, HotAction::Call)?;
    request(&mut caller, &registration.task_queue)?;
    let Frame::Push {
        correlation_id,
        payload,
        ..
    } = read(&mut worker)?
    else {
        return Err("the registered worker received no routed push".into());
    };
    assert_eq!(payload, b"request");
    let PushBinding::Subject(bound) = runtime.push_binding(handle.pid(), correlation_id) else {
        return Err("the routed push retained no subject binding".into());
    };
    assert_eq!(bound.as_bytes(), registration.task_queue.as_bytes());
    let words = "queue full\nPlease try again.";
    write(
        &mut worker,
        &Frame::PushReply {
            flags: PUSH_REPLY_REFUSED_FLAG,
            stream_id: STREAM,
            correlation_id,
            payload: WorkerRefusal::new("busy", words)?.canonical_bytes()?,
        },
    )?;
    let Frame::ConversationError {
        reason_code,
        message,
        conversation_id,
        stream_id,
        ..
    } = read(&mut caller)?
    else {
        return Err("the caller received no worker refusal".into());
    };
    assert_eq!(reason_code, WorkerRefusalCode::Busy.reason_code());
    assert_eq!(message.as_deref(), Some(words));
    assert_eq!(conversation_id, CONVERSATION);
    assert_eq!(stream_id, STREAM);
    assert_eq!(runtime.pending_push_count_for(handle.pid()), 0);
    Ok(())
}

struct Binding(AuthenticatedBinding);

impl std::fmt::Debug for Binding {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("Binding")
    }
}

impl GrantBindingSource for Binding {
    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> {
        if self.pass_digest(credential)? != self.0.pass_digest {
            return Err(RevocationRefusal::BindingMissing);
        }
        Ok(self.0.clone())
    }
}

fn authority(subject: &str) -> Result<Arc<RevocationAuthority>, Failure> {
    let log = GrantLogId::new("worker-grants", 0)?;
    let issuer = Digest32([7; 32]);
    let binding = AuthenticatedBinding {
        issuer,
        audience: "liminal".to_owned(),
        holder: b"fixture-holder".to_vec(),
        pass_digest: Digest32(
            *blake3::hash(&WireCredential::lys_record(b"fixture-pass")).as_bytes(),
        ),
        log: log.clone(),
        required: Revision::new(40),
        rights: vec![
            BoundRightClaim {
                action: RightAction::Connect,
                prefix: String::new(),
                chains: vec![vec![GrantId::new("connect")?]],
            },
            BoundRightClaim {
                action: RightAction::Hot(HotAction::Serve),
                prefix: subject.to_owned(),
                chains: vec![vec![GrantId::new("serve")?]],
            },
        ],
    };
    let authority = Arc::new(RevocationAuthority::new(
        RevocationPolicy {
            log: log.clone(),
            audience: "liminal".to_owned(),
            issuer,
        },
        Arc::new(Binding(binding)),
        Arc::new(UnpublishedGrantDecisions),
    ));
    assert_eq!(
        authority
            .apply(&GrantFeedEvent::Baseline {
                log: log.clone(),
                through: Revision::new(40),
                decisions: vec![
                    (GrantId::new("connect")?, false),
                    (GrantId::new("serve")?, false)
                ],
            })
            .applied,
        Applied::Baseline
    );
    assert_eq!(
        authority
            .apply(&GrantFeedEvent::Ready {
                log,
                through: Revision::new(40)
            })
            .applied,
        Applied::Ready
    );
    Ok(authority)
}

#[test]
fn an_admitted_workers_next_reply_names_the_applied_revocation() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, _) = admitted(&grants, &door)?;
    let registration = registration("refusal.request", &[]);
    let authority = authority(&registration.task_queue)?;
    runtime
        .install_revocation(Arc::clone(&authority))
        .map_err(|_| "authority already installed")?;
    let mut state = ConnectionProcessState::default();
    assert!(matches!(
        apply_frame(
            PID,
            &runtime,
            &mut state,
            Frame::Connect {
                flags: 0,
                min_version: REVOCATION_CLOSE_VERSION,
                max_version: REVOCATION_CLOSE_VERSION,
                auth_token: WireCredential::lys_record(b"fixture-pass"),
            }
        ),
        FrameAction::Respond(Frame::ConnectAck { .. })
    ));
    missing_serve(
        &rejected(response(register(&runtime, &mut state, &registration))?)?,
        &registration.task_queue,
    )?;
    grants.grant(&registration.task_queue, HotAction::Serve)?;
    accepted(&response(register(&runtime, &mut state, &registration))?)?;
    let admitted_reply = runtime.push_for_tests(PID, Some(&registration.task_queue))?;
    assert!(matches!(
        reply(&runtime, &mut state, admitted_reply.correlation_id()),
        FrameAction::NoResponse
    ));
    assert!(
        matches!(admitted_reply.try_receive(), Some(Ok(payload)) if payload.as_slice() == b"answer")
    );

    let awaiting = runtime.push_for_tests(PID, Some(&registration.task_queue))?;
    let outcome = authority.apply(&GrantFeedEvent::Change {
        log: GrantLogId::new("worker-grants", 0)?,
        revision: Revision::new(41),
        change: GrantFeedChange::Revoke(GrantId::new("serve")?),
        evidence: Digest32([41; 32]),
    });
    assert_eq!(outcome.applied, Applied::Change);
    assert_eq!(
        outcome.effects,
        vec![RevocationEffect::Withdraw { pid: PID, right: 1 }]
    );
    assert!(
        state
            .revocation
            .as_ref()
            .ok_or("no bound rights")?
            .close_pending()
            .is_none()
    );
    grants.revoke(&registration.task_queue, HotAction::Serve)?;
    assert!(matches!(
        reply(&runtime, &mut state, awaiting.correlation_id()),
        FrameAction::NoResponse
    ));
    let Some(Err(PushReplyFailure::Server(ServerError::PushReplyRefused {
        correlation_id,
        reason,
    }))) = awaiting.try_receive()
    else {
        return Err("the next reply was not refused by the applied revocation".into());
    };
    assert_eq!(correlation_id, awaiting.correlation_id());
    assert_eq!(reason, "grant_revoked");
    assert_eq!(runtime.pending_push_count_for(PID), 0);
    Ok(())
}