polyc-state-connect 2026.8.3

//! What the transport boundary checks before a call reaches a module.
//!
//! Four checks, and none of them is a State decision: version negotiation,
//! audience, the listener's own lifecycle, and whatever the transport's
//! deadline leaves. Each is a pure function over the kernel's vocabulary that
//! returns the kernel's own typed outcome, so a handler calls them and
//! branches on nothing itself (INV-24).
//!
//! Everything past this point — identity, digest, precondition, fence,
//! bounds, and the commit itself — belongs to the module.
//!
//! # Addressing and authorization
//!
//! [`check_audience`] establishes only that a caller reached the right surface.
//! That was enough while this listener mounted no authority method: there was
//! nothing behind it to authorize. It is not enough now. A method that commits
//! needs the caller's *proven* identity bound to what it may ask for, which is
//! what [`AudienceBinding`] and [`check_audience_binding`] add: a deny-by-default
//! map from a verified mutual-TLS peer to the audiences it may act for. A
//! correctly-addressed call from a workload the binding does not name is
//! denied.

use std::{
    collections::{BTreeMap, BTreeSet},
    fmt,
    time::Duration,
};

use polyc_state::{
    error::StateError,
    id::{Audience, OperationFamily},
};

use crate::wire::CallContextVersion;

/// The audience the State plane's authority surfaces serve, and the one every
/// call to them declares.
///
/// One definition for both halves of the check, and the only place the name is
/// written. A served audience and a declared audience that disagree are refused
/// as [`StateError::Denied`], which is fail-closed but silent until something
/// dials — so the way to keep them agreeing is to leave neither side a value of
/// its own to get wrong. [`state_audience`] is what a caller and a listener
/// both build from, and no constructor in this crate takes an audience for
/// these surfaces.
pub const STATE_AUDIENCE: &str = "polychrome.state";

/// Returns the audience [`STATE_AUDIENCE`] names.
#[must_use]
pub fn state_audience() -> Audience {
    Audience::new(STATE_AUDIENCE)
}

/// Refuses a peer speaking a protocol version this build does not.
///
/// Polychrome ships one lockstep-versioned product: an incompatible peer
/// refuses the operation rather than guessing at a shape it does not know.
/// The module makes this same check on the command it receives; the boundary
/// makes it first, so a mismatched peer never reaches a module at all.
///
/// # Errors
///
/// Returns [`StateError::Malformed`] naming `protocol_version`, which is
/// terminal: a retry at the same version changes nothing.
pub fn check_call_context_version(declared: CallContextVersion) -> Result<(), StateError> {
    if declared == CallContextVersion::CURRENT {
        return Ok(());
    }
    Err(StateError::Malformed {
        field: "protocol_version".to_owned(),
        reason: format!("this listener speaks {}", CallContextVersion::CURRENT),
    })
}

/// Refuses a caller asking on behalf of an audience this surface does not
/// serve.
///
/// This is an addressing check, not an authorization one. It establishes that
/// the caller is talking to the right surface — a call for another audience
/// reached the wrong listener — and nothing more. Any workload that completes
/// the mutual-TLS handshake can declare the served audience and pass, because
/// the peer's proven identity is never compared against the audience it
/// claims.
///
/// Binding the two is what turns this into authorization, and
/// [`check_audience_binding`] is that: an authority method calls it instead of
/// this, so a correctly-addressed call from an unknown workload is denied. This
/// function remains the right check for a surface that authorizes nothing —
/// the conformance surface, which commits nothing authoritative.
///
/// # Errors
///
/// Returns [`StateError::Denied`] naming `family` and nothing else. The reason
/// belongs in the audit record, not in a message the caller could probe.
pub fn check_audience(
    declared: &Audience,
    served: &Audience,
    family: &OperationFamily,
) -> Result<(), StateError> {
    if declared == served {
        return Ok(());
    }
    Err(StateError::Denied {
        family: family.clone(),
    })
}

/// Refuses a call whose transport budget is already spent.
///
/// `remaining` is what the transport's own deadline — the caller's
/// `Connect-Timeout-Ms`, moderated by the listener's policy — leaves at the
/// moment the handler starts. `None` means the caller named no deadline, which
/// is not an expiry; the module's own budget still governs.
///
/// Expiry claims nothing about an operation that already committed, which is
/// why the outcome is ambiguous rather than terminal: the durable receipt,
/// retrieved under the same command identity, is what settles it.
///
/// # Errors
///
/// Returns [`StateError::DeadlineExpired`] when nothing is left.
pub fn check_transport_deadline(
    remaining: Option<Duration>,
    family: &OperationFamily,
) -> Result<(), StateError> {
    if remaining.is_some_and(|left| left.is_zero()) {
        return Err(StateError::DeadlineExpired {
            family: family.clone(),
            overrun: Duration::ZERO,
        });
    }
    Ok(())
}

/// Refuses a new call once the listener has begun draining.
///
/// A call already admitted runs to completion; this only closes the door
/// behind it, so a drain finishes the work it holds instead of abandoning it.
/// The outcome is not a State outcome at all — nothing was decided about any
/// command — so it stays a transport refusal, and a caller that reads it back
/// through this crate's error mapping gets an explicitly unknown outcome.
///
/// # Errors
///
/// Returns [`connectrpc::ConnectError`] with
/// [`connectrpc::ErrorCode::Unavailable`] while draining.
pub fn check_not_draining(draining: bool) -> Result<(), connectrpc::ConnectError> {
    if draining {
        return Err(connectrpc::ConnectError::unavailable(
            "this listener is draining and is not admitting new calls",
        ));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use super::*;
    use polyc_state::{conformance::family, error::RetryClass};

    fn family_id() -> OperationFamily {
        OperationFamily::new(family::FAMILY)
    }

    #[test]
    fn the_current_version_is_admitted_and_every_other_one_is_not() {
        assert!(check_call_context_version(CallContextVersion::CURRENT).is_ok());

        let error = check_call_context_version(CallContextVersion::new(1)).unwrap_err();
        assert!(
            matches!(error, StateError::Malformed { ref field, .. } if field == "protocol_version")
        );
        assert_eq!(error.retry_class(), RetryClass::Terminal);
    }

    #[test]
    fn a_foreign_audience_is_denied_and_names_nothing_else() {
        let served = Audience::new(family::AUDIENCE);
        assert!(check_audience(&served, &served, &family_id()).is_ok());

        let error =
            check_audience(&Audience::new("somebody-else"), &served, &family_id()).unwrap_err();
        assert_eq!(
            error,
            StateError::Denied {
                family: family_id()
            }
        );
        assert_eq!(error.retry_class(), RetryClass::Terminal);
        assert!(
            !error.to_string().contains("somebody-else"),
            "a denial must not echo what the caller asked for"
        );
    }

    #[test]
    fn a_spent_transport_budget_expires_ambiguously() {
        assert!(check_transport_deadline(None, &family_id()).is_ok());
        assert!(check_transport_deadline(Some(Duration::from_millis(1)), &family_id()).is_ok());

        let error = check_transport_deadline(Some(Duration::ZERO), &family_id()).unwrap_err();
        assert!(matches!(error, StateError::DeadlineExpired { .. }));
        assert!(error.is_ambiguous() && error.is_retry_safe());
    }

    #[test]
    fn a_draining_listener_refuses_a_new_call() {
        assert!(check_not_draining(false).is_ok());
        let error = check_not_draining(true).unwrap_err();
        assert_eq!(error.code, connectrpc::ErrorCode::Unavailable);
    }
}

/// The proven identity of the calling workload.
///
/// Proven, not claimed: a workload identity exists here only because rustls
/// verified a client certificate chaining to this listener's authority, and the
/// value is a digest of that verified leaf. A caller that presented no
/// certificate is [`PeerIdentity::Anonymous`], which is an identity too — one a
/// binding can name, and which the shipped composition never names.
///
/// The identity is the certificate rather than a name inside it. That is
/// narrower than a deployment eventually wants, because rotating a workload's
/// certificate changes its identity and so needs the binding updated with it;
/// the credential registry that gives a workload a stable name outliving its
/// certificate is a later chunk's, and inventing half of one here would be a
/// second registry to migrate.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum PeerIdentity {
    /// The caller proved no workload identity.
    Anonymous,
    /// The caller proved this workload identity.
    Workload(String),
}

impl PeerIdentity {
    /// Derives the identity a verified leaf certificate carries.
    ///
    /// `leaf` is the verified end-entity certificate's raw bytes, or [`None`]
    /// when the transport verified none. Nothing is parsed out of the
    /// certificate: the digest of the exact bytes rustls accepted is what this
    /// listener knows about the peer, and a value derived from the whole leaf
    /// cannot be confused by a field inside it.
    #[must_use]
    pub fn from_verified_leaf(leaf: Option<&[u8]>) -> Self {
        leaf.map_or(Self::Anonymous, |bytes| {
            use sha2::Digest as _;
            let digest = sha2::Sha256::digest(bytes);
            Self::Workload(hex::encode(digest))
        })
    }

    /// Names a workload identity directly.
    ///
    /// A composition builds its binding out of these; nothing on the serving
    /// path does.
    #[must_use]
    pub fn workload(identity: impl Into<String>) -> Self {
        Self::Workload(identity.into())
    }

    /// Reports whether the caller proved no workload identity.
    #[must_use]
    pub const fn is_anonymous(&self) -> bool {
        matches!(self, Self::Anonymous)
    }
}

impl fmt::Display for PeerIdentity {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Anonymous => f.write_str("anonymous"),
            Self::Workload(identity) => f.write_str(identity),
        }
    }
}

/// Which audiences each proven workload identity may act for.
///
/// Deny-by-default and static: a newly built binding permits nothing at all,
/// including anonymous callers, and it only grows by an explicit
/// [`AudienceBinding::allow`] in a composition. There is no wildcard and no
/// inheritance — a richer delegated-capability model is a later chunk's, and
/// this is the narrowest thing that makes the first authority method safe to
/// mount.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AudienceBinding {
    permitted: BTreeMap<PeerIdentity, BTreeSet<Audience>>,
}

impl AudienceBinding {
    /// Builds a binding that permits nothing.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            permitted: BTreeMap::new(),
        }
    }

    /// Returns the same binding permitting `identity` to act for `audience`.
    #[must_use]
    pub fn allow(mut self, identity: PeerIdentity, audience: Audience) -> Self {
        self.permitted.entry(identity).or_default().insert(audience);
        self
    }

    /// Reports whether `identity` may act for `audience`.
    #[must_use]
    pub fn permits(&self, identity: &PeerIdentity, audience: &Audience) -> bool {
        self.permitted
            .get(identity)
            .is_some_and(|audiences| audiences.contains(audience))
    }

    /// Reports whether the binding permits nothing at all.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.permitted.is_empty()
    }
}

/// Refuses a caller that is not bound to the audience it asks on behalf of.
///
/// Two questions, in order, because they fail for different reasons and a
/// caller learns neither: first the addressing one [`check_audience`] asks — is
/// this the surface that serves the declared audience — and then the
/// authorization one — may the workload that proved its identity on this
/// connection act for that audience at all.
///
/// Deny-by-default: an identity the binding does not name is refused, and so is
/// an anonymous caller unless a composition named anonymity explicitly.
///
/// # Errors
///
/// Returns [`StateError::Denied`] naming `family` and nothing else, for either
/// question. The two are indistinguishable to the caller on purpose: which of
/// them failed is an audit-record detail, not a probe result.
pub fn check_audience_binding(
    identity: &PeerIdentity,
    declared: &Audience,
    served: &Audience,
    binding: &AudienceBinding,
    family: &OperationFamily,
) -> Result<(), StateError> {
    check_audience(declared, served, family)?;
    if binding.permits(identity, declared) {
        return Ok(());
    }
    Err(StateError::Denied {
        family: family.clone(),
    })
}

#[cfg(test)]
mod binding_tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use super::*;
    use polyc_state::{error::RetryClass, journal};

    fn family_id() -> OperationFamily {
        journal::family()
    }

    fn served() -> Audience {
        state_audience()
    }

    #[test]
    fn a_verified_leaf_yields_a_stable_workload_identity() {
        let one = PeerIdentity::from_verified_leaf(Some(b"leaf-bytes"));
        assert_eq!(one, PeerIdentity::from_verified_leaf(Some(b"leaf-bytes")));
        assert_ne!(one, PeerIdentity::from_verified_leaf(Some(b"other-bytes")));
        assert!(!one.is_anonymous());
        assert_eq!(one.to_string().len(), 64, "a sha-256 digest, hex encoded");

        let none = PeerIdentity::from_verified_leaf(None);
        assert!(none.is_anonymous());
        assert_eq!(none.to_string(), "anonymous");
    }

    #[test]
    fn a_new_binding_permits_nothing_including_anonymity() {
        let binding = AudienceBinding::new();
        assert!(binding.is_empty());
        for identity in [
            PeerIdentity::Anonymous,
            PeerIdentity::workload("control-plane"),
        ] {
            let error =
                check_audience_binding(&identity, &served(), &served(), &binding, &family_id())
                    .unwrap_err();
            assert_eq!(
                error,
                StateError::Denied {
                    family: family_id()
                },
                "{identity}"
            );
            assert_eq!(error.retry_class(), RetryClass::Terminal);
        }
    }

    /// The obligation this check exists for: the call is addressed correctly,
    /// and the caller still cannot make it.
    #[test]
    fn a_correctly_addressed_call_from_the_wrong_workload_is_denied() {
        let binding =
            AudienceBinding::new().allow(PeerIdentity::workload("control-plane"), served());

        assert!(
            check_audience_binding(
                &PeerIdentity::workload("control-plane"),
                &served(),
                &served(),
                &binding,
                &family_id()
            )
            .is_ok(),
            "the bound workload asks for the audience it is bound to"
        );

        let error = check_audience_binding(
            &PeerIdentity::workload("projector"),
            &served(),
            &served(),
            &binding,
            &family_id(),
        )
        .unwrap_err();
        assert_eq!(
            error,
            StateError::Denied {
                family: family_id()
            }
        );
        assert!(
            !error.to_string().contains("projector"),
            "a denial must not echo who was refused"
        );
    }

    /// A workload bound to one audience cannot act for another, even though
    /// this surface would serve that audience to somebody.
    #[test]
    fn a_binding_does_not_generalize_across_audiences() {
        let other = Audience::new("forensics");
        let binding = AudienceBinding::new().allow(PeerIdentity::workload("control-plane"), other);
        assert!(
            check_audience_binding(
                &PeerIdentity::workload("control-plane"),
                &served(),
                &served(),
                &binding,
                &family_id()
            )
            .is_err()
        );
    }

    /// The addressing check still runs first: a call for an audience this
    /// surface does not serve is refused whatever the binding says.
    #[test]
    fn a_bound_workload_asking_the_wrong_surface_is_still_denied() {
        let elsewhere = Audience::new("somewhere-else");
        let binding = AudienceBinding::new()
            .allow(PeerIdentity::workload("control-plane"), elsewhere.clone());
        assert!(
            check_audience_binding(
                &PeerIdentity::workload("control-plane"),
                &elsewhere,
                &served(),
                &binding,
                &family_id()
            )
            .is_err()
        );
    }

    /// Anonymity is an identity a composition may name — which a loopback test
    /// over plaintext does, deliberately and visibly — and never a default.
    #[test]
    fn anonymity_is_permitted_only_when_a_composition_names_it() {
        let binding = AudienceBinding::new().allow(PeerIdentity::Anonymous, served());
        assert!(
            check_audience_binding(
                &PeerIdentity::Anonymous,
                &served(),
                &served(),
                &binding,
                &family_id()
            )
            .is_ok()
        );
        assert!(binding.permits(&PeerIdentity::Anonymous, &served()));
        assert!(!binding.permits(&PeerIdentity::workload("x"), &served()));
    }
}