arcly_http_identity/

identity.rs

//! [`IdentityService`] — the orchestrator that ties stores, hashing, JWT
//! minting, lockout, and MFA together into the register / login / refresh /
//! logout flows. Provide it via DI and call it from your controllers (or mount
//! the ready-made ones behind the `controllers` feature).

use std::sync::Arc;

use arcly_http_core::auth::JwtService;
use rand::RngCore;

use crate::consent::ConsentStore;
use crate::error::{IdentityError, Result};
use crate::federation::{
    FederatedIdentity, FederatedIdentityStore, FederatedOutcome, FederatedProfile, FederationAudit,
    FederationDecision, FederationEvent, LinkingConfig,
};
use crate::lockout::LockoutPolicy;
use crate::mfa::MfaVerifier;
use crate::model::{AccountStatus, Identity, LoginOutcome, RegisterRequest, TokenResponse};
use crate::password::{BreachChecker, PasswordHasher, PasswordPolicy};
use crate::risk::{RiskEngine, RiskSignal, RiskVerdict};
use crate::session::{ChallengeStore, RefreshStore};
use crate::store::{CredentialStore, UserStore};

/// Behavioural knobs for the identity flows (B2C ↔ B2B).
#[derive(Debug, Clone)]
pub struct IdentityConfig {
    /// New accounts start `Pending` and cannot log in until their email is
    /// verified. B2C consumer default: `true`.
    pub require_email_verification: bool,
    /// Embed the identity's `tenant` into the access token so `TENANT` guard
    /// can cross-check it. B2B default: `true`.
    pub bind_tenant: bool,
    /// Registration is closed — only pre-provisioned / invited users exist.
    /// B2B org default: often `true`.
    pub invite_only: bool,
    /// TTL for a pending MFA challenge (seconds).
    pub mfa_challenge_ttl_secs: u64,
}

impl Default for IdentityConfig {
    fn default() -> Self {
        Self {
            require_email_verification: false,
            bind_tenant: true,
            invite_only: false,
            mfa_challenge_ttl_secs: 300,
        }
    }
}

/// Composes every collaborator needed for the core auth flows. Cheap to clone
/// (all fields are `Arc`), so it can be `provide`d and injected freely.
#[derive(Clone)]
pub struct IdentityService {
    users: Arc<dyn UserStore>,
    creds: Arc<dyn CredentialStore>,
    hasher: Arc<dyn PasswordHasher>,
    jwt: Arc<JwtService>,
    refresh: Arc<dyn RefreshStore>,
    challenges: Arc<dyn ChallengeStore>,
    lockout: Option<Arc<LockoutPolicy>>,
    mfa: Option<Arc<dyn MfaVerifier>>,
    federated: Option<Arc<dyn FederatedIdentityStore>>,
    risk: Option<Arc<dyn RiskEngine>>,
    fed_audit: Option<Arc<dyn FederationAudit>>,
    linking: LinkingConfig,
    password_policy: PasswordPolicy,
    breach: Option<Arc<dyn BreachChecker>>,
    consent: Option<Arc<dyn ConsentStore>>,
    required_consents: Vec<String>,
    config: IdentityConfig,
}

/// Builder — the constructor has too many collaborators for a positional `new`.
pub struct IdentityServiceBuilder {
    users: Arc<dyn UserStore>,
    creds: Arc<dyn CredentialStore>,
    hasher: Arc<dyn PasswordHasher>,
    jwt: Arc<JwtService>,
    refresh: Arc<dyn RefreshStore>,
    challenges: Arc<dyn ChallengeStore>,
    lockout: Option<Arc<LockoutPolicy>>,
    mfa: Option<Arc<dyn MfaVerifier>>,
    federated: Option<Arc<dyn FederatedIdentityStore>>,
    risk: Option<Arc<dyn RiskEngine>>,
    fed_audit: Option<Arc<dyn FederationAudit>>,
    linking: LinkingConfig,
    password_policy: PasswordPolicy,
    breach: Option<Arc<dyn BreachChecker>>,
    consent: Option<Arc<dyn ConsentStore>>,
    required_consents: Vec<String>,
    config: IdentityConfig,
}

impl IdentityService {
    /// Start building. Required collaborators are passed here; optional ones
    /// (lockout, MFA) and config are attached fluently.
    pub fn builder(
        users: Arc<dyn UserStore>,
        creds: Arc<dyn CredentialStore>,
        hasher: Arc<dyn PasswordHasher>,
        jwt: Arc<JwtService>,
        refresh: Arc<dyn RefreshStore>,
        challenges: Arc<dyn ChallengeStore>,
    ) -> IdentityServiceBuilder {
        IdentityServiceBuilder {
            users,
            creds,
            hasher,
            jwt,
            refresh,
            challenges,
            lockout: None,
            mfa: None,
            federated: None,
            risk: None,
            fed_audit: None,
            linking: LinkingConfig::default(),
            password_policy: PasswordPolicy::default(),
            breach: None,
            consent: None,
            required_consents: Vec::new(),
            config: IdentityConfig::default(),
        }
    }

    /// Register a new user. Hashes the password (if any) off the reactor, stores
    /// the identity, and returns it. Does **not** issue tokens — callers usually
    /// send a verification email first.
    pub async fn register(&self, req: RegisterRequest) -> Result<Identity> {
        if self.config.invite_only {
            return Err(IdentityError::Forbidden);
        }
        // Reject duplicates early (the store also enforces this atomically).
        if self
            .users
            .find_by_email(req.tenant.as_deref(), &req.email)
            .await?
            .is_some()
        {
            return Err(IdentityError::AlreadyExists);
        }

        let status = if self.config.require_email_verification {
            AccountStatus::Pending
        } else {
            AccountStatus::Active
        };
        let identity = Identity {
            id: new_id(),
            tenant: req.tenant.clone(),
            email: Some(req.email.clone()),
            email_verified: false,
            phone: None,
            phone_verified: false,
            status,
            roles: vec!["user".to_owned()],
            perms: Vec::new(),
            mfa: Default::default(),
            attributes: req.attributes.clone(),
        };
        // Password strength + breach checks (H4) run *before* the user is
        // created, and their reasons are safe to surface (registration, not login).
        if let Some(password) = &req.password {
            self.password_policy.validate(password)?;
            if let Some(checker) = &self.breach {
                if checker.is_breached(password).await? {
                    return Err(IdentityError::PasswordRejected(
                        "this password has appeared in a known data breach".into(),
                    ));
                }
            }
        }

        self.users.insert(&identity).await?;

        if let Some(password) = req.password {
            let phc = self.hasher.hash(password).await?;
            self.creds.set_password(&identity.id, &phc).await?;
        }
        Ok(identity)
    }

    /// Authenticate with email + password. Returns either issued tokens or an
    /// MFA challenge. `principal_ip` seeds lockout before the user is known.
    pub async fn login(
        &self,
        tenant: Option<&str>,
        email: &str,
        password: &str,
    ) -> Result<LoginOutcome> {
        self.login_bound(tenant, email, password, None).await
    }

    /// Like [`login`](Self::login) but **sender-constrains** the issued access
    /// token to a DPoP key (`cnf_jkt` from a verified proof, RFC 9449 / H3). The
    /// binding is applied on the direct-auth path; when MFA is enrolled the
    /// challenge path mints in `verify_mfa` (unbound for now).
    pub async fn login_bound(
        &self,
        tenant: Option<&str>,
        email: &str,
        password: &str,
        cnf_jkt: Option<&str>,
    ) -> Result<LoginOutcome> {
        // Throttle before the user is even known, keyed on tenant+email, so
        // unknown-account probing is rate-limited and cannot enumerate.
        if let Some(lock) = &self.lockout {
            let principal = format!("{}::{}", tenant.unwrap_or("-"), email.to_ascii_lowercase());
            lock.check(&principal).await?;
        }

        let user = self.users.find_by_email(tenant, email).await?;

        // Uniform failure: run through the same shape whether or not the user
        // exists to avoid timing / status-code enumeration.
        let Some(user) = user else {
            return Err(IdentityError::InvalidCredentials);
        };

        if !user.status.can_authenticate() {
            // Pending / suspended / locked / deleted all present as 401.
            return Err(IdentityError::InvalidCredentials);
        }

        let phc = self
            .creds
            .get_password(&user.id)
            .await?
            .ok_or(IdentityError::InvalidCredentials)?;

        let ok = self.hasher.verify(password.to_owned(), phc.clone()).await?;
        if !ok {
            return Err(IdentityError::InvalidCredentials);
        }

        // Upgrade a legacy (bcrypt) hash transparently on successful login.
        if self.hasher.needs_rehash(&phc) {
            if let Ok(new_phc) = self.hasher.hash(password.to_owned()).await {
                let _ = self.creds.set_password(&user.id, &new_phc).await;
            }
        }

        if user.mfa.is_enrolled() {
            let challenge_id = new_id();
            self.challenges
                .put(&challenge_id, &user.id, self.config.mfa_challenge_ttl_secs)
                .await?;
            return Ok(LoginOutcome::MfaChallenge {
                mfa_required: true,
                challenge_id,
                methods: self.enrolled_methods(&user),
            });
        }

        Ok(LoginOutcome::Authenticated(
            self.mint_bound(&user, cnf_jkt).await?,
        ))
    }

    /// Complete a pending MFA challenge with a factor response (`method` = e.g.
    /// `"totp"` / `"recovery"`, `code` = the user-supplied value).
    pub async fn verify_mfa(
        &self,
        challenge_id: &str,
        method: &str,
        code: &str,
    ) -> Result<TokenResponse> {
        let challenge = self
            .challenges
            .take(challenge_id)
            .await?
            .ok_or(IdentityError::InvalidToken)?;

        let verifier = self.mfa.as_ref().ok_or(IdentityError::MfaFailed)?;
        let ok = verifier.verify(&challenge.user_id, method, code).await?;
        if !ok {
            return Err(IdentityError::MfaFailed);
        }
        let user = self
            .users
            .find_by_id(&challenge.user_id)
            .await?
            .ok_or(IdentityError::NotFound)?;
        self.mint(&user).await
    }

    /// Authenticate a user from an **external identity provider** (social login /
    /// enterprise SSO), applying the takeover-safe linking policy, optional risk
    /// step-up, and auditing — then minting through the *same* path as password
    /// login. `signal` (when a `RiskEngine` is configured) drives adaptive MFA.
    ///
    /// See [`crate::federation`] for the security rationale behind each branch.
    pub async fn login_federated(
        &self,
        tenant: Option<&str>,
        profile: &FederatedProfile,
        signal: Option<RiskSignal>,
    ) -> Result<FederatedOutcome> {
        let fed = self
            .federated
            .as_ref()
            .ok_or_else(|| IdentityError::Internal("federated store not configured".into()))?;

        // 1. Known external identity → straight to (risk/MFA →) mint.
        if let Some(uid) = fed
            .find_user(&profile.provider, &profile.provider_id)
            .await?
        {
            let user = self
                .users
                .find_by_id(&uid)
                .await?
                .ok_or(IdentityError::NotFound)?;
            self.audit_fed(
                profile,
                Some(&user.id),
                tenant,
                FederationDecision::ReturningLogin,
                &signal,
            )
            .await;
            return self.finish_federated(user, signal).await;
        }

        // 2. First sighting of this external identity — decide the local account.
        let (user, decision) = match self.resolve_account(tenant, profile).await? {
            Resolution::LinkRequired(email) => {
                self.audit_fed(
                    profile,
                    None,
                    tenant,
                    FederationDecision::LinkRequired,
                    &signal,
                )
                .await;
                return Ok(FederatedOutcome::LinkRequired { email });
            }
            Resolution::Denied => {
                self.audit_fed(profile, None, tenant, FederationDecision::Denied, &signal)
                    .await;
                return Err(IdentityError::Forbidden);
            }
            Resolution::Existing(user) => (user, FederationDecision::AutoLinked),
            Resolution::Provisioned(user) => (user, FederationDecision::Provisioned),
        };

        // 3. Persist the link and continue.
        fed.link(FederatedIdentity {
            provider: profile.provider.clone(),
            provider_id: profile.provider_id.clone(),
            user_id: user.id.clone(),
            linked_at: JwtService::unix_now(),
        })
        .await?;
        self.audit_fed(profile, Some(&user.id), tenant, decision, &signal)
            .await;
        self.finish_federated(user, signal).await
    }

    /// Explicitly link an external identity to an **already-authenticated** local
    /// user (`user_id`). This is the takeover-proof path the app runs after the
    /// user proves ownership of the local account, in response to a
    /// [`FederatedOutcome::LinkRequired`].
    pub async fn link_federated(&self, user_id: &str, profile: &FederatedProfile) -> Result<()> {
        let fed = self
            .federated
            .as_ref()
            .ok_or_else(|| IdentityError::Internal("federated store not configured".into()))?;
        // The external identity must not already belong to someone else.
        if let Some(existing) = fed
            .find_user(&profile.provider, &profile.provider_id)
            .await?
        {
            if existing != user_id {
                return Err(IdentityError::Forbidden);
            }
            return Ok(());
        }
        fed.link(FederatedIdentity {
            provider: profile.provider.clone(),
            provider_id: profile.provider_id.clone(),
            user_id: user_id.to_owned(),
            linked_at: JwtService::unix_now(),
        })
        .await
    }

    /// Decide which local account a first-time external identity maps to.
    async fn resolve_account(
        &self,
        tenant: Option<&str>,
        profile: &FederatedProfile,
    ) -> Result<Resolution> {
        // If a local account already owns this email, we must never create a
        // duplicate and must be careful about linking.
        if let Some(email) = profile.email.as_deref() {
            if let Some(existing) = self.users.find_by_email(tenant, email).await? {
                // The *only* case where silent linking is safe: the provider
                // verified the email AND the local account's email is verified
                // AND policy opts in. Everything else (either side unverified,
                // or opt-in disabled) is held for an explicit, authenticated
                // linking step — this is the pre-hijacking defence.
                let both_verified = profile.email_verified && existing.email_verified;
                return Ok(if both_verified && self.linking.auto_link_verified_email {
                    Resolution::Existing(existing)
                } else {
                    Resolution::LinkRequired(email.to_owned())
                });
            }
        }
        // No email collision → JIT provision a brand-new account.
        if self.linking.jit_provisioning && !self.config.invite_only {
            Ok(Resolution::Provisioned(
                self.provision_from_profile(tenant, profile).await?,
            ))
        } else {
            Ok(Resolution::Denied)
        }
    }

    /// Create a brand-new local identity from an external profile (no password).
    async fn provision_from_profile(
        &self,
        tenant: Option<&str>,
        profile: &FederatedProfile,
    ) -> Result<Identity> {
        let identity = Identity {
            id: new_id(),
            tenant: tenant.map(str::to_owned),
            email: profile.email.clone(),
            // Trust the provider's verified flag; unverified stays unverified so
            // downstream flows can require confirmation.
            email_verified: profile.email_verified,
            phone: None,
            phone_verified: false,
            status: AccountStatus::Active,
            roles: vec!["user".to_owned()],
            perms: Vec::new(),
            mfa: Default::default(),
            attributes: profile.attributes.clone(),
        };
        self.users.insert(&identity).await?;
        Ok(identity)
    }

    /// Shared tail for federated logins: apply risk step-up and existing MFA
    /// enrolment, else mint.
    async fn finish_federated(
        &self,
        user: Identity,
        signal: Option<RiskSignal>,
    ) -> Result<FederatedOutcome> {
        if let (Some(engine), Some(sig)) = (&self.risk, signal.as_ref()) {
            match engine.assess(sig).await {
                RiskVerdict::Deny => return Err(IdentityError::Forbidden),
                // StepUp is honoured by the enrolment check below: if a factor
                // exists we challenge; if none exists we can't step up, so the
                // login proceeds (the audit trail records the risk decision).
                RiskVerdict::StepUp | RiskVerdict::Allow => {}
            }
        }
        // Challenge whenever the account has a second factor enrolled.
        if user.mfa.is_enrolled() {
            let challenge_id = new_id();
            self.challenges
                .put(&challenge_id, &user.id, self.config.mfa_challenge_ttl_secs)
                .await?;
            return Ok(FederatedOutcome::MfaChallenge {
                mfa_required: true,
                challenge_id,
                methods: self.enrolled_methods(&user),
            });
        }
        Ok(FederatedOutcome::Authenticated(self.mint(&user).await?))
    }

    async fn audit_fed(
        &self,
        profile: &FederatedProfile,
        user_id: Option<&str>,
        tenant: Option<&str>,
        decision: FederationDecision,
        signal: &Option<RiskSignal>,
    ) {
        if let Some(sink) = &self.fed_audit {
            sink.record(FederationEvent {
                provider: profile.provider.clone(),
                provider_id: profile.provider_id.clone(),
                user_id: user_id.map(str::to_owned),
                email: profile.email.clone(),
                tenant: tenant.map(str::to_owned),
                decision,
                ip: signal.as_ref().and_then(|s| s.ip.clone()),
                trace_id: signal.as_ref().and_then(|s| s.trace_id.clone()),
            })
            .await;
        }
    }

    /// Rotate a refresh token: validate signature + stored `jti`, atomically
    /// consume the old `jti`, and issue a fresh pair. Reuse of a consumed token
    /// fails closed.
    pub async fn refresh(&self, refresh_token: &str) -> Result<TokenResponse> {
        let (sub, jti) = self
            .jwt
            .validate_refresh(refresh_token)
            .ok_or(IdentityError::InvalidToken)?;

        // Atomic single-use: if this returns None the token was already rotated
        // (possible replay) — reject and revoke the whole chain.
        match self.refresh.take(&jti).await? {
            Some(uid) if uid == sub => {}
            Some(_) => return Err(IdentityError::InvalidToken),
            None => {
                self.refresh.revoke_all(&sub).await.ok();
                return Err(IdentityError::InvalidToken);
            }
        }

        let user = self
            .users
            .find_by_id(&sub)
            .await?
            .ok_or(IdentityError::NotFound)?;
        self.mint(&user).await
    }

    /// Revoke a specific refresh token (single-device logout).
    pub async fn logout(&self, refresh_token: &str) -> Result<()> {
        if let Some((_, jti)) = self.jwt.validate_refresh(refresh_token) {
            self.refresh.revoke(&jti).await?;
        }
        Ok(())
    }

    /// Revoke every refresh token for a user ("log out everywhere").
    pub async fn logout_all(&self, user_id: &str) -> Result<()> {
        self.refresh.revoke_all(user_id).await
    }

    /// Fetch the current identity by id (for `/auth/me`).
    pub async fn me(&self, user_id: &str) -> Result<Identity> {
        self.users
            .find_by_id(user_id)
            .await?
            .ok_or(IdentityError::NotFound)
    }

    // ── internals ──────────────────────────────────────────────────────────

    /// Mint an access+refresh pair for an authenticated identity and persist the
    /// refresh `jti` for rotation.
    pub async fn mint(&self, user: &Identity) -> Result<TokenResponse> {
        self.mint_bound(user, None).await
    }

    /// Like [`mint`](Self::mint) but **sender-constrains** the access token to a
    /// DPoP key thumbprint (RFC 9449, H3). `cnf_jkt` comes from a verified
    /// [`DpopProof`](crate::dpop::DpopProof); the resource server then requires a
    /// matching proof on every request, so a leaked token can't be replayed.
    pub async fn mint_bound(
        &self,
        user: &Identity,
        cnf_jkt: Option<&str>,
    ) -> Result<TokenResponse> {
        // Consent gate (H5): the single funnel every login path passes through,
        // so an unconsented subject can never be issued tokens regardless of the
        // authentication method (password / social / SSO / passwordless).
        if !self.required_consents.is_empty() {
            if let Some(consent) = &self.consent {
                let mut missing = Vec::new();
                for purpose in &self.required_consents {
                    if !consent.has_consent(&user.id, purpose).await? {
                        missing.push(purpose.clone());
                    }
                }
                if !missing.is_empty() {
                    return Err(IdentityError::ConsentRequired(missing));
                }
            }
        }

        let tenant = if self.config.bind_tenant {
            user.tenant.as_deref()
        } else {
            None
        };
        let email = user.email.clone().unwrap_or_default();
        let access = self
            .jwt
            .issue_access_bound_cnf(
                &user.id,
                user.primary_role(),
                &email,
                &user.perms,
                tenant,
                cnf_jkt,
            )
            .map_err(|e| IdentityError::Internal(e.to_string()))?;
        let (refresh, jti) = self
            .jwt
            .issue_refresh(&user.id)
            .map_err(|e| IdentityError::Internal(e.to_string()))?;
        self.refresh
            .save(&jti, &user.id, self.jwt.refresh_ttl_secs())
            .await?;
        Ok(TokenResponse::bearer(
            access,
            refresh,
            self.jwt.access_ttl_secs(),
        ))
    }

    fn enrolled_methods(&self, user: &Identity) -> Vec<String> {
        let mut m = Vec::new();
        if user.mfa.totp_enrolled {
            m.push("totp".to_owned());
        }
        if user.mfa.passkeys > 0 {
            m.push("passkey".to_owned());
        }
        if user.mfa.recovery_codes_remaining > 0 {
            m.push("recovery".to_owned());
        }
        m
    }
}

impl IdentityServiceBuilder {
    pub fn lockout(mut self, lockout: Arc<LockoutPolicy>) -> Self {
        self.lockout = Some(lockout);
        self
    }
    pub fn mfa(mut self, verifier: Arc<dyn MfaVerifier>) -> Self {
        self.mfa = Some(verifier);
        self
    }
    /// Enable federated login by supplying the identity-link store.
    pub fn federated(mut self, store: Arc<dyn FederatedIdentityStore>) -> Self {
        self.federated = Some(store);
        self
    }
    /// Attach a risk engine for adaptive step-up on (federated or password) login.
    pub fn risk(mut self, engine: Arc<dyn RiskEngine>) -> Self {
        self.risk = Some(engine);
        self
    }
    /// Attach an audit sink for federated-login decisions.
    pub fn federation_audit(mut self, sink: Arc<dyn FederationAudit>) -> Self {
        self.fed_audit = Some(sink);
        self
    }
    /// Override the account-linking / provisioning policy (default is the
    /// enterprise-safe posture).
    pub fn linking(mut self, linking: LinkingConfig) -> Self {
        self.linking = linking;
        self
    }
    /// Enforce a password-strength policy at registration (H4).
    pub fn password_policy(mut self, policy: PasswordPolicy) -> Self {
        self.password_policy = policy;
        self
    }
    /// Reject known-breached passwords via an app-supplied checker (HIBP).
    pub fn breach_checker(mut self, checker: Arc<dyn BreachChecker>) -> Self {
        self.breach = Some(checker);
        self
    }
    /// Require the listed consent purposes before any token is minted (H5). The
    /// gate lives at the single `mint` funnel, so it applies to every login path.
    pub fn require_consents(mut self, store: Arc<dyn ConsentStore>, purposes: Vec<String>) -> Self {
        self.consent = Some(store);
        self.required_consents = purposes;
        self
    }
    pub fn config(mut self, config: IdentityConfig) -> Self {
        self.config = config;
        self
    }
    pub fn build(self) -> IdentityService {
        IdentityService {
            users: self.users,
            creds: self.creds,
            hasher: self.hasher,
            jwt: self.jwt,
            refresh: self.refresh,
            challenges: self.challenges,
            lockout: self.lockout,
            mfa: self.mfa,
            federated: self.federated,
            risk: self.risk,
            fed_audit: self.fed_audit,
            linking: self.linking,
            password_policy: self.password_policy,
            breach: self.breach,
            consent: self.consent,
            required_consents: self.required_consents,
            config: self.config,
        }
    }
}

/// How a first-time external identity resolved to a local account.
enum Resolution {
    /// Silent-link into this existing (both-verified) account.
    Existing(Identity),
    /// A new local account was provisioned (JIT).
    Provisioned(Identity),
    /// A verified-email collision needs an explicit, authenticated linking step.
    LinkRequired(String),
    /// Not allowed (invite-only, or unverified email with JIT disabled).
    Denied,
}

/// Collision-resistant opaque identifier (user ids, challenge ids). 128 bits of
/// randomness, URL-safe base64.
pub fn new_id() -> String {
    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
    use base64::Engine;
    let mut b = [0u8; 16];
    rand::thread_rng().fill_bytes(&mut b);
    URL_SAFE_NO_PAD.encode(b)
}