gunnar-grpc 3.0.0

//! **SCIM 2.0 and the CSV import, on Gunnar's identity table.** The protocol
//! and the import rules are edda's (`nornir-identity-directory`, A34 M-6: one
//! copy for every product); this module is only the seam between them and
//! [`SsoPlane`]: a [`PlaneDirectory`] is one account's directory for one call
//! (its room, its actor, its clock), and [`run_import`] is the body of the
//! import job the server runs on its jera lane (A34-17).

use std::sync::{Arc, LazyLock, Weak};

pub use nornir_identity_directory::import::{ImportOp, ImportPlan, Imported};
pub use nornir_identity_directory::scim::bearer::{self, Bearer, Refusal, RefusalGuard, Scope};
pub use nornir_identity_directory::scim::{
    error_answer, group_resource, handle, user_resource, Answer, Call, Effect, CONTENT_TYPE,
    ERROR_SCHEMA, GROUP_SCHEMA, LIST_SCHEMA, MAX_BODY, MAX_PAGE, PATCH_SCHEMA, USER_SCHEMA,
};

use nornir_identity_directory::model::{
    DirectoryStore, Group, NewUser, StoreError, User, UserPatch as DirPatch,
};

use super::plane::{DirUser, GroupRow, ProvisionAsk, Room, SsoPlane, UserPatch};
use crate::seam::ControlError;

/// **The flood guard for refused SCIM bearers** (A34-27), one per process:
/// an unknown, revoked or expired bearer is audited a few times a minute per
/// client, then held back, then answered 429.
pub static REFUSALS: LazyLock<RefusalGuard> = LazyLock::new(RefusalGuard::new);

fn to_store(err: ControlError) -> StoreError {
    match err {
        ControlError::NotFound(m) => StoreError::NotFound(m),
        ControlError::AlreadyExists(m) => StoreError::AlreadyExists(m),
        ControlError::InvalidArgument(m) => StoreError::Invalid(m),
        ControlError::FailedPrecondition(m) => StoreError::Precondition(m),
        ControlError::PermissionDenied(m) => StoreError::Denied(m),
        ControlError::ResourceExhausted(m) => StoreError::Exhausted(m),
        other => StoreError::Internal(other.to_string()),
    }
}

fn user_of(row: DirUser) -> User {
    let u = row.user;
    User {
        principal_id: u.principal_id,
        user_name: u.user_name,
        display_name: u.display_name,
        email: u.email,
        external_id: u.external_id,
        active: u.active,
        created_at_unix_ms: u.created_at_unix_ms,
        deleted: row.deleted,
    }
}

fn patch_of(p: &DirPatch) -> UserPatch {
    UserPatch {
        user_name: p.user_name.clone(),
        display_name: p.display_name.clone(),
        email: p.email.clone(),
        external_id: p.external_id.clone(),
        active: p.active,
    }
}

/// **One account's directory for one call**, over the plane.
pub struct PlaneDirectory<'a> {
    pub plane: &'a dyn SsoPlane,
    pub account: &'a str,
    /// The room a NEW employee holds.
    pub room: Room,
    /// Who the audit trail names: `scim:<account>`, `csv:<admin>`.
    pub actor: &'a str,
    pub now_unix_ms: i64,
}

impl PlaneDirectory<'_> {
    fn ask(&self, user: &NewUser) -> ProvisionAsk {
        ProvisionAsk {
            account: self.account.to_owned(),
            user_name: user.user_name.clone(),
            display_name: user.display_name.clone(),
            email: user.email.clone(),
            external_id: user.external_id.clone(),
            active: user.active,
            source: user.source.as_str(),
            room: self.room.clone(),
            actor: self.actor.to_owned(),
            now_unix_ms: self.now_unix_ms,
        }
    }
}

impl DirectoryStore for PlaneDirectory<'_> {
    fn users(&self) -> Result<Vec<User>, StoreError> {
        Ok(self
            .plane
            .directory(self.account)
            .map_err(to_store)?
            .into_iter()
            .map(user_of)
            .collect())
    }

    fn provision(&self, user: &NewUser) -> Result<User, StoreError> {
        self.plane
            .provision(self.ask(user))
            .map(user_of)
            .map_err(to_store)
    }

    fn update_user(
        &self,
        principal_id: &str,
        patch: &DirPatch,
    ) -> Result<(User, bool), StoreError> {
        self.plane
            .update_user(self.account, principal_id, patch_of(patch), self.actor)
            .map(|(u, changed)| (user_of(u), changed))
            .map_err(to_store)
    }

    fn delete_user(&self, principal_id: &str) -> Result<User, StoreError> {
        self.plane
            .delete_user(self.account, principal_id, self.actor)
            .map(user_of)
            .map_err(to_store)
    }

    fn groups(&self) -> Result<Vec<Group>, StoreError> {
        Ok(self
            .plane
            .groups(self.account)
            .map_err(to_store)?
            .into_iter()
            .map(|g| Group {
                id: g.id,
                display_name: g.display_name,
                external_id: g.external_id,
                members: g.members,
                created_at_unix_ms: g.created_at_unix_ms,
            })
            .collect())
    }

    fn put_group(&self, g: Group) -> Result<Group, StoreError> {
        let saved = self
            .plane
            .put_group(GroupRow {
                id: g.id,
                account: self.account.to_owned(),
                display_name: g.display_name,
                external_id: g.external_id,
                members: g.members,
                created_at_unix_ms: g.created_at_unix_ms,
            })
            .map_err(to_store)?;
        Ok(Group {
            id: saved.id,
            display_name: saved.display_name,
            external_id: saved.external_id,
            members: saved.members,
            created_at_unix_ms: saved.created_at_unix_ms,
        })
    }

    fn remove_group(&self, id: &str) -> Result<bool, StoreError> {
        self.plane.remove_group(self.account, id).map_err(to_store)
    }

    fn new_group_id(&self) -> String {
        crate::webauthn::hex(&crate::webauthn::random_bytes(12))
    }

    fn now_unix_ms(&self) -> i64 {
        self.now_unix_ms
    }

    fn apply_import(&self, plan: &ImportPlan, commit: bool) -> Result<(), StoreError> {
        let rows: Vec<(String, ImportRow)> = plan
            .ops
            .iter()
            .map(|op| match op {
                ImportOp::Create { line, user } => (
                    format!("line {line}: {}", user.user_name),
                    ImportRow::Create(self.ask(user)),
                ),
                ImportOp::Update {
                    line,
                    principal_id,
                    user_name,
                    patch,
                    ..
                } => (
                    format!("line {line}: {user_name}"),
                    ImportRow::Update {
                        principal_id: principal_id.clone(),
                        patch: patch_of(patch),
                    },
                ),
            })
            .collect();
        self.plane
            .import(self.account, &rows, self.actor, self.now_unix_ms, commit)
            .map_err(to_store)
    }
}

/// One row of an import as the plane applies it (all rows in one write).
#[derive(Debug, Clone)]
pub enum ImportRow {
    Create(ProvisionAsk),
    Update {
        principal_id: String,
        patch: UserPatch,
    },
}

/// What an import job calls once when it ends (the audit line).
pub type ImportFinished = Arc<dyn Fn(&Result<Imported, String>) + Send + Sync>;

/// **What an import job is asked to do** (`Sso.ImportUsers`): the server
/// runs it on its jera lane, one per account at a time, and the console
/// polls `Sso.ImportStatus`.
#[derive(Clone)]
pub struct ImportOrder {
    pub account: String,
    pub csv: Arc<str>,
    pub dry_run: bool,
    pub room: Room,
    pub actor: String,
    /// The plane the job writes through. Weak: the plane holds the queue,
    /// and a finished job is kept a while.
    pub plane: Weak<dyn crate::seam::ControlPlane>,
    /// Called once when the job ends (the audit line).
    pub finished: ImportFinished,
}

impl std::fmt::Debug for ImportOrder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ImportOrder")
            .field("account", &self.account)
            .field("bytes", &self.csv.len())
            .field("dry_run", &self.dry_run)
            .finish_non_exhaustive()
    }
}

/// Where an import job is, as `Sso.ImportUsers`/`Sso.ImportStatus` answer.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ImportJob {
    pub job_id: String,
    /// `queued`, `running`, `done` or `failed`.
    pub state: &'static str,
    pub error: String,
    pub done_rows: u64,
    pub total_rows: u64,
    pub outcome: Imported,
}

/// **The import job's body**: plan the CSV against the account's directory
/// and apply it in one write (A34-17), naming the people it deactivated
/// (A34-9). `progress` is `(done, total)` rows.
pub fn run_import(
    order: &ImportOrder,
    now_unix_ms: i64,
    progress: &dyn Fn(u64, u64),
) -> Result<Imported, String> {
    let control = order
        .plane
        .upgrade()
        .ok_or_else(|| "the server is shutting down; nothing was imported".to_owned())?;
    let plane = control.sso().ok_or_else(|| {
        "single sign-on needs an identity table; this control plane has none".to_owned()
    })?;
    let store = PlaneDirectory {
        plane,
        account: &order.account,
        room: order.room.clone(),
        actor: &order.actor,
        now_unix_ms,
    };
    nornir_identity_directory::import::run(&store, &order.csv, order.dry_run, progress)
}