liminal-server 0.25.4

//! Test fixtures for channel membership (ACCESS-006).
//!
//! A seeded world standing where Lys's membership provider stands, a grant
//! authority whose state a test moves, and a verified session naming a holder
//! and its kind.
//!
//! Compiled only for tests and the `test-support` feature. The seeded world
//! decides the way the shared contract describes (a grant reaches its own
//! resource and, unless the placement is restricted, the children placed
//! under it), so the gate's handling is exercised against answers of the
//! contract's shape; it is never installed as a production provider and its
//! answers are never called Lys's.

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

use lys_pass::membership::{
    self, CONTRACT_UNSUPPORTED, CONTRACT_VERSION, GrantLog, MembershipDecision, MembershipRequest,
    OUTSIDE_WORKSPACE, Subject, Verdict, refused_before_lookup,
};
use lys_pass::membership_admission::{self, AdmissionDecision, AdmissionRequest, NO_CURRENT_GRANT};
use lys_pass::membership_pages::{
    self, CURSOR_FOREIGN, MembershipPage, PageOutcome, PageRow, ROW_OVER_BUDGET,
    RecipientPageRequest, ResourcePageRequest, refused_page,
};
use lys_pass::rights::{Mode, Resource};

use super::{AuthorityState, MembershipAuthority, MembershipProvider, ProviderError};
use crate::auth_lys::session::LysSession;
use crate::auth_lys::{
    AccessAction, AccessDecisionError, AccessRefusal, AccessResource, DeliberateDecision,
    GrantReceipt, HotAction, RefusalCause, ResourceKind, SubjectBinding,
};
use crate::config::types::{
    MembershipChannelConfig, MembershipConfig, MembershipConversationConfig,
};
use crate::link::LinkAction;

/// The refusal a decision the grants could not make at the asked revision
/// carries, as Lys names it.
pub const STALE_DECISION: &str = "StaleDecision";
/// The refusal for an action no current grant reaches, as Lys names it.
pub const NOT_HELD: &str = "NotHeld";
/// The refusal for an action only a revoked grant reached, as Lys names it.
pub const REVOKED: &str = "Revoked";

/// A contract subject.
#[must_use]
pub fn subject(id: &str, kind: &str) -> Subject {
    Subject {
        id: id.to_owned(),
        kind: kind.to_owned(),
    }
}

/// A contract resource.
#[must_use]
pub fn resource(kind: &str, id: &str) -> Resource {
    Resource {
        kind: kind.to_owned(),
        id: id.to_owned(),
    }
}

/// A membership configuration over `log`, mapping each
/// `(channel, workspace, resource)` and each `(conversation, channel)`, with
/// the given bounds.
#[must_use]
pub fn config(
    log: &GrantLog,
    kinds: (&str, &str),
    channels: &[(&str, &str, &str)],
    conversations: &[(u64, &str)],
    bounds: (u32, u32, usize),
) -> MembershipConfig {
    let (workspace_kind, channel_kind) = kinds;
    let (page_rows, page_bytes, max_decisions) = bounds;
    MembershipConfig {
        contract: CONTRACT_VERSION,
        log_identity: log.identity.clone(),
        log_epoch: log.epoch,
        workspace_kind: workspace_kind.to_owned(),
        channel_kind: channel_kind.to_owned(),
        credential_path: std::path::PathBuf::from("fixture-credential-never-read"),
        page_rows,
        page_bytes,
        max_decisions,
        connect_workspace: None,
        channels: channels
            .iter()
            .map(|(channel, workspace, placed)| MembershipChannelConfig {
                channel: (*channel).to_owned(),
                workspace: (*workspace).to_owned(),
                resource: (*placed).to_owned(),
            })
            .collect(),
        conversations: conversations
            .iter()
            .map(|(conversation, channel)| MembershipConversationConfig {
                conversation: *conversation,
                channel: (*channel).to_owned(),
            })
            .collect(),
    }
}

/// A verified session naming a holder and, unless `None`, its kind. Its hot,
/// link and deliberate decisions refuse: membership tests reach the gate
/// directly.
#[derive(Debug)]
pub struct FixtureSession {
    holder: String,
    kind: Option<String>,
}

impl FixtureSession {
    /// A session for `holder` of `kind`.
    #[must_use]
    pub fn new(holder: &str, kind: Option<&str>) -> Self {
        Self {
            holder: holder.to_owned(),
            kind: kind.map(str::to_owned),
        }
    }
}

fn session_refusal(id: &str, action: AccessAction) -> AccessDecisionError {
    AccessDecisionError {
        cause: RefusalCause::MissingGrant,
        refusal: AccessRefusal::new(
            AccessResource {
                kind: ResourceKind::Subject,
                id: id.to_owned(),
            },
            action,
            String::new(),
        ),
    }
}

impl LysSession for FixtureSession {
    fn holder(&self) -> &str {
        &self.holder
    }

    fn holder_kind(&self) -> Option<&str> {
        self.kind.as_deref()
    }

    fn decide(
        &self,
        subject: &SubjectBinding,
        action: HotAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        Err(session_refusal(subject.subject(), action.into()))
    }

    fn decide_link(
        &self,
        subject: &SubjectBinding,
        action: LinkAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        Err(session_refusal(subject.subject(), action.access_action()))
    }

    fn open_link(
        &self,
        link: &AccessResource,
        end: LinkAction,
        _now: u64,
    ) -> Result<DeliberateDecision, AccessDecisionError> {
        Err(session_refusal(&link.id, end.access_action()))
    }

    fn decide_live(
        &self,
        resource: &AccessResource,
        action: AccessAction,
        _now: u64,
    ) -> Result<DeliberateDecision, AccessDecisionError> {
        Err(session_refusal(&resource.id, action))
    }
}

/// A grant authority whose state the test sets.
#[derive(Debug)]
pub struct FixtureAuthority {
    state: Mutex<AuthorityState>,
    reads: AtomicU64,
}

impl FixtureAuthority {
    /// An authority `Ready` on `log` at `revision` in `generation`.
    #[must_use]
    pub fn ready(log: &GrantLog, revision: u64, generation: u64) -> Self {
        Self {
            state: Mutex::new(AuthorityState::Ready {
                log: log.clone(),
                revision,
                generation,
            }),
            reads: AtomicU64::new(0),
        }
    }

    /// Moves the authority to `state`.
    ///
    /// # Errors
    /// The state lock was poisoned.
    pub fn set(&self, state: AuthorityState) -> Result<(), &'static str> {
        *self
            .state
            .lock()
            .map_err(|_| "the fixture authority was poisoned")? = state;
        Ok(())
    }

    /// How many times the gate read the state.
    #[must_use]
    pub fn reads(&self) -> u64 {
        self.reads.load(Ordering::Relaxed)
    }
}

impl MembershipAuthority for FixtureAuthority {
    fn state(&self) -> AuthorityState {
        self.reads.fetch_add(1, Ordering::Relaxed);
        self.state.lock().map_or_else(
            |_| AuthorityState::Unavailable {
                cause: "fixture_authority_poisoned".to_owned(),
            },
            |state| state.clone(),
        )
    }
}

/// One seeded grant.
#[derive(Clone, Debug, PartialEq, Eq)]
struct SeededGrant {
    label: String,
    holder: Subject,
    resource: Resource,
    actions: Vec<String>,
    mode: Mode,
    revoked: bool,
}

/// A seeded world answering the shared contract's questions.
#[derive(Debug)]
pub struct SeededProvider {
    log: GrantLog,
    parents: BTreeMap<String, (Resource, bool)>,
    grants: Mutex<Vec<SeededGrant>>,
    revision: AtomicU64,
    asked: AtomicU64,
    unavailable: AtomicBool,
    substitute: AtomicBool,
}

fn poisoned() -> ProviderError {
    ProviderError {
        cause: "fixture_world_poisoned".to_owned(),
    }
}

impl SeededProvider {
    /// An empty world serving `log` at revision 0.
    #[must_use]
    pub fn new(log: &GrantLog) -> Self {
        Self {
            log: log.clone(),
            parents: BTreeMap::new(),
            grants: Mutex::new(Vec::new()),
            revision: AtomicU64::new(0),
            asked: AtomicU64::new(0),
            unavailable: AtomicBool::new(false),
            substitute: AtomicBool::new(false),
        }
    }

    /// Places `child` under `parent`, restricted or not.
    #[must_use]
    pub fn place(mut self, child: Resource, parent: Resource, restricted: bool) -> Self {
        self.parents.insert(child.id, (parent, restricted));
        self
    }

    /// Issues a grant, advancing the world's revision; returns it.
    ///
    /// # Errors
    /// The world lock was poisoned.
    pub fn grant(
        &self,
        label: &str,
        holder: Subject,
        on: Resource,
        actions: &[&str],
        mode: Mode,
    ) -> Result<u64, ProviderError> {
        self.grants
            .lock()
            .map_err(|_| poisoned())?
            .push(SeededGrant {
                label: label.to_owned(),
                holder,
                resource: on,
                actions: actions.iter().map(|action| (*action).to_owned()).collect(),
                mode,
                revoked: false,
            });
        Ok(self.advance())
    }

    /// Revokes the grant `label`, advancing the world's revision; returns it.
    ///
    /// # Errors
    /// The world lock was poisoned.
    pub fn revoke(&self, label: &str) -> Result<u64, ProviderError> {
        for grant in self
            .grants
            .lock()
            .map_err(|_| poisoned())?
            .iter_mut()
            .filter(|grant| grant.label == label)
        {
            grant.revoked = true;
        }
        Ok(self.advance())
    }

    fn advance(&self) -> u64 {
        self.revision
            .fetch_add(1, Ordering::Relaxed)
            .saturating_add(1)
    }

    /// The world's current revision.
    #[must_use]
    pub fn revision(&self) -> u64 {
        self.revision.load(Ordering::Relaxed)
    }

    /// How many questions the world was asked, of every kind.
    #[must_use]
    pub fn asked(&self) -> u64 {
        self.asked.load(Ordering::Relaxed)
    }

    /// Makes every later question fail as if Lys could not be asked.
    pub fn set_unavailable(&self, unavailable: bool) {
        self.unavailable.store(unavailable, Ordering::Relaxed);
    }

    /// Makes every later decision echo a substituted subject.
    pub fn set_substitute(&self, substitute: bool) {
        self.substitute.store(substitute, Ordering::Relaxed);
    }

    fn begin(&self) -> Result<u64, ProviderError> {
        self.asked.fetch_add(1, Ordering::Relaxed);
        if self.unavailable.load(Ordering::Relaxed) {
            return Err(ProviderError {
                cause: "fixture_unavailable".to_owned(),
            });
        }
        Ok(self.revision())
    }

    fn within(&self, child: &Resource, workspace: &Resource) -> bool {
        child == workspace
            || self
                .parents
                .get(&child.id)
                .is_some_and(|(parent, _)| parent == workspace)
    }

    fn verdict(
        &self,
        holder: &Subject,
        on: &Resource,
        action: &str,
    ) -> Result<Verdict, ProviderError> {
        let grants = self.grants.lock().map_err(|_| poisoned())?;
        let parent = self.parents.get(&on.id);
        let mut revoked = false;
        for grant in grants.iter() {
            let reaches = grant.resource == *on
                || parent
                    .is_some_and(|(parent, restricted)| !restricted && *parent == grant.resource);
            if grant.holder != *holder
                || !reaches
                || !grant.actions.iter().any(|held| held == action)
            {
                continue;
            }
            if grant.revoked {
                revoked = true;
                continue;
            }
            return Ok(if grant.mode == Mode::Outright {
                Verdict::Allowed {
                    grant: grant.label.clone(),
                    path: vec![grant.label.clone()],
                    scope: grant.resource.clone(),
                }
            } else {
                Verdict::Held {
                    grant: grant.label.clone(),
                    scope: grant.resource.clone(),
                    mode: grant.mode,
                }
            });
        }
        drop(grants);
        let refusal = if revoked { REVOKED } else { NOT_HELD };
        Ok(Verdict::Refused {
            refusal: refusal.to_owned(),
            reason: "no current grant reaches the action".to_owned(),
        })
    }

    fn holders(&self) -> Result<BTreeSet<(String, String)>, ProviderError> {
        Ok(self
            .grants
            .lock()
            .map_err(|_| poisoned())?
            .iter()
            .map(|grant| (grant.holder.id.clone(), grant.holder.kind.clone()))
            .collect())
    }

    fn row(
        &self,
        holder: &Subject,
        on: &Resource,
        action: &str,
    ) -> Result<Option<PageRow>, ProviderError> {
        Ok(match self.verdict(holder, on, action)? {
            Verdict::Allowed { grant, scope, .. } => Some(PageRow {
                subject: holder.clone(),
                resource: on.clone(),
                grant,
                scope,
                mode: Mode::Outright,
            }),
            Verdict::Held { grant, scope, mode } => Some(PageRow {
                subject: holder.clone(),
                resource: on.clone(),
                grant,
                scope,
                mode,
            }),
            Verdict::Refused { .. } => None,
        })
    }

    /// Pages `candidates` (each a key and its question) after `after`,
    /// within `bounds`, issuing `prefix`-bound continuations.
    fn page(
        &self,
        revision: u64,
        candidates: Vec<(String, Subject, Resource)>,
        action: &str,
        bounds: lys_pass::membership_pages::PageBounds,
        prefix: &str,
        after: Option<&str>,
    ) -> Result<MembershipPage, ProviderError> {
        let after_key = match after {
            None => None,
            Some(token) => match token.strip_prefix(prefix) {
                Some(key) => Some(key.to_owned()),
                None => {
                    return Ok(refused_page(
                        &self.log,
                        CURSOR_FOREIGN,
                        "another question's cursor",
                    ));
                }
            },
        };
        let mut rows = Vec::new();
        let mut skipped = 0_u64;
        let mut bytes = 0_u64;
        let mut last = None;
        let mut more = false;
        let remaining = candidates
            .into_iter()
            .filter(|(key, _, _)| after_key.as_ref().is_none_or(|after| key > after));
        for (key, holder, on) in remaining {
            if rows.len() >= usize::try_from(bounds.rows).unwrap_or(usize::MAX) {
                more = true;
                break;
            }
            let Some(row) = self.row(&holder, &on, action)? else {
                skipped = skipped.saturating_add(1);
                last = Some(key);
                continue;
            };
            let size = serde_json::to_vec(&row)
                .map(|encoded| u64::try_from(encoded.len()).unwrap_or(u64::MAX))
                .map_err(|_| poisoned())?;
            if bytes.saturating_add(size) > u64::from(bounds.bytes) {
                if rows.is_empty() {
                    return Ok(refused_page(
                        &self.log,
                        ROW_OVER_BUDGET,
                        "one row is over budget",
                    ));
                }
                more = true;
                break;
            }
            bytes = bytes.saturating_add(size);
            rows.push(row);
            last = Some(key);
        }
        let returned = u64::try_from(rows.len()).unwrap_or(u64::MAX);
        Ok(MembershipPage {
            contract: CONTRACT_VERSION,
            log: self.log.clone(),
            revision: Some(revision),
            outcome: PageOutcome::Page {
                rows,
                returned,
                skipped,
                complete: !more,
                next: if more {
                    last.map(|key| format!("{prefix}{key}"))
                } else {
                    None
                },
            },
        })
    }
}

impl MembershipProvider for SeededProvider {
    fn decide(&self, request: &MembershipRequest) -> Result<MembershipDecision, ProviderError> {
        let revision = self.begin()?;
        if request.contract != CONTRACT_VERSION {
            return Ok(refused_before_lookup(
                request,
                &self.log,
                CONTRACT_UNSUPPORTED,
                "",
            ));
        }
        if let Err(refused) = membership::validate(request, &self.log) {
            return Ok(refused_before_lookup(
                request,
                &self.log,
                refused.name,
                &refused.reason,
            ));
        }
        if !self.within(&request.resource, &request.workspace) {
            return Ok(refused_before_lookup(
                request,
                &self.log,
                OUTSIDE_WORKSPACE,
                "",
            ));
        }
        if request.at_least > revision {
            return Ok(refused_before_lookup(
                request,
                &self.log,
                STALE_DECISION,
                "",
            ));
        }
        let verdict = self.verdict(&request.subject, &request.resource, &request.action)?;
        let mut answered = request.clone();
        if self.substitute.load(Ordering::Relaxed) {
            answered.subject.id = format!("{}-substituted", answered.subject.id);
        }
        Ok(MembershipDecision {
            contract: CONTRACT_VERSION,
            log: self.log.clone(),
            request: answered,
            revision: Some(revision),
            verdict,
        })
    }

    fn admission(&self, request: &AdmissionRequest) -> Result<AdmissionDecision, ProviderError> {
        let revision = self.begin()?;
        if let Err(refused) = membership_admission::validate(request, &self.log) {
            return Ok(membership_admission::refused_before_lookup(
                request,
                &self.log,
                refused.name,
                &refused.reason,
            ));
        }
        let held = self
            .grants
            .lock()
            .map_err(|_| poisoned())?
            .iter()
            .find(|grant| {
                !grant.revoked
                    && grant.holder == request.subject
                    && self.within(&grant.resource, &request.workspace)
            })
            .cloned();
        let verdict = held.map_or_else(
            || Verdict::Refused {
                refusal: NO_CURRENT_GRANT.to_owned(),
                reason: "no current grant within the workspace".to_owned(),
            },
            |grant| Verdict::Allowed {
                path: vec![grant.label.clone()],
                grant: grant.label,
                scope: grant.resource,
            },
        );
        Ok(AdmissionDecision {
            contract: CONTRACT_VERSION,
            log: self.log.clone(),
            request: request.clone(),
            revision: Some(revision),
            verdict,
        })
    }

    fn resources(&self, request: &ResourcePageRequest) -> Result<MembershipPage, ProviderError> {
        let revision = self.begin()?;
        if let Err(refused) = membership_pages::validate_resources(request, &self.log) {
            return Ok(refused_page(&self.log, refused.name, &refused.reason));
        }
        if request.at_least > revision {
            return Ok(refused_page(&self.log, STALE_DECISION, ""));
        }
        let candidates = self
            .parents
            .iter()
            .filter(|(_, (parent, _))| *parent == request.workspace)
            .map(|(id, _)| {
                (
                    id.clone(),
                    request.subject.clone(),
                    resource(&request.kind, id),
                )
            })
            .collect();
        let prefix = format!(
            "resources:{}:{}:{}:{}:",
            request.workspace.id, request.subject.id, request.subject.kind, request.action
        );
        self.page(
            revision,
            candidates,
            &request.action,
            request.bounds,
            &prefix,
            request.after.as_deref(),
        )
    }

    fn recipients(&self, request: &RecipientPageRequest) -> Result<MembershipPage, ProviderError> {
        let revision = self.begin()?;
        if let Err(refused) = membership_pages::validate_recipients(request, &self.log) {
            return Ok(refused_page(&self.log, refused.name, &refused.reason));
        }
        if request.at_least > revision {
            return Ok(refused_page(&self.log, STALE_DECISION, ""));
        }
        let candidates = self
            .holders()?
            .into_iter()
            .map(|(id, kind)| {
                (
                    format!("{id}/{kind}"),
                    subject(&id, &kind),
                    request.resource.clone(),
                )
            })
            .collect();
        let prefix = format!(
            "recipients:{}:{}:{}:",
            request.resource.kind, request.resource.id, request.action
        );
        self.page(
            revision,
            candidates,
            &request.action,
            request.bounds,
            &prefix,
            request.after.as_deref(),
        )
    }
}