didbot_serve/oauth/

par.rs

//! Pushed authorization requests (RFC 9126), and where a sign-in is decided.
//!
//! **One seam, and this is it.** PAR is the earliest point the client is
//! known — the first request that both names a `client_id` and can be
//! validated against fetched metadata — and it is where the decision record
//! an agent later answers is written. Two inputs reach that verdict and
//! nothing else does: `plan/policy.md`'s deny-only tree, asked about a
//! `didbot_policy::Subject::Grant`, and `plan/scope-policy.md`'s ceiling,
//! applied by intersection.
//!
//! Client admission is one of those denials rather than a seam beside them.
//! `plan/app-allowlist.md`'s "refuse at the pushed authorization request" is
//! a `denyClient`-shaped policy in the tree, and it is asked about before
//! anything else here: `didbot_policy::Subject::Grant` names a `client_id`
//! and the requested scope but no account, so the gate's answer cannot
//! depend on `login_hint` and does not wait for it. A client the gate
//! refuses outright gets no decision record and no `request_uri` — see
//! [`ParError::ClientRefused`] — the opposite of every other denial this
//! module writes, and deliberately so: the record would have to say so to
//! be useful, and saying so to an unauthenticated caller is exactly what
//! `plan/app-allowlist.md` keeps the operator's to know.
//!
//! The record is never judged again. `GET /oauth/authorize` and
//! `bot.did.approveAuthorization` read the record this module wrote — see
//! [`crate::oauth::decision`] — so a policy edited after a push never
//! changes an answer already given. It does bind the token that answer
//! would lead to: `POST /oauth/token` asks the gate again before issuing
//! one — see `crate::oauth::token::GateAdmission`.
//!
//! **`login_hint` naming nobody is not a third seam either — it is not a
//! seam at all.** This endpoint is unauthenticated, and no observable
//! answer may depend on whether an account exists: an account this
//! deployment hosts, an account `plan/policy.md`'s gate would narrow or
//! deny, an account whose own pending bound is already full, and a
//! `login_hint` matching nobody here all get the same `201`, or pushing one
//! request per candidate handle — or enough variously-spelled requests for
//! one candidate — is a directory of this deployment's accounts. The one
//! `403` this endpoint answers with is the server's own `pending_total`
//! bound, which is identity-independent by construction: true for every
//! caller alike, so answering it truthfully leaks nothing. See
//! `push_for_unresolved_hint` (private, this module) for what a nonexistent
//! hint gets instead of a decision record, and why an *account-keyed*
//! bound on it — an earlier version of this fix — reopened the hole by a
//! different door.
//!
//! **An account at its own bound displaces its oldest undecided record.**
//! `login_hint` is whatever the caller writes, so an account's eight slots
//! are fillable by a stranger. Dropping the newest push let that stranger
//! keep an account from signing in to anything; displacing the oldest keeps
//! the bound and keeps the account's freshest attempt answerable — including
//! when this server's own `pending_total` is full, since a push that gives a
//! slot back before it takes one does not grow the store. See
//! [`crate::oauth::decision::DecisionStore::put`].

use std::collections::HashMap;
use std::sync::RwLock;

use rand::rngs::OsRng;
use rand::TryRngCore;
use time::{Duration, OffsetDateTime};

use crate::oauth::authorize::{ResponseMode, ScopePolicy};
use crate::oauth::client_auth::{ClientAssertionVerifier, ClientAuthError, ClientCredentials};
use crate::oauth::client_metadata::{ClientKey, ClientMetadataDocument};
use crate::oauth::consent::{self, ConsentStore};
use crate::oauth::decision::{
    verdict_for, ClientSelf, DecisionRecord, DecisionRefused, DecisionState, DecisionStore,
    Verdict, CEILING_RULE,
};
use crate::oauth::dpop_seam::DpopThumbprint;
use crate::oauth::permission_set::PermissionSets;
use crate::oauth::scope::{Scope, ScopeSet};

/// How long a pushed request stays redeemable. RFC 9126 recommends 60
/// seconds; this server gives a little more room for the agent-driven
/// browser `plan/oauth.md` describes to load `authorize` before the request
/// URI it just pushed goes stale.
pub const REQUEST_URI_TTL: Duration = Duration::minutes(2);

/// Random bytes a `request_uri` is drawn from, before encoding.
const REQUEST_URI_BYTES: usize = 32;

/// The prefix RFC 9126 requires: "the URI ... MUST include the
/// `urn:ietf:params:oauth:request_uri` prefix".
const REQUEST_URI_PREFIX: &str = "urn:ietf:params:oauth:request_uri:";

/// The longest `login_hint` a push may carry, in bytes.
///
/// A hint is a handle or a DID. atproto's specifications cap a handle at 253
/// characters and a DID at 2048, so nothing longer names an account.
pub const MAX_LOGIN_HINT_BYTES: usize = 2048;

/// The longest `state` a push may carry, in bytes.
///
/// `state` is the client's own random token, which it checks when the
/// redirect comes back. jacquard-oauth and `@atproto/oauth-client` both send
/// 16 random bytes as 22 characters. A bound over forty times that reaches
/// no real client, and caps what this server keeps for a stranger's push.
pub const MAX_STATE_BYTES: usize = 1024;

/// How long an S256 `code_challenge` is: the unpadded base64url encoding of
/// a SHA-256 digest (RFC 7636 §4.2).
const S256_CHALLENGE_LENGTH: usize = 43;

/// One pushed authorization request, as pushed.
#[derive(Debug, Clone)]
pub struct PushedRequest {
    /// The client that pushed it, exactly as presented — `authorize` and
    /// `token` both re-check this against what they are given, rather than
    /// trusting a `request_uri` to imply it.
    pub client_id: String,
    /// The document `client_id` resolved to at push time, and its content
    /// key. Carried through rather than re-fetched, so a document that
    /// changes between PAR and `authorize` cannot smuggle a different
    /// redirect URI or capability into a request that was already admitted.
    pub client_key: ClientKey,
    /// The resolved document itself, carried through for the same reason.
    pub client_metadata: ClientMetadataDocument,
    /// Where an authorization code may be delivered; checked against
    /// `client_metadata.redirect_uris` at push time and again at `token`.
    pub redirect_uri: String,
    /// The scope requested, before any ceiling narrows it, with each
    /// `include:` replaced by what its permission set grants.
    pub scope: ScopeSet,
    /// The client's own opaque state, echoed back unexamined.
    pub state: Option<String>,
    /// Where the authorization response carries its parameters.
    pub response_mode: ResponseMode,
    /// PKCE, mandatory for a public client and the norm for atproto OAuth.
    pub code_challenge: String,
    /// The PKCE method the challenge was computed with; only `S256` is
    /// accepted.
    pub code_challenge_method: String,
    /// The account this authorization is for. atproto's own extension:
    /// `login_hint` names the at-identifier the client wants to sign in as.
    pub login_hint: Option<String>,
    /// The account `login_hint` resolved to, looked up here rather than at
    /// `authorize`: the decision record this push mints names an account.
    /// A `login_hint` naming nobody never reaches this struct at all — see
    /// [`push_authorization_request`]'s own doc for what it gets instead.
    pub did: String,
    /// The RFC 7638 thumbprint of the key the push's DPoP proof was signed
    /// with: the request's `dpop_jkt` (RFC 9449 §10.1). The code it leads to
    /// is exchanged only under the same key.
    pub dpop_jkt: DpopThumbprint,
    /// When this pushed request must be redeemed by, per [`REQUEST_URI_TTL`].
    pub expires_at: OffsetDateTime,
}

/// Why a pushed request could not be accepted.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ParError {
    /// A required OAuth parameter was absent.
    #[error("`{0}` is a required parameter")]
    MissingParameter(&'static str),
    /// `response_type` named something other than `code`.
    #[error("only `code` is accepted for response_type")]
    UnsupportedResponseType,
    /// `response_mode` named a mode this server does not answer in.
    #[error("response_mode `{0}` is not one this server answers in")]
    UnsupportedResponseMode(String),
    /// `redirect_uri` is not one the client's own metadata declares.
    #[error("redirect_uri `{0}` is not one this client's metadata declares")]
    RedirectUriNotRegistered(String),
    /// `code_challenge_method` named something other than `S256`.
    #[error("only `S256` is accepted for code_challenge_method, got `{0}`")]
    UnsupportedCodeChallengeMethod(String),
    /// `code_challenge` is not 43 base64url characters, which every S256
    /// challenge is.
    #[error("code_challenge must be the 43 base64url characters of an S256 digest")]
    MalformedCodeChallenge,
    /// A parameter is longer than any value a client sends for it; see
    /// [`MAX_STATE_BYTES`] and [`MAX_LOGIN_HINT_BYTES`].
    #[error("`{field}` is longer than the {max} bytes this server accepts")]
    TooLong {
        /// The parameter.
        field: &'static str,
        /// Its bound, in bytes.
        max: usize,
    },
    /// `scope` did not parse as the atproto scope grammar.
    #[error("scope did not parse: {0}")]
    BadScope(#[from] crate::oauth::scope::ScopeParseError),
    /// `client_id` is not in the one spelling a policy matches; see
    /// `didbot_policy::ClientId`.
    #[error(transparent)]
    ClientId(#[from] didbot_pds::policy::ClientIdError),
    /// The client's metadata document could not be fetched or validated.
    ///
    /// One sentence for every way that can happen, naming nothing the fetch
    /// found. `client_id` is a URL this unauthenticated caller chose, so a
    /// message that distinguished "nothing answered" from "it redirected"
    /// from "the body was too large" would answer questions about this
    /// deployment's own network for whoever asked them — see
    /// [`didbot_http::REFUSED_BLOCKS`] for the addresses
    /// that are refused before the fetch. The cause this variant carries is
    /// written to the journal instead; see `crate::refusal`.
    #[error("this client's metadata document could not be used")]
    ClientMetadata(#[from] crate::oauth::client_metadata::ClientMetadataError),
    /// The client did not authenticate the way its metadata says it does;
    /// see [`crate::oauth::client_auth`].
    #[error(transparent)]
    ClientAuthentication(#[from] ClientAuthError),
    /// `dpop_jkt` names a key other than the one the push's DPoP proof was
    /// signed with (RFC 9449 §10.1).
    #[error("dpop_jkt names a different key than this request's DPoP proof")]
    DpopKeyMismatch,
    /// `plan/policy.md`'s gate refuses every grant this `client_id` could
    /// ever ask for here — a `denyClient` rule, or any rule shaped the same
    /// way — independent of which account `login_hint` names.
    /// `didbot_policy::Subject::Grant` carries no account field at all (see
    /// `didbot_pds::policy::grant_subject`), so this judgment is a fact
    /// about the client and the requested scope alone; checking it before
    /// `login_hint` is even resolved changes nothing it could answer.
    ///
    /// Refused here, before a decision record exists, and the same way for
    /// every `login_hint` alike — a real account, an account this gate
    /// would otherwise grant, or one nobody has ever heard of, since the
    /// judgment is made before any of them is looked at. That is what keeps
    /// this refusal from being a second account-enumeration oracle beside
    /// the one [`push_authorization_request`]'s own doc closes: it fires on
    /// `client_id` and requested scope alone, so no `login_hint` value can
    /// change which way it comes out. A grant this deployment *would* make
    /// still gets the ordinary opaque-decision-record treatment
    /// (`plan/oauth.md`'s "a pushed request is a decision record") — only a
    /// client this gate refuses unconditionally skips minting one, because
    /// there is nothing an account could approve.
    #[error("this pushed authorization request was not accepted")]
    ClientRefused,
    /// This server already holds as many undecided records as it will at
    /// once — server load, not a fact about any account, so it is safe to
    /// answer identically to every caller. Only ever built from
    /// [`DecisionRefused::TooMany`]: an account's own bound overflowing is
    /// answered as a plain success instead, never as this — see
    /// [`push_authorization_request`]'s own doc for why.
    #[error(transparent)]
    TooManyPending(#[from] DecisionRefused),
}

/// Where pushed requests live between `POST /oauth/par` and `authorize`.
pub trait ParStore: Send + Sync {
    /// Mints a fresh `request_uri`, drawn from the same random source and
    /// format [`ParStore::push`] uses, without storing anything under it.
    ///
    /// `push_for_unresolved_hint` calls this rather than reaching for its
    /// own random source: a `request_uri` this endpoint hands back is this
    /// store's format to own, whether or not anything is ever stored under
    /// it, so there is exactly one place that decides what one looks like.
    fn mint(&self) -> String;
    /// Stores a freshly pushed request, under a `request_uri` from
    /// [`ParStore::mint`].
    fn push(&self, request: PushedRequest) -> String;
    /// The request named by `request_uri`, left where it is, expired or not.
    /// `authorize` answers every load of the same `request_uri`, which RFC
    /// 9126 §4 allows, and refuses an expired one to its client.
    fn get(&self, request_uri: &str) -> Option<PushedRequest>;
    /// Removes the request named by `request_uri`, handing it back if it has
    /// not expired. What undoes a push whose decision record was refused or
    /// displaced.
    fn take(&self, request_uri: &str) -> Option<PushedRequest>;
}

/// The in-process PAR store.
#[derive(Debug, Default)]
pub struct MemoryParStore {
    entries: RwLock<HashMap<String, PushedRequest>>,
}

impl MemoryParStore {
    /// An empty store.
    pub fn new() -> Self {
        Self::default()
    }

    /// How many pushed requests this store still holds. Test use only.
    #[cfg(test)]
    pub(crate) fn len(&self) -> usize {
        self.entries.read().unwrap_or_else(|p| p.into_inner()).len()
    }
}

fn random_request_uri() -> String {
    let mut bytes = [0u8; REQUEST_URI_BYTES];
    OsRng
        .try_fill_bytes(&mut bytes)
        .expect("the OS random source is available");
    format!("{REQUEST_URI_PREFIX}{}", hex::encode(bytes))
}

impl ParStore for MemoryParStore {
    fn mint(&self) -> String {
        random_request_uri()
    }

    fn push(&self, request: PushedRequest) -> String {
        let uri = self.mint();
        let now = OffsetDateTime::now_utc();
        let mut entries = self.entries.write().unwrap_or_else(|p| p.into_inner());
        // Nothing else removes a request whose decision was stored, so each
        // one is swept here once it has expired, the same way
        // `MemoryConsentStore::mint` sweeps its own map.
        entries.retain(|_, pushed| now < pushed.expires_at);
        entries.insert(uri.clone(), request);
        uri
    }

    fn get(&self, request_uri: &str) -> Option<PushedRequest> {
        self.entries
            .read()
            .unwrap_or_else(|p| p.into_inner())
            .get(request_uri)
            .cloned()
    }

    fn take(&self, request_uri: &str) -> Option<PushedRequest> {
        let mut entries = self.entries.write().unwrap_or_else(|p| p.into_inner());
        let request = entries.remove(request_uri)?;
        if OffsetDateTime::now_utc() >= request.expires_at {
            return None;
        }
        Some(request)
    }
}

/// The raw wire fields `POST /oauth/par` accepts, before client metadata is
/// resolved.
#[derive(Debug, Clone)]
pub struct ParRequest {
    /// The HTTPS URL naming the client's metadata document.
    pub client_id: String,
    /// The OAuth response type. RFC 6749 §4.1.1 requires it, and only
    /// `code` is accepted.
    pub response_type: Option<String>,
    /// Where the authorization response carries its parameters; see
    /// [`ResponseMode`]. Absent is `query`.
    pub response_mode: Option<String>,
    /// Where an authorization code should be delivered.
    pub redirect_uri: Option<String>,
    /// The requested scope, unparsed.
    pub scope: Option<String>,
    /// The client's own opaque state, echoed back unexamined.
    pub state: Option<String>,
    /// PKCE code challenge.
    pub code_challenge: Option<String>,
    /// PKCE code challenge method; only `S256` is accepted.
    pub code_challenge_method: Option<String>,
    /// The at-identifier the client wants to sign in as.
    pub login_hint: Option<String>,
    /// The client assertion a confidential client authenticates with.
    pub credentials: ClientCredentials,
    /// The key the client names for exchanging the code, when it names one
    /// (RFC 9449 §10).
    pub dpop_jkt: Option<String>,
    /// The thumbprint of the key the push's DPoP proof proved possession of.
    pub proof_key: DpopThumbprint,
}

/// The seams PAR reaches to work out what it would grant, bundled so
/// [`push_authorization_request`] takes one argument rather than seven.
///
/// None of these is PAR's own: the ceiling is `plan/scope-policy.md`'s, the
/// gate is `plan/policy.md`'s, and the two stores are where the answer goes.
/// This module calls them in the order below and decides nothing itself.
///
/// They are the whole of what judges a sign-in. There is no fourth seam and
/// no second moment — see this module's own doc. The assertion verifier
/// judges nothing about the grant: it checks that a confidential client is
/// who it says it is.
pub struct DecisionSeams<'a> {
    /// The ceiling this account may ever hold from this client.
    pub scope_policy: &'a dyn ScopePolicy,
    /// Where the one-time reference an approval carries is minted.
    pub consent_store: &'a dyn ConsentStore,
    /// Where the record itself goes.
    pub decisions: &'a dyn DecisionStore,
    /// `plan/policy.md`'s deny-only gate, asked about the grant.
    pub gate: &'a dyn didbot_pds::policy::PolicyGate,
    /// Where the *ceiling*'s narrowing or refusal is written down,
    /// payload-free. A policy's refusal is written by
    /// [`DecisionSeams::gate`] itself, so this must be the same log that
    /// gate holds or the two halves of one deployment's answer land in
    /// different places.
    pub evaluation_log: &'a didbot_pds::evaluation_log::EvaluationLog,
    /// This deployment's own DID, for the subject the gate judges.
    pub pds_did: &'a str,
    /// This deployment's own hostname, for the same subject.
    pub pds_hostname: &'a str,
    /// Whether this deployment fetches a loopback `http://` client metadata
    /// document; see [`crate::oauth::client_metadata::LoopbackClients`].
    pub loopback_clients: crate::oauth::client_metadata::LoopbackClients,
    /// Where a confidential client's assertion is checked; see
    /// [`crate::oauth::client_auth`].
    pub client_assertions: &'a ClientAssertionVerifier,
    /// This deployment's issuer, which a client assertion's `aud` must name.
    pub issuer: &'a str,
    /// Where an `include:` scope's permission set is resolved; see
    /// [`crate::oauth::permission_set`].
    pub permission_sets: &'a PermissionSets,
}

/// Handles one pushed authorization request end to end: resolves and
/// validates the client's metadata, checks the request against it, resolves
/// the account it names, decides what it would grant, and stores both the
/// request and the decision.
///
/// Returns the minted `request_uri` and its lifetime — for every
/// well-formed push alike, whatever it goes on to answer. `resolve` turns
/// `login_hint` into an account DID — the HTTP layer's own
/// `didbot_pds::Registry` lookup, passed in because this module holds no
/// registry — and a hint that resolves to nobody gets the *same* `201`, not
/// a failure: see `push_for_unresolved_hint` (private, this module) for
/// what it gets instead of a decision record, and why. This is the fix for
/// this endpoint's own
/// account-enumeration hole — unauthenticated, and the earliest point a
/// `login_hint` is even looked at — which a distinguishable refusal here
/// reopened no matter how carefully worded, since a caller only has to
/// compare it against what a *resolvable* hint gets, and nothing here ever
/// refuses one of those. `login_hint` is mandatory here — RFC 9126 itself
/// does not require one, but this profile addresses the decision record it
/// mints to an account, and there is no other field naming which one; a
/// missing hint is [`ParError::MissingParameter`], the same as any other
/// field this endpoint cannot proceed without, since every account alike
/// would refuse the request regardless of which one asked.
///
/// **A `403` must depend on server load, never on an account.** A real
/// account whose own `pending_per_account` bound is already full is *not*
/// refused — that would resurrect the same oracle as a `login_hint` naming
/// nobody, just aimed at an account that exists rather than one that does
/// not (push enough candidate spellings and time which ones eventually
/// answer `403`). Its push is stored, and the account's oldest undecided
/// record makes way for it
/// ([`crate::oauth::decision::DecisionStore::put`]). That record's pushed
/// request is taken back out of [`ParStore`] and its
/// [`crate::oauth::consent::ConsentReference`] discarded from
/// [`crate::oauth::consent::ConsentStore`] — both undone, or an
/// unauthenticated caller could leak one live consent entry per push
/// against any account it names, forever. An `authorize` for the displaced
/// request then fails exactly as it would for a `request_uri` that expired.
/// The only `403` this endpoint answers with is
/// [`crate::oauth::decision::DecisionRefused::TooMany`] — the server itself
/// out of room, true for every caller identically — and a displacement is
/// logged at `info`, sampled, so an operator can still see a flood
/// happening even though the caller cannot tell it apart from an ordinary
/// push.
///
/// The decision is minted in the same call, and the two are stored
/// together on the path that stores anything at all: if the record is
/// refused by the server-wide bound the pushed request is taken straight
/// back out, so a `request_uri` never exists without the decision that
/// explains it — and neither does a displaced one.
pub async fn push_authorization_request(
    http: &reqwest::Client,
    cache: &dyn crate::oauth::client_metadata::ClientMetadataCache,
    store: &dyn ParStore,
    request: ParRequest,
    resolve: &(dyn Fn(&str) -> Option<String> + Send + Sync),
    seams: &DecisionSeams<'_>,
) -> Result<(String, Duration), ParError> {
    // Checked before the client's document is fetched: it needs no document,
    // and a request refused here costs this server no outbound request.
    match request.response_type.as_deref() {
        None => return Err(ParError::MissingParameter("response_type")),
        Some("code") => {}
        Some(_) => return Err(ParError::UnsupportedResponseType),
    }
    let response_mode = match request.response_mode.as_deref() {
        None => ResponseMode::default(),
        Some(named) => ResponseMode::named(named)
            .ok_or_else(|| ParError::UnsupportedResponseMode(named.to_owned()))?,
    };
    // Both are kept with the pushed request, and `state` with its pending
    // consent too, so both are bounded before anything else is spent.
    at_most("state", request.state.as_deref(), MAX_STATE_BYTES)?;
    at_most(
        "login_hint",
        request.login_hint.as_deref(),
        MAX_LOGIN_HINT_BYTES,
    )?;
    let client_id = didbot_pds::policy::ClientId::parse(&request.client_id)?;

    let scope_str = request.scope.ok_or(ParError::MissingParameter("scope"))?;
    let (scope, forms) = ScopeSet::read(&scope_str)?;
    let login_hint = request
        .login_hint
        .ok_or(ParError::MissingParameter("login_hint"))?;

    // `didbot_policy::Subject::Grant` names a `client_id` and the requested
    // scope but no account (`didbot_pds::policy::grant_subject`), so this is
    // a fact about the client alone and is asked about before `login_hint`
    // is even resolved — see [`ParError::ClientRefused`]. Built once and
    // reused by [`decide`] for its own row in the evaluation log, rather
    // than rebuilt there, so a caller of this function is not the only
    // thing this judgment is asked on behalf of exactly once.
    //
    // **Asked before the document is fetched.** Everything it judges —
    // `client_id` and the requested scope — is in the request itself, and
    // the fetch is an outbound request to a host this unauthenticated caller
    // named. A client the gate refuses should not cost this deployment one.
    let atoms: Vec<String> = scope.0.iter().map(ToString::to_string).collect();
    let borrowed: Vec<&str> = atoms.iter().map(String::as_str).collect();
    let subject =
        didbot_pds::policy::grant_subject(seams.pds_did, seams.pds_hostname, &client_id, &borrowed);
    if !matches!(
        seams.gate.judge(&subject),
        didbot_pds::policy::Outcome::Allow
    ) {
        // Nothing downstream has anything to approve, and no account was
        // ever looked at, so nothing is minted.
        return Err(ParError::ClientRefused);
    }

    // The document is built from, or fetched at, the id the client sent: the
    // profile has a local client's document carry that exact string.
    let (document, client_key) = crate::oauth::client_metadata::resolve_client(
        http,
        cache,
        &request.client_id,
        seams.loopback_clients,
    )
    .await?;

    // The client proves who it is before anything it asked for is read. A
    // confidential client's assertion is spent here, and the login this push
    // leads to stays bound to the key that signed it.
    let client_auth_key =
        seams
            .client_assertions
            .authenticate(&document, &request.credentials, seams.issuer)?;

    // RFC 9449 §10.1: a push that names its key in `dpop_jkt` names the key
    // its proof was signed with. Checked where the reference provider checks
    // it, once the client has authenticated and before the rest of the
    // request is read.
    if request
        .dpop_jkt
        .as_deref()
        .is_some_and(|named| named != request.proof_key.0)
    {
        return Err(ParError::DpopKeyMismatch);
    }

    let redirect_uri = request
        .redirect_uri
        .ok_or(ParError::MissingParameter("redirect_uri"))?;
    if !document.admits_redirect_uri(&redirect_uri) {
        return Err(ParError::RedirectUriNotRegistered(redirect_uri));
    }

    let code_challenge = request
        .code_challenge
        .ok_or(ParError::MissingParameter("code_challenge"))?;
    let code_challenge_method = request
        .code_challenge_method
        .ok_or(ParError::MissingParameter("code_challenge_method"))?;
    if code_challenge_method != "S256" {
        return Err(ParError::UnsupportedCodeChallengeMethod(
            code_challenge_method,
        ));
    }
    if !is_s256_challenge(&code_challenge) {
        return Err(ParError::MalformedCodeChallenge);
    }

    // An atom in a form the permission spec does not define was read as well
    // as it could be. Said once the client has shown who it is, so that a
    // caller naming no real client writes nothing at `warn`.
    for (atom, warning) in &forms {
        tracing::warn!(
            client_id = client_id.as_str(),
            atom = atom.as_str(),
            %warning,
            "a pushed authorization request asked for a scope in a form the permission spec does not define"
        );
    }

    // Each `include:` becomes the atoms its permission set grants, before
    // the ceiling sees the request. Resolved before `login_hint` is, so
    // that how long a push takes depends on no account.
    let scope = seams.permission_sets.expand(&scope).await;

    let now = OffsetDateTime::now_utc();
    let expires_at = now + REQUEST_URI_TTL;

    let Some(did) = resolve(&login_hint) else {
        return push_for_unresolved_hint(store, seams.decisions);
    };

    let verdict = decide(&did, client_id.as_str(), &scope, &subject, seams);

    let pushed = PushedRequest {
        client_id: request.client_id.clone(),
        client_key: client_key.clone(),
        client_metadata: document.clone(),
        redirect_uri: redirect_uri.clone(),
        scope: scope.clone(),
        state: request.state.clone(),
        response_mode,
        code_challenge: code_challenge.clone(),
        code_challenge_method: code_challenge_method.clone(),
        login_hint: Some(login_hint),
        did: did.clone(),
        dpop_jkt: request.proof_key.clone(),
        expires_at,
    };
    let uri = store.push(pushed);

    // A denied request gets a record and no reference: there is nothing to
    // approve, so nothing that could approve it is ever minted.
    //
    // What the reference carries is the *verdict's* scope, not the request:
    // the agent is shown `granted` and approves that, so that is what the
    // code and the login it becomes are issued at. A ceiling loosened
    // afterwards cannot hand back an atom the record named as `cut`.
    let token = (!verdict.is_deny()).then(|| {
        consent::begin(
            seams.consent_store,
            uri.clone(),
            did.clone(),
            request.client_id.clone(),
            verdict.granted(&scope),
            redirect_uri,
            request.state,
            code_challenge,
            code_challenge_method,
            client_auth_key,
            request.proof_key,
        )
    });

    let record = DecisionRecord {
        request_uri: uri.clone(),
        account: did.clone(),
        client_origin: origin_of(&request.client_id),
        client_key,
        first_time: false,
        client_self: ClientSelf::of(&document),
        requested: scope,
        verdict,
        // Cloned rather than moved: `put` may refuse the record, and both
        // undo branches below still need the reference `consent::begin`
        // already minted, to discard it — see those branches' own doc.
        token: token.clone(),
        created_at: now,
        expires_at,
        state: DecisionState::Pending,
        sequence: 0,
    };
    match seams.decisions.put(record) {
        Ok(stored) => {
            if let Some(displaced) = stored.displaced {
                // This account was at its own bound, so its oldest undecided
                // record made way for this one. Both of that record's own
                // stores are undone here, for the reason the refusal branch
                // below undoes them: a pushed request nothing will redeem and
                // a reference nothing will present are memory this server
                // holds for the rest of the process's life, and this endpoint
                // is unauthenticated. An operator sees it in the log, sampled
                // so logging is not itself something a flood can amplify.
                let _ = store.take(&displaced.request_uri);
                if let Some(reference) = &displaced.token {
                    seams.consent_store.discard(reference);
                }
                log_displaced_push(&did);
            }
            Ok((uri, REQUEST_URI_TTL))
        }
        Err(refused) => {
            // Server state, identity-independent: the one `403` this
            // endpoint answers with. Undoing both stores fully, not just
            // `ParStore` — a reference `MemoryConsentStore` never sees
            // consumed is a reference it holds for the rest of the
            // process's life otherwise, and this endpoint is
            // unauthenticated.
            let _ = store.take(&uri);
            if let Some(reference) = &token {
                seams.consent_store.discard(reference);
            }
            SERVER_FULL_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            Err(ParError::TooManyPending(refused))
        }
    }
}

/// Refuses `value` if it is longer than `max` bytes.
fn at_most(field: &'static str, value: Option<&str>, max: usize) -> Result<(), ParError> {
    match value {
        Some(value) if value.len() > max => Err(ParError::TooLong { field, max }),
        _ => Ok(()),
    }
}

/// Whether `challenge` has the shape of an S256 code challenge: exactly
/// [`S256_CHALLENGE_LENGTH`] characters of the base64url alphabet.
fn is_s256_challenge(challenge: &str) -> bool {
    challenge.len() == S256_CHALLENGE_LENGTH
        && challenge
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
}

/// How often [`log_displaced_push`] actually emits a log line, rather than
/// only counting one.
///
/// The account whose bound was hit is already known to be at it, so an
/// attacker who has pushed it there once can trigger this branch again for
/// free, from as many caller addresses as it likes — logging every one
/// would make the log itself the thing a flood amplifies. This is one
/// process-wide counter rather than one per account: a per-account count
/// would be the same unbounded-growth shape this fix exists to close, just
/// moved one field over.
const DISPLACED_PUSH_LOG_SAMPLE: u64 = 100;

/// How many undecided records have been displaced past an account's own
/// pending bound, this process's lifetime; see
/// [`DISPLACED_PUSH_LOG_SAMPLE`].
static DISPLACED_PUSH_COUNT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// How many pushes named a `login_hint` this deployment resolved to no
/// account; see [`push_for_unresolved_hint`].
static UNRESOLVED_HINT_COUNT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// How many pushes this server's own `pending_total` refused; see
/// [`ParError::TooManyPending`].
static SERVER_FULL_COUNT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// How many pushes have not left an answerable decision record behind, this
/// process's lifetime, by reason.
///
/// **What an operator reads instead of the log.** Every one of these answers
/// the caller `201` except `server_full`, because this endpoint is
/// unauthenticated and no observable answer may depend on an account — see
/// [`push_authorization_request`]. That is what makes a flood aimed at one
/// account invisible from outside, and the `info` line these count alongside
/// is sampled so that logging cannot itself be flooded. `GET /health`
/// carries them under `oauthPushes`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DroppedPushes {
    /// Pushes past an account's own bound, each of which took that account's
    /// oldest undecided record with it.
    pub displaced: u64,
    /// Pushes whose `login_hint` named no account here.
    pub unresolved_hint: u64,
    /// Pushes refused because this server holds as many undecided records as
    /// it will.
    pub server_full: u64,
}

/// [`DroppedPushes`] as it stands now.
pub fn dropped_pushes() -> DroppedPushes {
    use std::sync::atomic::Ordering;
    DroppedPushes {
        displaced: DISPLACED_PUSH_COUNT.load(Ordering::Relaxed),
        unresolved_hint: UNRESOLVED_HINT_COUNT.load(Ordering::Relaxed),
        server_full: SERVER_FULL_COUNT.load(Ordering::Relaxed),
    }
}

/// Logs a displaced record at `info`, sampled to one in
/// [`DISPLACED_PUSH_LOG_SAMPLE`] so the log line itself cannot be flooded.
fn log_displaced_push(did: &str) {
    use std::sync::atomic::Ordering;
    let seen = DISPLACED_PUSH_COUNT.fetch_add(1, Ordering::Relaxed) + 1;
    if seen % DISPLACED_PUSH_LOG_SAMPLE == 1 {
        tracing::info!(
            %did,
            seen,
            sampled_one_in = DISPLACED_PUSH_LOG_SAMPLE,
            "a push past this account's own pending bound displaced its oldest undecided record"
        );
    }
}

/// What `login_hint` naming nobody gets instead of
/// [`push_authorization_request`]'s ordinary decision record: the identical
/// success this endpoint hands back for a real account, and nothing an
/// unauthenticated caller — or this store — can act on.
///
/// **No account state, ever — that is the whole fix, and the whole reason
/// this stores nothing at all.** An earlier version of this function put a
/// placeholder [`DecisionRecord`] through [`DecisionStore::put`], keyed on
/// the raw `login_hint` string, so a nonexistent hint would cost the same
/// `pending_per_account` slot a real account's push would. That reopened
/// the hole a different way: `resolve` folds every spelling of a *real*
/// account — a handle, its DID, one with a redundant `:port` — into the one
/// DID that names it, so a real account's pending bucket is one bucket
/// however it is addressed, while every distinct spelling of a nonexistent
/// one is its own bucket forever. Push eight variously-spelled requests for
/// one candidate and a ninth: a real account's ninth answers differently
/// from its first eight, and a made-up one's ninth never does — the
/// aliasing itself is the oracle, no matter how the two buckets' *refusals*
/// are worded. Worse, since nothing bounds how many distinct invented
/// strings exist, that placeholder was also a new denial-of-service surface
/// past the account-existence one: a handful of addresses at the per-address
/// rate limit could keep `pending_total` permanently full with entries no
/// real account ever asked for, since before this a stranger could not
/// occupy a slot without naming one.
///
/// The fix an account-keyed bound cannot have is not to bound unresolved
/// hints at all — there is no account here for a bound to be *of*. A fresh
/// `request_uri` is minted through [`ParStore::mint`] and handed back with
/// the same `expires_in` a real push gets, and nothing is stored under it
/// anywhere: not in [`ParStore`], so `GET /oauth/authorize` answers
/// [`crate::oauth::authorize::AuthorizeError::UnknownRequestUri`] for it,
/// as it does for a real one this server has swept; not in
/// [`DecisionStore`], so
/// `bot.did.getAuthorization` answers not-found for the same reason it
/// always would — no credential can authenticate as a string that resolved
/// to no account.
///
/// That leaves one thing this must still answer identically to a real
/// push: [`DecisionStore::check_room`], the *server's* own bound, which is
/// a fact about this deployment's load and not about `login_hint` at all —
/// true for every caller alike, so answering it truthfully leaks nothing.
/// Skipping this check would make a full store itself an oracle (real
/// accounts start answering `403`, unresolved hints never would), so the
/// close reads it — without writing anything, whichever way it answers.
fn push_for_unresolved_hint(
    store: &dyn ParStore,
    decisions: &dyn DecisionStore,
) -> Result<(String, Duration), ParError> {
    UNRESOLVED_HINT_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    decisions.check_room().map_err(|refused| {
        SERVER_FULL_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        ParError::TooManyPending(refused)
    })?;
    Ok((store.mint(), REQUEST_URI_TTL))
}

/// What a ceiling's row names where a gate's row names the policy version
/// it judged under.
///
/// A ceiling is this deployment's own configuration, not one of the
/// operator's numbered policies — see [`CEILING_RULE`] — so there is no tree
/// version to put here, and putting the tree's would say a policy tree
/// refused something no policy tree ever saw.
const CEILING_VERSION: &str = "ceiling";

/// The verdict for one request: what the ceiling narrows or denies. The
/// gate is not asked here — [`push_authorization_request`] already asked it,
/// once, before this account was even known to exist, because
/// `didbot_policy::Subject::Grant` has no account for this function's `did`
/// to change the answer of. A gate's own refusal is
/// [`ParError::ClientRefused`], not a [`Verdict`] this function could
/// return; only the ceiling's per-account narrowing lives here.
///
/// `didbot_pds::policy_tree::TreePolicyGate::judge` records its own refusal
/// as it makes it, naming the policies that fired and the version they were
/// judged under; nothing here writes that down a second time. What only this
/// function knows is what the *ceiling* did, which no gate saw, and that is
/// the one row it writes.
///
/// `subject` is [`push_authorization_request`]'s own — built once there for
/// the gate it already asked, and borrowed here rather than rebuilt, since
/// the two calls describe the same request.
fn decide(
    did: &str,
    client_id: &str,
    scope: &ScopeSet,
    subject: &didbot_pds::policy::Subject<'_>,
    seams: &DecisionSeams<'_>,
) -> Verdict {
    let ceiling = seams.scope_policy.ceiling(did, client_id);
    let verdict = verdict_for(scope, &ceiling, CEILING_RULE);

    // `plan/scope-policy.md`'s "log every scopedown": the record itself is
    // one account of it, and this is the operator's. Payload-free, like
    // every other row in that log — the reason names what was cut, never
    // what the client sent, so an atom the grammar could not read, which is
    // the client's own text, is not among the cut atoms. A narrowing is not
    // a refusal and does not claim to be one. Written before the gate is
    // asked, because what the ceiling did is a fact about this request
    // whatever the gate goes on to say.
    let ceiling_row = match &verdict {
        Verdict::Allow => None,
        Verdict::Narrow { cut, rule, .. } => Some((
            didbot_pds::evaluation_log::Verdict::Narrow,
            rule.clone(),
            format!("scope narrowed by `{rule}`"),
            cut.0
                .iter()
                .filter(|atom| !matches!(atom, Scope::Unknown(_)))
                .map(ToString::to_string)
                .collect::<Vec<String>>(),
        )),
        Verdict::Deny { rule, .. } => Some((
            didbot_pds::evaluation_log::Verdict::Reject,
            rule.clone(),
            format!("grant refused by `{rule}`"),
            Vec::<String>::new(),
        )),
    };
    if let Some((log_verdict, rule, reason, cut)) = ceiling_row {
        seams
            .evaluation_log
            .record_denial(didbot_pds::evaluation_log::Denial {
                account: did,
                policies_fired: &[rule],
                verdict: log_verdict,
                reason: &reason,
                cut: &cut,
                payload_hash: &seams.evaluation_log.hash_subject(subject),
                policy_version: &didbot_pds::policy::PolicyVersion(CEILING_VERSION.to_owned()),
                now: subject.universal().now,
                subject,
            });
    }

    verdict
}

/// The scheme, host and port of a `client_id`.
///
/// `plan/app-allowlist.md`: the origin is the only part of a `client_id`
/// that is a fact about the world, since serving a metadata document there
/// means controlling that host. Falls back to the whole string when it will
/// not parse as a URL, which `client_metadata::resolve_client` has already
/// ruled out by the time this is called.
fn origin_of(client_id: &str) -> String {
    url::Url::parse(client_id)
        .map(|url| url.origin().ascii_serialization())
        .unwrap_or_else(|_| client_id.to_owned())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::oauth::authorize::GrantAnyScope;
    use crate::oauth::client_metadata::{ClientMetadataCache, MemoryClientMetadataCache};
    use crate::oauth::consent::MemoryConsentStore;
    use crate::oauth::decision::MemoryDecisionStore;

    const AGENT: &str = "did:web:agent.example";

    fn document() -> ClientMetadataDocument {
        ClientMetadataDocument {
            client_id: "https://client.example/id.json".to_owned(),
            application_type: crate::oauth::client_metadata::ApplicationType::Web,
            client_name: Some("Krillbridge".to_owned()),
            client_uri: None,
            logo_uri: None,
            redirect_uris: vec!["https://client.example/cb".to_owned()],
            grant_types: vec!["authorization_code".to_owned()],
            response_types: vec!["code".to_owned()],
            authentication: crate::oauth::client_auth::ClientAuthentication::public(),
            scope: Some("atproto".to_owned()),
            dpop_bound_access_tokens: true,
        }
    }

    fn cache_with(document: ClientMetadataDocument) -> MemoryClientMetadataCache {
        let cache = MemoryClientMetadataCache::new();
        let key = ClientKey::compute(&document);
        cache.put(
            &document.client_id.clone(),
            crate::oauth::client_metadata::CachedClientMetadata {
                document,
                key,
                fetched_at: OffsetDateTime::now_utc(),
            },
        );
        cache
    }

    fn request() -> ParRequest {
        ParRequest {
            client_id: "https://client.example/id.json".to_owned(),
            response_type: Some("code".to_owned()),
            response_mode: None,
            redirect_uri: Some("https://client.example/cb".to_owned()),
            scope: Some("atproto".to_owned()),
            state: Some("xyz".to_owned()),
            code_challenge: Some(CHALLENGE.to_owned()),
            code_challenge_method: Some("S256".to_owned()),
            login_hint: Some(AGENT.to_owned()),
            credentials: ClientCredentials::default(),
            dpop_jkt: None,
            proof_key: DpopThumbprint("the-pushs-own-key".to_owned()),
        }
    }

    /// The S256 challenge RFC 7636's appendix B works through.
    const CHALLENGE: &str = "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM";

    /// Everything a push needs beyond the request, owned by the test so the
    /// borrowed [`DecisionSeams`] can point into it.
    struct Fixture {
        consent: MemoryConsentStore,
        decisions: MemoryDecisionStore,
        assertions: ClientAssertionVerifier,
        permission_sets: PermissionSets,
        gate: Box<dyn didbot_pds::policy::PolicyGate>,
        sink: std::sync::Arc<didbot_pds::evaluation_log::MemorySink>,
        log: std::sync::Arc<didbot_pds::evaluation_log::EvaluationLog>,
    }

    impl Fixture {
        /// A deployment with nothing loaded: the empty policy set
        /// `plan/policy.md` says permits everything.
        fn new() -> Self {
            let sink = std::sync::Arc::new(didbot_pds::evaluation_log::MemorySink::new());
            Self {
                consent: MemoryConsentStore::new(),
                decisions: MemoryDecisionStore::new(),
                assertions: ClientAssertionVerifier::new(),
                permission_sets: PermissionSets::default(),
                gate: Box::new(didbot_pds::policy::NoPolicyGate),
                log: std::sync::Arc::new(didbot_pds::evaluation_log::EvaluationLog::new(
                    sink.clone(),
                )),
                sink,
            }
        }

        /// The same, with one gate loaded.
        fn with_gate(gate: Box<dyn didbot_pds::policy::PolicyGate>) -> Self {
            Self {
                gate,
                ..Self::new()
            }
        }

        /// A deployment whose startup source supplies one real policy,
        /// denying every grant from a client whose id matches `pattern`.
        /// This is what `plan/app-allowlist.md`'s admission refusal *is*
        /// now: a denial in the tree, judged at PAR like any other.
        fn denying_client(pattern: &str) -> Self {
            let fixture = Self::new();
            let mut records = std::collections::HashMap::new();
            records.insert("startup[0]".to_owned(), tree_denying_client(pattern));
            let (tree, report) = didbot_pds::policy_tree::build_tree(
                Vec::new(),
                &records,
                &std::collections::HashMap::new(),
                &Default::default(),
            );
            assert!(report.scoped_denials.is_empty(), "the policy compiled");
            let gate = didbot_pds::policy_tree::TreePolicyGate::new(tree)
                .with_evaluation_log(std::sync::Arc::clone(&fixture.log));
            Self {
                gate: Box::new(gate),
                ..fixture
            }
        }

        /// Every row this fixture's log has written. A window stays open
        /// until something closes it, so this closes them all first — see
        /// `EvaluationLog::flush_stale`.
        fn rows(&self) -> Vec<didbot_pds::evaluation_log::DenialRow> {
            self.log
                .flush_stale(OffsetDateTime::now_utc() + Duration::hours(1));
            self.sink.rows()
        }

        fn seams<'a>(&'a self, ceiling: &'a dyn ScopePolicy) -> DecisionSeams<'a> {
            DecisionSeams {
                scope_policy: ceiling,
                consent_store: &self.consent,
                decisions: &self.decisions,
                gate: self.gate.as_ref(),
                evaluation_log: self.log.as_ref(),
                pds_did: "did:web:pds.example",
                pds_hostname: "pds.example",
                loopback_clients: crate::oauth::client_metadata::LoopbackClients::for_zone(
                    "pds.example",
                ),
                client_assertions: &self.assertions,
                issuer: "https://pds.example",
                permission_sets: &self.permission_sets,
            }
        }
    }

    /// The account every hint in this module resolves to, standing in for
    /// the HTTP layer's own registry lookup.
    fn resolves(hint: &str) -> Option<String> {
        (hint == AGENT).then(|| AGENT.to_owned())
    }

    async fn push(
        fixture: &Fixture,
        store: &dyn ParStore,
        ceiling: &dyn ScopePolicy,
        request: ParRequest,
    ) -> Result<(String, Duration), ParError> {
        push_authorization_request(
            &didbot_http::client(),
            &cache_with(document()),
            store,
            request,
            &resolves,
            &fixture.seams(ceiling),
        )
        .await
    }

    /// `plan/app-allowlist.md`'s exit criterion, met by a policy rather than
    /// by a seam of its own: a client the tree denies outright gets refused
    /// at PAR itself, before a `request_uri` or a decision record exists —
    /// the same shape an unresolvable `login_hint` leaves behind, and for
    /// the same reason: neither is a fact this unauthenticated endpoint may
    /// hand back distinguishably. See [`ParError::ClientRefused`].
    #[tokio::test]
    async fn a_denied_client_is_refused_at_par() {
        let fixture = Fixture::denying_client("^https://client\\.example/.*$");
        let store = MemoryParStore::new();
        let err = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .unwrap_err();
        assert_eq!(err, ParError::ClientRefused);
        let (pending, _) = fixture.decisions.pending_for(AGENT, None);
        assert!(
            pending.is_empty(),
            "a client refused outright gets no decision record"
        );
    }

    /// **The outbound request this closes.** `POST /oauth/par` is
    /// unauthenticated and `client_id` is a URL the caller chose, so a fetch
    /// before the gate is asked lets anybody make this server reach a host
    /// of their choosing — even one the operator has denied outright. The
    /// gate judges `client_id` and the requested scope, both of which are in
    /// the request, so it is asked first and a refused client costs no
    /// outbound request at all.
    ///
    /// Measured on the clock: the `client_id` names a host that accepts and
    /// never answers, so a fetch would cost `FETCH_TIMEOUT`.
    #[tokio::test(start_paused = true)]
    async fn a_denied_client_is_refused_before_its_metadata_is_fetched() {
        let addr = didbot_http::test_support::hung_server();
        let fixture = Fixture::denying_client("^http://.*$");
        let store = MemoryParStore::new();
        let mut hung = request();
        hung.client_id = format!("http://{addr}/id.json");

        let started = tokio::time::Instant::now();
        let err = push_authorization_request(
            &crate::oauth::client_metadata_http_client(),
            &MemoryClientMetadataCache::new(),
            &store,
            hung,
            &resolves,
            &fixture.seams(&GrantAnyScope),
        )
        .await
        .unwrap_err();
        assert_eq!(err, ParError::ClientRefused);
        assert_eq!(
            started.elapsed(),
            std::time::Duration::ZERO,
            "the document was fetched before the gate was asked"
        );
    }

    /// **The oracle this closes.** `client_id` names a host the caller
    /// chose, so what the fetch found is an answer about this deployment's
    /// own network. Two pushes whose fetches fail in different ways — one
    /// aimed at an address this server refuses outright, one at a host that
    /// answers with a redirect — read identically to the caller, and the
    /// refusal names neither the address nor the redirect target.
    #[tokio::test]
    async fn every_way_a_client_metadata_fetch_can_fail_reads_the_same() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let cache = MemoryClientMetadataCache::new();
        // A host that answers with a redirect, reached the one way a fetch
        // to loopback is allowed at all — a `.localhost` deployment.
        let host = crate::oauth::client_metadata::tests_client_host().await;
        let http =
            crate::oauth::client_metadata::tests_reaching(host, reqwest::redirect::Policy::none());

        // An address this server refuses outright, on an ordinary
        // deployment.
        let mut refused_address = request();
        refused_address.client_id = "https://169.254.169.254/id.json".to_owned();
        let mut redirecting = request();
        redirecting.client_id = format!("{}/moved.json", crate::oauth::client_metadata::TESTS_HOST);

        let mut answers = Vec::new();
        for (probe, zone) in [(refused_address, "pds.example"), (redirecting, "localhost")] {
            let seams = DecisionSeams {
                loopback_clients: crate::oauth::client_metadata::LoopbackClients::for_zone(zone),
                ..fixture.seams(&GrantAnyScope)
            };
            let err = push_authorization_request(&http, &cache, &store, probe, &resolves, &seams)
                .await
                .expect_err("neither client_id yields a usable document");
            assert!(matches!(err, ParError::ClientMetadata(_)), "{err:?}");
            answers.push(err.to_string());
        }
        assert_eq!(answers[0], answers[1]);
        assert!(
            !answers[0].contains("169.254") && !answers[0].contains("moved"),
            "the refusal names what the fetch found: {}",
            answers[0]
        );
    }

    /// A native CLI signs in with the atproto profile's virtual
    /// `http://localhost` `client_id`, whose document is built from its own
    /// query. That fetches nothing, so a production deployment takes it.
    /// What it still refuses is a `client_id` that would make it fetch from
    /// loopback: the host accepts and never answers, so a fetch would cost
    /// `FETCH_TIMEOUT`.
    #[tokio::test(start_paused = true)]
    async fn a_production_zone_takes_a_virtual_loopback_client_but_fetches_no_loopback_one() {
        let fixture = Fixture::new();
        assert_eq!(
            fixture.seams(&GrantAnyScope).loopback_clients,
            crate::oauth::client_metadata::LoopbackClients::Refused,
            "the fixture is a production zone"
        );
        let store = MemoryParStore::new();
        let cache = MemoryClientMetadataCache::new();

        let mut native = request();
        native.client_id =
            "http://localhost?redirect_uri=http%3A%2F%2F127.0.0.1%3A4567%2Fcallback&scope=atproto"
                .to_owned();
        native.redirect_uri = Some("http://127.0.0.1:4567/callback".to_owned());
        let seams = fixture.seams(&GrantAnyScope);
        push_authorization_request(
            &didbot_http::client(),
            &cache,
            &store,
            native,
            &resolves,
            &seams,
        )
        .await
        .expect("a virtual loopback client is accepted on a production zone");

        let addr = didbot_http::test_support::hung_server();
        let mut fetched = request();
        fetched.client_id = format!("http://127.0.0.1:{}/client-metadata.json", addr.port());
        let started = tokio::time::Instant::now();
        let err = push_authorization_request(
            &crate::oauth::client_metadata_http_client(),
            &cache,
            &store,
            fetched,
            &resolves,
            &seams,
        )
        .await
        .expect_err("a production zone does not fetch from loopback");
        assert!(matches!(err, ParError::ClientMetadata(_)), "{err:?}");
        assert_eq!(started.elapsed(), std::time::Duration::ZERO);
    }

    #[tokio::test]
    async fn an_unregistered_redirect_uri_is_refused_even_when_admitted() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let mut bad = request();
        bad.redirect_uri = Some("https://client.example/somewhere-else".to_owned());
        let err = push(&fixture, &store, &GrantAnyScope, bad)
            .await
            .unwrap_err();
        assert!(matches!(err, ParError::RedirectUriNotRegistered(_)));
    }

    /// A pushed request nobody takes is swept by the next push once it has
    /// expired, so an abandoned sign-in does not stay until restart.
    #[tokio::test]
    async fn an_untaken_request_is_swept_once_it_has_expired() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let (abandoned, _) = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .unwrap();
        store
            .entries
            .write()
            .unwrap()
            .get_mut(&abandoned)
            .unwrap()
            .expires_at = OffsetDateTime::now_utc() - time::Duration::seconds(1);

        let (fresh, _) = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .unwrap();
        assert_eq!(store.len(), 1);
        assert!(store.take(&fresh).is_some());
        assert_eq!(store.len(), 0);
    }

    #[tokio::test]
    async fn a_valid_admitted_push_mints_a_one_shot_request_uri_and_a_decision() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let (uri, ttl) = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .unwrap();
        assert!(uri.starts_with(REQUEST_URI_PREFIX));
        assert_eq!(ttl, REQUEST_URI_TTL);

        let record = fixture.decisions.get(&uri).expect("a decision was minted");
        assert_eq!(record.account, AGENT);
        assert_eq!(record.client_origin, "https://client.example");
        assert_eq!(record.verdict, Verdict::Allow);
        assert!(record.first_time);
        assert!(record.token.is_some());
        // The client's own copy is kept, and is not on the wire.
        assert_eq!(record.client_self.name.as_deref(), Some("Krillbridge"));
        assert!(!record.to_json().to_string().contains("Krillbridge"));

        let pushed = store.take(&uri).expect("request is there once");
        assert_eq!(pushed.redirect_uri, "https://client.example/cb");
        assert_eq!(pushed.did, AGENT);
        // One-shot: gone the second time.
        assert!(store.take(&uri).is_none());
    }

    #[tokio::test]
    async fn plain_code_challenge_method_is_refused() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let mut bad = request();
        bad.code_challenge_method = Some("plain".to_owned());
        let err = push(&fixture, &store, &GrantAnyScope, bad)
            .await
            .unwrap_err();
        assert!(matches!(err, ParError::UnsupportedCodeChallengeMethod(_)));
    }

    /// Every S256 challenge is 43 base64url characters, whatever verifier it
    /// digests, so a challenge of any other shape can never be met.
    #[tokio::test]
    async fn a_challenge_that_is_not_s256_shaped_is_refused() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        for challenge in [
            "challenge".to_owned(),
            format!("{CHALLENGE}A"),
            CHALLENGE.replace('-', "+"),
            "a".repeat(2 * 1024 * 1024),
        ] {
            let mut bad = request();
            bad.code_challenge = Some(challenge);
            assert_eq!(
                push(&fixture, &store, &GrantAnyScope, bad)
                    .await
                    .unwrap_err(),
                ParError::MalformedCodeChallenge
            );
        }
        assert!(push(&fixture, &store, &GrantAnyScope, request())
            .await
            .is_ok());
    }

    /// `state` and `login_hint` are kept with the pushed request, so both
    /// are bounded before anything else is spent on it. A value at the
    /// bound still passes.
    #[tokio::test]
    async fn an_oversized_state_or_login_hint_is_refused() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();

        let mut longest = request();
        longest.state = Some("s".repeat(MAX_STATE_BYTES));
        assert!(push(&fixture, &store, &GrantAnyScope, longest)
            .await
            .is_ok());

        let mut bad = request();
        bad.state = Some("s".repeat(MAX_STATE_BYTES + 1));
        assert_eq!(
            push(&fixture, &store, &GrantAnyScope, bad)
                .await
                .unwrap_err(),
            ParError::TooLong {
                field: "state",
                max: MAX_STATE_BYTES
            }
        );

        let mut bad = request();
        bad.login_hint = Some(format!("{AGENT}:{}", "9".repeat(MAX_LOGIN_HINT_BYTES)));
        assert_eq!(
            push(&fixture, &store, &GrantAnyScope, bad)
                .await
                .unwrap_err(),
            ParError::TooLong {
                field: "login_hint",
                max: MAX_LOGIN_HINT_BYTES
            }
        );
    }

    /// A request naming no `login_hint` at all is malformed the same way a
    /// request naming no `scope` is — every account alike would refuse it,
    /// so saying so is not an enumeration hole.
    #[tokio::test]
    async fn a_missing_login_hint_is_a_missing_parameter() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let mut missing = request();
        missing.login_hint = None;
        assert_eq!(
            push(&fixture, &store, &GrantAnyScope, missing)
                .await
                .unwrap_err(),
            ParError::MissingParameter("login_hint")
        );
    }

    /// A `login_hint` naming nobody gets the same `201` a real account
    /// gets — see [`push_for_unresolved_hint`] — and nothing is stored
    /// anywhere under it: not in [`ParStore`], so there is nothing for
    /// `authorize` to redeem, and not in [`DecisionStore`] either, so no
    /// bound anywhere — this hint's own or the server's total — is any
    /// different after this push than before it.
    #[tokio::test]
    async fn an_unresolvable_hint_is_accepted_and_stores_nothing_at_all() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let mut bad = request();
        bad.login_hint = Some("did:web:nobody.example".to_owned());
        let (uri, ttl) = push(&fixture, &store, &GrantAnyScope, bad)
            .await
            .expect("an unresolvable hint is still a well-formed push");
        assert_eq!(ttl, REQUEST_URI_TTL);
        assert!(uri.starts_with(REQUEST_URI_PREFIX));

        // Nothing for `authorize` to take, and no record for anything else
        // to read back.
        assert!(store.take(&uri).is_none());
        assert!(fixture.decisions.get(&uri).is_none());
    }

    /// Round 2's fix bounded a nonexistent hint the same way a real
    /// account is bounded, keyed on the raw `login_hint` string. That
    /// reopened the hole: a real account's bucket is one DID however it is
    /// spelled (`resolve_identifier` folds a handle and its DID together),
    /// while every distinct spelling of a nonexistent one is its own
    /// bucket. Pushing many invented hints must never accumulate anywhere,
    /// so the server's total bound — the one shared, identity-independent
    /// resource — cannot be filled by them either.
    #[tokio::test]
    async fn many_invented_hints_leave_the_store_empty() {
        let fixture = Fixture {
            decisions: MemoryDecisionStore::with_bounds(1, 512),
            ..Fixture::new()
        };
        let store = MemoryParStore::new();
        for n in 0..512 {
            let mut bad = request();
            bad.login_hint = Some(format!("did:web:nobody-{n}.example"));
            push(&fixture, &store, &GrantAnyScope, bad)
                .await
                .expect("an unresolvable hint is always well-formed");
        }
        // If any of the above had cost a real slot, this would answer
        // `TooMany` instead of admitting the one real account asking.
        let (uri, _) = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .expect("512 invented hints left the store's total bound untouched");
        assert!(fixture.decisions.get(&uri).is_some());
    }

    /// The one `403` this endpoint answers with is server load, not
    /// account state: a full store refuses a real account and an
    /// unresolvable hint with the identical body.
    #[tokio::test]
    async fn a_full_store_refuses_a_real_account_and_an_unresolvable_hint_alike() {
        let fixture = Fixture {
            decisions: MemoryDecisionStore::with_bounds(10, 1),
            ..Fixture::new()
        };
        let store = MemoryParStore::new();
        push(&fixture, &store, &GrantAnyScope, request())
            .await
            .expect("the one total slot admits the first push");

        let from_real = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .unwrap_err();
        let mut bad = request();
        bad.login_hint = Some("did:web:nobody.example".to_owned());
        let from_unresolved = push(&fixture, &store, &GrantAnyScope, bad)
            .await
            .unwrap_err();

        assert!(matches!(from_real, ParError::TooManyPending(_)));
        assert_eq!(from_real, from_unresolved);
    }

    struct EmptyCeiling;
    impl ScopePolicy for EmptyCeiling {
        fn ceiling(&self, _did: &str, _client_id: &str) -> ScopeSet {
            ScopeSet::default()
        }
    }

    /// A refused request still gets a record — the agent is told it was
    /// asked and told the answer — and gets no reference, so there is
    /// nothing to approve.
    #[tokio::test]
    async fn a_request_outside_the_ceiling_is_recorded_as_a_denial_with_no_token() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let (uri, _) = push(&fixture, &store, &EmptyCeiling, request())
            .await
            .unwrap();
        let record = fixture.decisions.get(&uri).expect("a decision was minted");
        assert!(record.verdict.is_deny());
        assert!(record.token.is_none());
    }

    struct BaseScopeOnly;
    impl ScopePolicy for BaseScopeOnly {
        fn ceiling(&self, _did: &str, _client_id: &str) -> ScopeSet {
            ScopeSet::parse("atproto").unwrap()
        }
    }

    #[tokio::test]
    async fn a_request_the_ceiling_shrinks_is_recorded_as_a_narrowing() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let mut wide = request();
        wide.scope = Some("atproto repo:app.bsky.feed.post".to_owned());
        let (uri, _) = push(&fixture, &store, &BaseScopeOnly, wide).await.unwrap();
        let record = fixture.decisions.get(&uri).expect("a decision was minted");
        let Verdict::Narrow { granted, cut, .. } = &record.verdict else {
            panic!("expected a narrowing, got {:?}", record.verdict);
        };
        assert_eq!(granted.to_string(), "atproto");
        assert_eq!(cut.to_string(), "repo:app.bsky.feed.post");
        assert!(record.token.is_some());
    }

    /// **What the agent approves is what the login is issued at.** The
    /// record shows `granted` and names `cut`; the reference the agent
    /// answers with carries `granted` alone, so a ceiling loosened after the
    /// approval cannot turn a cut atom into one the login holds.
    #[tokio::test]
    async fn a_narrowed_push_is_approvable_only_at_the_granted_scope() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let mut wide = request();
        wide.scope = Some("atproto repo:app.bsky.feed.post".to_owned());
        let (uri, _) = push(&fixture, &store, &BaseScopeOnly, wide).await.unwrap();
        let record = fixture.decisions.get(&uri).expect("a decision was minted");
        let reference = record.token.clone().expect("a narrowing is approvable");

        let pending = fixture
            .consent
            .consume(&reference)
            .expect("the reference is live");
        assert_eq!(pending.granted_scope.to_string(), "atproto");
        assert_eq!(
            record.requested.to_string(),
            "atproto repo:app.bsky.feed.post"
        );
    }

    /// `plan/scope-policy.md`'s "log every scopedown", and the shape the
    /// row takes: a narrowing says it narrowed, names the ceiling rule that
    /// did it, is attributed to the ceiling rather than to a policy version
    /// no policy tree ever produced, and carries what it cut as atoms an
    /// operator can count rather than prose they would have to parse.
    #[tokio::test]
    async fn a_narrowing_is_logged_as_a_narrowing_by_the_ceiling() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let mut wide = request();
        wide.scope = Some("atproto repo:app.bsky.feed.post".to_owned());
        push(&fixture, &store, &BaseScopeOnly, wide).await.unwrap();

        let rows = fixture.rows();
        assert_eq!(rows.len(), 1, "one narrowing, one row: {rows:?}");
        assert_eq!(
            rows[0].verdict,
            didbot_pds::evaluation_log::Verdict::Narrow,
            "a narrowing must not claim the request was refused"
        );
        assert_eq!(rows[0].account, AGENT);
        assert_eq!(rows[0].policies_fired, vec![CEILING_RULE.to_owned()]);
        assert_eq!(rows[0].policy_version, CEILING_VERSION);
        assert_eq!(rows[0].cut, vec!["repo:app.bsky.feed.post".to_owned()]);
        assert_eq!(
            rows[0].reason,
            format!("scope narrowed by `{CEILING_RULE}`")
        );
    }

    /// Two narrowings of one account by one rule that cut different atoms
    /// are two rows: the cut is part of what makes a denial the same denial,
    /// so collapsing them would lose one of the two answers.
    #[tokio::test]
    async fn two_narrowings_that_cut_different_atoms_are_two_rows() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        for scope in [
            "atproto repo:app.bsky.feed.post",
            "atproto repo:app.bsky.actor.profile",
        ] {
            let mut wide = request();
            wide.scope = Some(scope.to_owned());
            push(&fixture, &store, &BaseScopeOnly, wide).await.unwrap();
        }

        let mut cuts: Vec<Vec<String>> = fixture.rows().into_iter().map(|row| row.cut).collect();
        cuts.sort();
        assert_eq!(
            cuts,
            vec![
                vec!["repo:app.bsky.actor.profile".to_owned()],
                vec!["repo:app.bsky.feed.post".to_owned()],
            ]
        );
    }

    /// One policy, supplied the way a deployment's startup source supplies
    /// one, denying every grant from `client_id_pattern`.
    fn tree_denying_client(client_id_pattern: &str) -> didbot_policy_records::ParsedPolicy {
        let applicability = didbot_policy_records::Applicability::Grant {
            client_ids: didbot_policy_records::Scope::Any,
        };
        didbot_policy_records::ParsedPolicy {
            observes: applicability.clone(),
            applicability,
            staleness: didbot_policy_records::Staleness::Linearized,
            language: "regex".to_owned(),
            predicate: serde_json::json!({
                "kind": "denyClient",
                "clientId": client_id_pattern,
                "reason": "this client is not admitted here",
            }),
            revision: didbot_policy_records::Revision(1),
            lookups: Vec::new(),
        }
    }

    /// A policy's refusal is written down once, by the gate that made it —
    /// `TreePolicyGate::judge` records as it judges — and PAR writes no
    /// second row saying the same thing in different words. `push` asks the
    /// gate exactly once regardless of what it answers (see
    /// `push_authorization_request`'s own doc), so this holds whether or not
    /// a decision record ever exists to write a second row about.
    #[tokio::test]
    async fn a_policy_denial_is_logged_once_by_the_gate() {
        let fixture = Fixture::denying_client("^https://client\\.example/.*$");
        let store = MemoryParStore::new();
        assert_eq!(
            push(&fixture, &store, &GrantAnyScope, request())
                .await
                .unwrap_err(),
            ParError::ClientRefused
        );

        let rows = fixture.rows();
        assert_eq!(rows.len(), 1, "one refusal, one row: {rows:?}");
        assert_eq!(rows[0].verdict, didbot_pds::evaluation_log::Verdict::Reject);
        assert_eq!(
            rows[0].policies_fired,
            vec!["0".to_owned()],
            "the gate names the policy that fired"
        );
    }

    /// A gate that refuses the grant. `plan/policy.md`: policy denies, and
    /// its answer beats whatever the ceiling would have reached — so
    /// thoroughly that the ceiling is never consulted at all.
    struct RefuseEveryGrant;
    impl didbot_pds::policy::PolicyGate for RefuseEveryGrant {
        fn judge(&self, _subject: &didbot_pds::policy::Subject<'_>) -> didbot_pds::policy::Outcome {
            didbot_pds::policy::Outcome::Reject {
                reason: "this client is not admitted".to_owned(),
            }
        }
        fn version(&self) -> didbot_pds::policy::PolicyVersion {
            didbot_pds::policy::PolicyVersion("test-1".to_owned())
        }
    }

    #[tokio::test]
    async fn a_policy_refusal_beats_a_ceiling_that_would_have_allowed() {
        let fixture = Fixture::with_gate(Box::new(RefuseEveryGrant));
        let store = MemoryParStore::new();
        assert_eq!(
            push(&fixture, &store, &GrantAnyScope, request())
                .await
                .unwrap_err(),
            ParError::ClientRefused,
            "the gate's own reason and rule never reach the wire"
        );
    }

    /// A push from the profile's local client, `client_id` and the
    /// `redirect_uri` its listener is on, to a deployment that builds its
    /// document.
    async fn push_local(
        fixture: &Fixture,
        store: &dyn ParStore,
        client_id: &str,
        redirect_uri: &str,
    ) -> Result<(String, Duration), ParError> {
        let mut local = request();
        local.client_id = client_id.to_owned();
        local.redirect_uri = Some(redirect_uri.to_owned());
        let mut seams = fixture.seams(&GrantAnyScope);
        seams.loopback_clients = crate::oauth::client_metadata::LoopbackClients::Fetched;
        push_authorization_request(
            &didbot_http::client(),
            &MemoryClientMetadataCache::new(),
            store,
            local,
            &resolves,
            &seams,
        )
        .await
    }

    /// **The impersonation this closes.** Any process on the client's
    /// machine can send a local client's `client_id` and get the same
    /// document, so a copy used to read "seen before" once the original had
    /// been approved. A client at the local origin is never seen before.
    #[tokio::test]
    async fn a_second_process_sending_a_local_clients_id_is_never_seen_before() {
        let fixture = Fixture::new();
        let store = MemoryParStore::new();
        let client_id =
            "http://localhost?redirect_uri=http%3A%2F%2F127.0.0.1%2Fcallback&scope=atproto";

        let (real, _) = push_local(
            &fixture,
            &store,
            client_id,
            "http://127.0.0.1:4567/callback",
        )
        .await
        .unwrap();
        fixture.decisions.settle(&real, DecisionState::Approved);

        let (copy, _) = push_local(
            &fixture,
            &store,
            client_id,
            "http://127.0.0.1:6000/callback",
        )
        .await
        .unwrap();
        let copy = fixture.decisions.get(&copy).unwrap().to_json();
        assert_eq!(copy["client"]["origin"], "http://localhost");
        assert_eq!(copy["client"]["firstTime"], true);
    }

    /// **The bypass this closes.** A rule against a local client used to
    /// match its `client_id` as written, so another spelling of the same
    /// client walked past it. Every spelling is now judged as the one
    /// `ClientId` writes, and a parameter the profile does not define is
    /// refused outright.
    #[tokio::test]
    async fn a_rule_against_a_local_client_holds_for_every_spelling_of_it() {
        let fixture = Fixture::denying_client(
            "^http://localhost/\\?redirect_uri=http%3A%2F%2F127\\.0\\.0\\.1%2Fcallback&scope=atproto$",
        );
        let store = MemoryParStore::new();
        let redirect = "http://127.0.0.1:4567/callback";
        for spelling in [
            "http://localhost?redirect_uri=http%3A%2F%2F127.0.0.1%2Fcallback&scope=atproto",
            "http://localhost/?redirect_uri=http%3A%2F%2F127.0.0.1%2Fcallback&scope=atproto",
            "http://localhost/?scope=atproto&redirect_uri=http%3A%2F%2F127.0.0.1%3A4567%2Fcallback",
            "http://localhost?redirect_uri=http://127.0.0.1:9/callback&scope=atproto+atproto",
            "HTTP://LOCALHOST:80/?redirect_uri=http%3A%2F%2F127.0.0.1%2Fcallback&scope=atproto",
        ] {
            assert_eq!(
                push_local(&fixture, &store, spelling, redirect).await,
                Err(ParError::ClientRefused),
                "{spelling}"
            );
        }
        let junk =
            "http://localhost/?redirect_uri=http%3A%2F%2F127.0.0.1%2Fcallback&scope=atproto&a";
        assert!(matches!(
            push_local(&fixture, &store, junk, redirect).await,
            Err(ParError::ClientId(_))
        ));

        // The rule names one client, not every local one.
        push_local(
            &fixture,
            &store,
            "http://localhost?redirect_uri=http%3A%2F%2F127.0.0.1%2Fother&scope=atproto",
            "http://127.0.0.1:4567/other",
        )
        .await
        .expect("another local client is not the one the rule names");
    }

    /// **The lockout this closes.** The bound `plan/oauth.md` needs because
    /// PAR is unauthenticated used to drop the newest push, so a stranger
    /// naming an account in `login_hint` could keep that account's own
    /// sign-ins from ever landing. Past the bound the account's oldest
    /// undecided record makes way instead: the newest push is the one that
    /// survives, and the displaced one leaves nothing to redeem.
    #[tokio::test]
    async fn a_push_past_the_accounts_own_bound_displaces_the_oldest_not_the_newest() {
        let fixture = Fixture {
            decisions: MemoryDecisionStore::with_bounds(1, 100),
            ..Fixture::new()
        };
        let store = MemoryParStore::new();
        let (first, _) = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .unwrap();
        // A `403` here would depend on this account already being loaded —
        // exactly the fact an unresolvable hint must not be able to probe
        // for by comparison. It answers success instead.
        let (second, _) = push(&fixture, &store, &GrantAnyScope, request())
            .await
            .expect("a full per-account bound displaces, it does not refuse");
        assert_ne!(first, second);
        // The newest push is answerable; the displaced one is gone from both
        // stores, exactly as an expired one would be.
        assert!(fixture.decisions.get(&second).is_some());
        assert!(store.take(&second).is_some());
        assert!(fixture.decisions.get(&first).is_none());
        assert!(store.take(&first).is_none());
    }

    /// **The consent-store leak this fix closes.** `consent::begin` mints a
    /// reference before `DecisionStore::put` is even asked, since a
    /// reference has to name the record it approves. Undoing a record
    /// without also discarding that reference left one live
    /// `PendingConsent` behind per push past the bound, in a store with no
    /// other way to remove one short of it being presented — which nothing
    /// ever does for a record that no longer exists. This endpoint is
    /// unauthenticated, so that was unbounded memory growth one stranger
    /// could trigger against any real account merely by naming it enough.
    /// One record in, one record out: the store does not grow.
    #[tokio::test]
    async fn a_displacing_push_leaves_the_consent_store_at_its_previous_size() {
        let fixture = Fixture {
            decisions: MemoryDecisionStore::with_bounds(1, 100),
            ..Fixture::new()
        };
        let store = MemoryParStore::new();
        push(&fixture, &store, &GrantAnyScope, request())
            .await
            .expect("the first push fits under the bound");
        let after_first = fixture.consent.len();
        assert_eq!(after_first, 1, "the first push's own reference is real");

        for _ in 0..5 {
            push(&fixture, &store, &GrantAnyScope, request())
                .await
                .expect("displaced past the bound, not refused");
            assert_eq!(
                fixture.consent.len(),
                after_first,
                "a push past the bound must not grow the consent store"
            );
        }
    }

    /// Round 2's exact reopening, checked directly against `resolve`: eight
    /// pushes naming a real account (however addressed — this fixture's
    /// `resolve` always answers `AGENT` for anything, standing in for
    /// `resolve_identifier` folding a handle, its DID, and a redundant
    /// `:port` into one) and a ninth naming nobody must every one answer
    /// `201`, with no ninth-push aliasing effect for a caller to time or
    /// compare against a nonexistent hint's own ninth push.
    #[tokio::test]
    async fn aliased_spellings_of_a_real_account_never_produce_a_refusal_either() {
        let fixture = Fixture {
            decisions: MemoryDecisionStore::with_bounds(1, 100),
            ..Fixture::new()
        };
        let store = MemoryParStore::new();
        for _ in 0..8 {
            push(&fixture, &store, &GrantAnyScope, request())
                .await
                .expect("every spelling of a real account still answers 201");
        }
        let mut unresolved = request();
        unresolved.login_hint = Some("did:web:nobody.example".to_owned());
        push(&fixture, &store, &GrantAnyScope, unresolved)
            .await
            .expect("a nonexistent hint answers 201 too, at every step");
    }
}