acme-proxy 0.6.0

//! Shared integration-test harness: an in-memory app builder, response helpers,
//! and EC/RSA JWS signers producing bodies in the exact shape the `AcmeRequest`
//! extractor expects.
//!
//! Each `tests/*.rs` file is its own crate that includes this module, so not
//! every crate uses every helper — hence the crate-wide `dead_code` allow, and
//! the `unused_imports` one for the re-exports only some of them need.
//!
//! ## Building an app
//!
//! [`test_app_full`] is the single place the ACME app is built: one profile,
//! `default`, through the real `build_app`, so paths carry the endpoint prefix
//! ([`p`] builds one). Every other `test_app_*` and `test_admin_app*` is a
//! variant of it, and `one_profile` is shared with the admin builders so the
//! two cannot drift into mounting different endpoints.
//!
//! - **A profile is served from its signer's read side**, never the backend.
//!   [`read_side`] is `signer.info()`, except for a local CA whose revocation
//!   state lives outside the app's database: its route becomes `Delegated`,
//!   since a ledger row in the app's database would be one its CRL never sees.
//!   `test_app_with_db` and the admin apps build their CA over the app's own
//!   database, so they exercise the route production takes.
//! - **Every builder calls [`met_by_a_worker`] first** — a local CA's first
//!   CRL stored, which the read side never signs.
//! - **Every app runs a real worker** (`spawn_worker_runner`): the real
//!   validation, issuance, revocation and CRL jobs over the real runner, at a
//!   10 ms poll with no backoff. Its shutdown sender is leaked on purpose, since
//!   the runner treats a closed channel as a signal to stop.
//! - A fixture wrapping a signer overrides `info()` (and, around a CA,
//!   `crl_refresher()`); the read methods are not on `SignerBackend`.
//!
//! ## Driving it
//!
//! [`acme`] is the ACME ladder — `post`, `register`, `new_order`,
//! `drive_to_ready`, `finalize`, `issue_certificate`, `trigger` and the rest.
//! Use it rather than writing another copy.
//!
//! - **`trigger` and `finalize` decide nothing.** Both answer `processing`;
//!   poll with `await_challenge`/`await_order` (or the `*_and_settle` pairs),
//!   and a CRL with `await_crl_listing`.
//! - **Assert with `assert_problem`, not a bare status.** `unauthorized` and
//!   `rejectedIdentifier` are both `403`; a refusal that keeps its status and
//!   changes its `type` is a different answer to the client.
//! - **Import `TestSigner`** wherever a test calls `signer.sign(…)`; the
//!   methods are its defaults.
//! - `oneshot` has no socket, so `send_from`/`post_from`/`fetch_nonce_from`
//!   insert the `ConnectInfo<SocketAddr>` by hand. That is how a test gives the
//!   filters a client address.
//! - The admin builders mount an *inactive* filter policy, except
//!   `test_admin_app_logged_in_with_filter`, for the surfaces whose subject is
//!   the policy itself.
#![allow(dead_code, unused_imports)]

use std::net::SocketAddr;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use acme_proxy_jobs::auditor::Auditor;
// Re-exported for the suites that build an `Account` directly.
pub use acme_proxy_core::audit::ClientContext;
use acme_proxy_protocol::profile::Profile;
use acme_proxy_protocol::profile::ProfileParts;
use acme_proxy_protocol::router::build_app;

use acme_proxy_admin::admin::password::PasswordContext;
use acme_proxy_core::client::ProxyPolicy;
use acme_proxy_core::config::Config;
use acme_proxy_core::config::JobsConfig;
use acme_proxy_core::identifier::Identifier;
use acme_proxy_jobs::jobs::JobQueue;
use acme_proxy_jobs::jobs::JobRegistry;
pub use acme_proxy_jobs::metrics::Metrics;
use acme_proxy_jobs::notify::BackendSlot;
use acme_proxy_jobs::notify::NotifyBackend;
use acme_proxy_jobs::notify::NotifyDispatcher;
use acme_proxy_jobs::notify::NotifyError;
use acme_proxy_jobs::notify::NotifyEvent;
use acme_proxy_jobs::notify::NotifyJob;
use acme_proxy_net::challenge::ChallengeError;
use acme_proxy_net::challenge::ChallengeRegistry;
use acme_proxy_net::challenge::ChallengeValidator;
use acme_proxy_net::challenge::ValidationContext;
use acme_proxy_policy::filter::ConnectionContext;
use acme_proxy_policy::filter::FilterPolicy;
use acme_proxy_policy::filter::IdentifierContext;
use acme_proxy_policy::filter::Stage;
use acme_proxy_policy::filter::expr::Condition;
use acme_proxy_policy::filter::policy::Check;
use acme_proxy_policy::filter::policy::Effect;
use acme_proxy_policy::filter::policy::Mode;
use acme_proxy_policy::filter::policy::Rule;
use acme_proxy_policy::filter::policy::StageSet;
use acme_proxy_policy::filter::policy::Verdict;
use acme_proxy_protocol::acme::issue::SignerIssueJob;
use acme_proxy_protocol::acme::revoke::SignerRevokeJob;
use acme_proxy_protocol::acme::validate::ChallengeValidateJob;
use acme_proxy_signer::CrlRefresher;
use acme_proxy_signer::Http01TokenStore;
use acme_proxy_signer::IssueOutcome;
use acme_proxy_signer::RenewalWindow;
use acme_proxy_signer::RequestedValidity;
use acme_proxy_signer::RevocationRoute;
use acme_proxy_signer::SignerBackend;
use acme_proxy_signer::SignerError;
use acme_proxy_signer::SignerInfo;
use acme_proxy_signer::local_ca::LocalCa;
use acme_proxy_signer::local_ca::sweep::CrlRegenerateJob;
use acme_proxy_signer::relay::http01::DbTokenStore;
use acme_proxy_store::db::Database;
use async_trait::async_trait;
use axum::Router;
use axum::body::Body;
use axum::extract::ConnectInfo;
use axum::http::{Method, Request, StatusCode, header};
use axum::response::Response;
use base64::prelude::*;
use http_body_util::BodyExt;
use ring::hmac;
use ring::rand::SystemRandom;
use ring::signature::{self, EcdsaKeyPair, KeyPair, RsaKeyPair};
use serde_json::{Value, json};
use simple_asn1::ASN1Block;
use tower::ServiceExt;

/// Installs a `tracing` subscriber once per test binary.
///
/// Without a subscriber, `LevelFilter::current()` is `OFF` and `tracing`'s
/// `event!` short-circuits before evaluating its field expressions — they sit
/// inside its `if enabled` block. So every `error = %error` / `id = ?id` in the
/// server runs its `Display`/`Debug` impl for the first time *in production*.
/// This makes the suite exercise them instead, which is what would catch a
/// field expression that panics or is accidentally expensive.
///
/// Output goes to a sink: the point is to run the formatting, not to print it.
/// `try_init` (not `init`) because each `tests/*.rs` is its own binary but a
/// binary may call this from several tests at once.
///
/// Note this is *not* a coverage lever, though it looks like one: measured
/// before and after, it moved the line total by 0.03 points. Handlers carrying
/// `#[instrument]` report oddly under `llvm-cov` for an unrelated reason — the
/// attribute moves the body into a generated `async` block, so the visible
/// signature lines read as 0 hits and the body lines carry no region at all,
/// however thoroughly the handler is tested.
fn init_tracing() {
    static ONCE: std::sync::Once = std::sync::Once::new();
    ONCE.call_once(|| {
        let _ = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::TRACE)
            .with_writer(std::io::sink)
            .try_init();
    });
}

/// Builds a real PKCS#10 CSR for a single `dns` SAN and returns it base64url-DER
/// encoded (the shape a finalize payload's `csr` field carries). Uses `rcgen`
/// (a normal crate dependency, available to integration tests).
pub fn make_csr(dns: &str) -> String {
    make_csr_for(&[dns])
}

/// [`make_csr`] over several names, for a multi-identifier order.
///
/// `tests/security.rs` had its own verbatim copy of this; a CSR builder is
/// exactly the sort of thing that ends up subtly different in two places (one
/// setting a common name, the other not) and then tests two different things
/// while reading as though it tests one.
pub fn make_csr_for(names: &[&str]) -> String {
    let key_pair = rcgen::KeyPair::generate().unwrap();
    let params = rcgen::CertificateParams::new(
        names
            .iter()
            .map(std::string::ToString::to_string)
            .collect::<Vec<_>>(),
    )
    .unwrap();
    let csr = params.serialize_request(&key_pair).unwrap();
    BASE64_URL_SAFE_NO_PAD.encode(csr.der())
}

/// Like [`make_csr`], but with extra subject alternative names appended — so a
/// test can build the CSR that smuggles an IP address alongside its DNS name.
pub fn make_csr_with_sans(dns: &str, extra: Vec<rcgen::SanType>) -> String {
    let key_pair = rcgen::KeyPair::generate().unwrap();
    let mut params = rcgen::CertificateParams::new(vec![dns.to_string()]).unwrap();
    params.subject_alt_names.extend(extra);
    let csr = params.serialize_request(&key_pair).unwrap();
    BASE64_URL_SAFE_NO_PAD.encode(csr.der())
}

/// Builds a base64url-DER CSR for `dns` from an EC key pair a test can *also*
/// sign an ACME JWS with — the shape "revoke via the certificate's own
/// keypair" needs (RFC 8555 §7.6's second authorization case, no account
/// involved at all). The same PKCS8 document backs two independent objects:
/// a `ring` `EcdsaKeyPair` (fixed-format ES256 signing, wrapped as
/// [`EcSigner`] so [`TestSigner::sign`] can build the embedded-`jwk` revoke
/// JWS) and an `rcgen::KeyPair` (ASN.1-format signing, used only to
/// self-sign the CSR) — both derive the *same* public key, so the leaf
/// certificate later issued from this CSR carries the exact DER-SPKI
/// [`EcSigner::jwk`] encodes.
pub fn make_csr_and_keypair(dns: &str) -> (String, EcSigner) {
    let rng = SystemRandom::new();
    let pkcs8 =
        EcdsaKeyPair::generate_pkcs8(&signature::ECDSA_P256_SHA256_FIXED_SIGNING, &rng).unwrap();
    let signer = EcSigner::from_pkcs8(pkcs8.as_ref());

    let rcgen_key = rcgen::KeyPair::from_pkcs8_der_and_sign_algo(
        &rustls_pki_types::PrivatePkcs8KeyDer::from(pkcs8.as_ref().to_vec()),
        &rcgen::PKCS_ECDSA_P256_SHA256,
    )
    .unwrap();
    let params = rcgen::CertificateParams::new(vec![dns.to_string()]).unwrap();
    let csr = params.serialize_request(&rcgen_key).unwrap();

    (BASE64_URL_SAFE_NO_PAD.encode(csr.der()), signer)
}

/// The leaf out of a `leaf + CA` PEM chain, as DER — the first `CERTIFICATE`
/// block.
pub fn first_certificate(chain: &str) -> Vec<u8> {
    acme_proxy_core::cert::leaf_der_from_chain(chain).unwrap()
}

/// The server-level base URL — `server.base_url`, naming the process rather
/// than any one endpoint. Nothing ACME is served here.
pub const HOST: &str = "http://localhost:3000";

/// The profile every helper below mounts.
pub const PROFILE: &str = "default";

/// Where that profile is mounted: `/profile/<name>`, the namespace `build_app`
/// reserves for ACME endpoints.
pub const PREFIX: &str = "/profile/default";

/// The base URL for that profile — what its `directory` advertises, and what
/// every JWS `url` must name (RFC 8555 §6.4).
pub const BASE: &str = "http://localhost:3000/profile/default";

/// A route path under the test profile: `p("/newOrder")` →
/// `/profile/default/newOrder`.
///
/// An absolute URL a response handed back (a `Location` header, an
/// `authorizations` entry) becomes a request path by stripping [`HOST`], not
/// [`BASE`] — the prefix has to stay.
#[must_use]
pub fn p(path: &str) -> String {
    format!("{PREFIX}{path}")
}

/// Builds the full app backed by a throwaway in-memory database and the
/// default configuration.
pub async fn test_app() -> Router {
    test_app_with_db().await.0
}

/// Like [`test_app`], but also returns the backing database so a test can, e.g.,
/// close its pool to exercise DB-failure paths. The signer is an in-memory local
/// CA so the suite stays disk- and network-free.
pub async fn test_app_with_db() -> (Router, Arc<Database>) {
    let database = Arc::new(Database::connect_in_memory().await.unwrap());
    // Over the app's own database, as a real CA is: a revocation is then a
    // ledger row the worker signs into the CRL — the production path.
    let signer = Arc::new(
        LocalCa::generate_in_memory("ecdsa-p256", 90, database.clone())
            .expect("an in-memory CA is always available"),
    );
    (test_app_over(database.clone(), signer).await, database)
}

/// An in-memory local CA over its own throwaway database.
///
/// Its revocations and CRL live in that database, which nothing else here
/// reads: a test that needs two CAs — or a CA and an app — to share revocation
/// state builds the CA over the app's database itself.
pub async fn memory_ca() -> LocalCa {
    LocalCa::generate_in_memory(
        "ecdsa-p256",
        90,
        Arc::new(Database::connect_in_memory().await.unwrap()),
    )
    .expect("an in-memory CA is always available")
}

/// Builds the full router with a caller-supplied signer backend (and a throwaway
/// in-memory database), returning both. Lets a test inject a failing signer to
/// drive the issuance-failure path.
pub async fn test_app_with_signer(signer: Arc<dyn SignerBackend>) -> (Router, Arc<Database>) {
    test_app_full(
        Config::default(),
        signer,
        Arc::new(FilterPolicy::default()),
        default_challenges(),
        no_notifications().await,
    )
    .await
}

/// Builds the full router with a caller-supplied filter chain, the default
/// config and an in-memory local CA.
pub async fn test_app_with_filter(filter: Arc<FilterPolicy>) -> (Router, Arc<Database>) {
    let signer = Arc::new(memory_ca().await);
    test_app_full(
        Config::default(),
        signer,
        filter,
        default_challenges(),
        no_notifications().await,
    )
    .await
}

/// Builds the full router with a caller-supplied challenge registry and config,
/// so a test can choose which types are offered and how they validate.
pub async fn test_app_with_challenges(
    config: Config,
    challenges: Arc<ChallengeRegistry>,
) -> (Router, Arc<Database>) {
    let signer = Arc::new(memory_ca().await);
    test_app_full(
        config,
        signer,
        Arc::new(FilterPolicy::default()),
        challenges,
        no_notifications().await,
    )
    .await
}

/// Builds the full router with a caller-supplied notify dispatcher — usually
/// wrapping a [`RecordingNotifyBackend`] — plus the default config, an
/// in-memory local CA, no filters and a bypassing `http-01` registry.
pub async fn test_app_with_notify(notify: Arc<NotifyDispatcher>) -> (Router, Arc<Database>) {
    let signer = Arc::new(memory_ca().await);
    test_app_full(
        Config::default(),
        signer,
        Arc::new(FilterPolicy::default()),
        default_challenges(),
        notify,
    )
    .await
}

/// The no-op dispatcher every `test_app_with_*` helper defaults to when a test
/// isn't asserting on notifications at all.
///
/// It still carries a queue — every dispatcher does, since a delivery is a job
/// row — but with no backends configured nothing is ever written to it, so the
/// throwaway database behind it stays empty.
pub async fn no_notifications() -> Arc<NotifyDispatcher> {
    Arc::new(NotifyDispatcher::disabled(test_job_queue().await))
}

/// A queue over its own in-memory database, for a dispatcher a test builds
/// directly rather than through `server::profile::build_all`.
pub async fn test_job_queue() -> JobQueue {
    let database = Arc::new(Database::connect_in_memory().await.unwrap());
    JobQueue::new(database, &JobsConfig::default())
}

/// A dispatcher over one [`RecordingNotifyBackend`], plus the runner that
/// actually performs what it queues.
///
/// `dispatch` writes a `notify_deliver` row and returns; without something
/// draining the queue nothing would ever reach the recorder. So every test
/// asserting "this handler notified" needs a runner, and this is it — the guard
/// stops it gracefully when the test ends.
pub struct NotifyHarness {
    pub dispatcher: Arc<NotifyDispatcher>,
    pub recorder: Arc<RecordingNotifyBackend>,
    shutdown: tokio::sync::watch::Sender<bool>,
}

impl NotifyHarness {
    /// A harness whose backend always delivers.
    pub async fn new() -> Self {
        Self::over(Arc::new(RecordingNotifyBackend::default())).await
    }

    /// A harness whose backend always fails, retryably.
    pub async fn failing() -> Self {
        Self::over(Arc::new(RecordingNotifyBackend::failing())).await
    }

    /// A harness keyed under [`acme_proxy_jobs::notify::ADMIN_DISPATCHER_KEY`], for
    /// the web-admin security events, plus the `Notifiers` handle to hand to
    /// `build_admin_app`.
    pub async fn admin() -> (Self, acme_proxy_jobs::notify::Notifiers) {
        let harness = Self::over_key(
            acme_proxy_jobs::notify::ADMIN_DISPATCHER_KEY,
            Arc::new(RecordingNotifyBackend::default()),
        )
        .await;
        let mut map = std::collections::HashMap::new();
        map.insert(
            acme_proxy_jobs::notify::ADMIN_DISPATCHER_KEY.to_string(),
            harness.dispatcher.clone(),
        );
        let notifiers: acme_proxy_jobs::notify::Notifiers = Arc::new(map).into();
        (harness, notifiers)
    }

    async fn over(recorder: Arc<RecordingNotifyBackend>) -> Self {
        Self::over_key(PROFILE, recorder).await
    }

    async fn over_key(key: &str, recorder: Arc<RecordingNotifyBackend>) -> Self {
        let every: Vec<String> = acme_proxy_core::config::ALL_NOTIFY_EVENTS
            .iter()
            .map(|kind| (*kind).to_string())
            .collect();
        // Retries off: these tests assert on the *first* delivery, and a backoff
        // between attempts would only make them wait. The retry behaviour itself
        // is covered inline, in `crates/jobs/src/notify/job.rs`.
        let config = JobsConfig {
            poll_interval_ms: 5,
            max_attempts: 1,
            retry_base_seconds: 0,
            ..JobsConfig::default()
        };
        let database = Arc::new(Database::connect_in_memory().await.unwrap());
        let queue = JobQueue::new(database, &config);

        let dispatcher = Arc::new(NotifyDispatcher::new(
            key,
            vec![BackendSlot::new("recording", recorder.clone(), &every)],
            queue.clone(),
        ));

        let mut registry = JobRegistry::new();
        let mut dispatchers = std::collections::HashMap::new();
        dispatchers.insert(key.to_string(), dispatcher.clone());
        registry
            .register(Arc::new(NotifyJob::new(Arc::new(dispatchers))))
            .unwrap();

        let (shutdown, receiver) = tokio::sync::watch::channel(false);
        acme_proxy_jobs::jobs::spawn_runner(queue, Arc::new(registry), &config, receiver);

        Self {
            dispatcher,
            recorder,
            shutdown,
        }
    }

    /// Waits until the recorder has seen at least `count` events, then returns
    /// them all.
    ///
    /// Polling rather than sleeping a fixed span: delivery is two hops now — the
    /// handler queues, the runner delivers — so a fixed sleep is a flake waiting
    /// to happen and a longer one is a slow suite.
    pub async fn recorded(&self, count: usize) -> Vec<NotifyEvent> {
        for _ in 0..400 {
            let events = self.recorder.events();
            if events.len() >= count {
                return events;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        panic!(
            "expected at least {count} notification(s), saw {:?}",
            self.recorder.events()
        );
    }

    /// Everything recorded once the queue has gone quiet — for the assertions
    /// that a *further* event was **not** dispatched, where waiting for a count
    /// that will never arrive is not an option.
    pub async fn settled(&self) -> Vec<NotifyEvent> {
        tokio::time::sleep(Duration::from_millis(150)).await;
        self.recorder.events()
    }
}

impl Drop for NotifyHarness {
    fn drop(&mut self) {
        let _ = self.shutdown.send(true);
    }
}

/// The bypassing `http-01`-only registry every other helper uses — the same
/// thing a server with no `[challenge]` section gets.
pub fn default_challenges() -> Arc<ChallengeRegistry> {
    Arc::new(ChallengeRegistry::default())
}

/// A registry offering `types` and validating them with `validator`.
pub fn challenges_with(
    types: &[&str],
    validators: Vec<Arc<dyn ChallengeValidator>>,
) -> Arc<ChallengeRegistry> {
    Arc::new(ChallengeRegistry::new(
        validators,
        types.iter().map(std::string::ToString::to_string).collect(),
        false,
        Duration::from_secs(5),
    ))
}

/// A registry offering `types` but validating nothing, as the default config does.
pub fn bypassing_challenges(types: &[&str]) -> Arc<ChallengeRegistry> {
    Arc::new(ChallengeRegistry::new(
        Vec::new(),
        types.iter().map(std::string::ToString::to_string).collect(),
        true,
        Duration::from_secs(5),
    ))
}

/// One endpoint in a multi-profile app, for `tests/profiles.rs`.
///
/// Defaults mirror [`test_app_full`]'s single profile — an in-memory local CA,
/// no filters, a bypassing `http-01` registry — so a test states only the
/// difference it is about.
pub struct TestProfile {
    pub name: &'static str,
    pub signer: Arc<dyn SignerBackend>,
    pub filter: Arc<FilterPolicy>,
    pub challenges: Arc<ChallengeRegistry>,
    pub eab: acme_proxy_core::config::EabConfig,
    pub meta: acme_proxy_core::config::MetaConfig,
    pub notify: Option<Arc<NotifyDispatcher>>,
}

impl TestProfile {
    /// A profile named `name`, with its **own** local CA — two of these issue
    /// certificates with different issuers, which is how a test proves which
    /// endpoint actually signed.
    pub async fn new(name: &'static str) -> Self {
        Self {
            name,
            signer: Arc::new(memory_ca().await),
            filter: Arc::new(FilterPolicy::default()),
            challenges: default_challenges(),
            eab: acme_proxy_core::config::EabConfig::default(),
            meta: acme_proxy_core::config::MetaConfig::default(),
            // Resolved when the app is built: a dispatcher needs a `JobQueue`
            // over the app's own database, which does not exist yet.
            notify: None,
        }
    }

    #[must_use]
    pub fn requiring_eab(mut self) -> Self {
        self.eab.enabled = true;
        self
    }

    #[must_use]
    pub fn with_filter(mut self, filter: Arc<FilterPolicy>) -> Self {
        self.filter = filter;
        self
    }

    #[must_use]
    pub fn with_notify(mut self, notify: Arc<NotifyDispatcher>) -> Self {
        self.notify = Some(notify);
        self
    }

    /// This endpoint's base URL — what its directory advertises and what every
    /// JWS `url` sent to it must name.
    #[must_use]
    pub fn base(name: &str) -> String {
        format!("{HOST}/profile/{name}")
    }

    /// A request path under this endpoint.
    #[must_use]
    pub fn path(name: &str, path: &str) -> String {
        format!("/profile/{name}{path}")
    }
}

/// Builds an app mounting several endpoints over one database — the real
/// multi-profile shape, through the real `build_app`.
pub async fn test_app_with_profiles(profiles: Vec<TestProfile>) -> (Router, Arc<Database>) {
    init_tracing();
    let config = Config::default();
    let database = Arc::new(Database::connect_in_memory().await.unwrap());
    // A `for` loop rather than `map`: resolving the default dispatcher opens a
    // database, so the body has to be able to await.
    let mut built = Vec::with_capacity(profiles.len());
    let mut backends = Vec::with_capacity(profiles.len());
    for profile in profiles {
        let notify = match profile.notify {
            Some(notify) => notify,
            None => no_notifications().await,
        };
        met_by_a_worker(profile.signer.as_ref()).await;
        backends.push((profile.name.to_string(), profile.signer.clone()));
        built.push(Arc::new(Profile::new(
            profile.name,
            &config.server.base_url,
            ProfileParts {
                signer_info: read_side(profile.signer.as_ref(), &database).await,
                filter: profile.filter,
                challenges: profile.challenges,
                order: config.order.clone(),
                eab: profile.eab,
                meta: profile.meta,
                notify,
            },
        )));
    }
    let jobs = spawn_worker_runner(&database, &built, &backends, test_auditor(database.clone()));
    let router = build_app(
        database.clone(),
        Arc::new(config),
        built,
        test_auditor(database.clone()),
        Arc::new(Metrics::new(database.clone())),
        jobs,
    );
    (router, database)
}

/// The queue the app enqueues into, with a runner draining it — the worker of
/// a real deployment.
///
/// Challenge validation and issuance are both queued work: `POST /chall/{id}`
/// claims the challenge and returns it `processing`, `finalize` claims the order
/// and returns it `processing`, and the outbound check and the signing happen in
/// the job runner. So a suite that triggers a challenge or finalizes an order
/// needs something to perform it, and this is the same `ChallengeValidateJob`,
/// `SignerIssueJob` and `spawn_runner` production uses — rather than reaching
/// into the queue by hand, so the suites go on exercising the real claim, lease
/// and settle path.
///
/// Paced for a test rather than a deployment: a 10 ms poll and no backoff, so
/// a backend failing `Internal` spends its attempt budget in milliseconds and
/// the order reaches `invalid` while the test is still polling for it.
///
/// `backends` is each profile's signer — what a `worker` holds and a
/// [`Profile`] never does.
///
/// `audit` is what the worker writes its rows through — the app's own, so an
/// issuance recorded here is counted into the same registry the app renders,
/// as the `worker` role's is in production.
///
/// The shutdown sender is deliberately leaked rather than dropped: the loop
/// treats a closed shutdown channel as a signal (`shutdown.changed()` on a
/// dropped sender returns `Err` at once, and the arm `break`s), so letting it
/// fall out of scope would stop the runner before it claimed anything. The task
/// is then left to the test's own runtime, which ends with the test — there is
/// nothing to drain gracefully, the database being in memory and going with it.
fn spawn_worker_runner(
    database: &Arc<Database>,
    profiles: &[Arc<Profile>],
    backends: &[(String, Arc<dyn SignerBackend>)],
    audit: Arc<Auditor>,
) -> JobQueue {
    let config = worker_jobs_config();
    let queue = JobQueue::new(database.clone(), &config);
    let mut registry = JobRegistry::new();
    registry
        .register(Arc::new(ChallengeValidateJob::new(
            database.clone(),
            audit.clone(),
            profiles
                .iter()
                .map(|profile| (profile.name.clone(), profile.clone()))
                .collect(),
        )))
        .expect("the validation handler registers");
    // The notifier each profile announces through, as the process-wide map a
    // worker reads them from. The sender is leaked for the shutdown sender's
    // reason: the map never changes here, and a dropped sender is harmless to
    // a `watch` receiver but pointless to keep.
    let (notifiers_tx, notifiers) = acme_proxy_jobs::notify::notifiers_channel(
        profiles
            .iter()
            .map(|profile| (profile.name.clone(), profile.notify.clone()))
            .collect(),
    );
    let revoke_notifiers = notifiers.clone();
    std::mem::forget(notifiers_tx);
    let audit_for_revocations = audit.clone();
    registry
        .register(Arc::new(SignerIssueJob::new(
            database.clone(),
            audit,
            backends.to_vec(),
            notifiers,
        )))
        .expect("the issuance handler registers");
    // Revocation, as a request makes it: a local CA's CRL is signed after its
    // ledger row by `local_ca_crl_regenerate`, and anything else is queued as
    // `signer_revoke` for the backend only a worker holds.
    registry
        .register(Arc::new(SignerRevokeJob::new(
            database.clone(),
            audit_for_revocations,
            backends.to_vec(),
            revoke_notifiers,
        )))
        .expect("the revocation handler registers");
    let mut refreshers: Vec<Arc<dyn CrlRefresher>> = Vec::new();
    for (_, backend) in backends {
        if let Some(refresher) = backend.crl_refresher()
            && !refreshers
                .iter()
                .any(|known| known.issuer() == refresher.issuer())
        {
            refreshers.push(refresher);
        }
    }
    if !refreshers.is_empty() {
        registry
            .register(Arc::new(CrlRegenerateJob::new(refreshers)))
            .expect("the CRL handler registers");
    }

    let (shutdown, rx) = tokio::sync::watch::channel(false);
    std::mem::forget(shutdown);

    acme_proxy_jobs::jobs::spawn_runner(queue.clone(), Arc::new(registry), &config, rx);
    queue
}

/// The `[jobs]` pacing [`spawn_worker_runner`] drains at.
fn worker_jobs_config() -> JobsConfig {
    JobsConfig {
        poll_interval_ms: 10,
        retry_base_seconds: 0,
        retry_max_seconds: 0,
        ..JobsConfig::default()
    }
}

/// The one place the app is actually constructed: every other `test_app_*`
/// helper delegates here, so a new `build_app`/`Profile` parameter is added in
/// one place rather than at each call site.
///
/// It builds the real thing — root routes plus one profile mounted under
/// [`PREFIX`] — so the suite exercises the mounting itself, not a router that
/// only exists in tests.
pub async fn test_app_full(
    config: Config,
    signer: Arc<dyn SignerBackend>,
    filter: Arc<FilterPolicy>,
    challenges: Arc<ChallengeRegistry>,
    notify: Arc<NotifyDispatcher>,
) -> (Router, Arc<Database>) {
    init_tracing();
    let database = Arc::new(Database::connect_in_memory().await.unwrap());
    let (profile, backends) =
        one_profile(&database, &config, signer, filter, challenges, notify).await;
    let jobs = spawn_worker_runner(
        &database,
        std::slice::from_ref(&profile),
        &backends,
        test_auditor(database.clone()),
    );
    let router = build_app(
        database.clone(),
        Arc::new(config),
        vec![profile],
        test_auditor(database.clone()),
        Arc::new(Metrics::new(database.clone())),
        jobs,
    );
    (router, database)
}

/// [`test_app_with_signer`] over a caller-supplied database, for a test whose
/// signer keeps state in that same database — a local CA whose revocations a
/// second instance, standing in for another process, must see.
pub async fn test_app_over(database: Arc<Database>, signer: Arc<dyn SignerBackend>) -> Router {
    init_tracing();
    let config = Config::default();
    let (profile, backends) = one_profile(
        &database,
        &config,
        signer,
        Arc::new(FilterPolicy::default()),
        default_challenges(),
        no_notifications().await,
    )
    .await;
    let jobs = spawn_worker_runner(
        &database,
        std::slice::from_ref(&profile),
        &backends,
        test_auditor(database.clone()),
    );
    build_app(
        database.clone(),
        Arc::new(config),
        vec![profile],
        test_auditor(database.clone()),
        Arc::new(Metrics::new(database)),
        jobs,
    )
}

/// A file-backed database, removed with its WAL sidecars when the test ends.
///
/// Hold it for as long as the app that opened it: dropping it deletes the file.
pub struct DiskDb(std::path::PathBuf);

impl Drop for DiskDb {
    fn drop(&mut self) {
        for suffix in ["", "-wal", "-shm"] {
            let _ = std::fs::remove_file(format!("{}{suffix}", self.0.display()));
        }
    }
}

/// [`test_app_with_db`] over a **file-backed** database, so the pool opens more
/// than one connection.
///
/// `Database::connect_in_memory` pins `max_connections(1)` — one shared
/// in-memory database is the whole point of it — which serializes every caller
/// and makes a genuine write race impossible to reproduce. A suite asserting
/// what two simultaneous requests do to one row needs this instead; everything
/// else should keep using the in-memory helper, which is faster and leaves
/// nothing behind.
pub async fn test_app_on_disk() -> (Router, Arc<Database>, DiskDb) {
    init_tracing();
    let path = std::env::temp_dir().join(format!("acme-proxy-test-{}.db", uuid::Uuid::now_v7()));
    let database = Arc::new(
        Database::connect_and_migrate(&format!("sqlite://{}", path.display()))
            .await
            .unwrap(),
    );
    let config = Config::default();
    let (profile, backends) = one_profile(
        &database,
        &config,
        Arc::new(memory_ca().await),
        Arc::new(FilterPolicy::default()),
        default_challenges(),
        no_notifications().await,
    )
    .await;
    let jobs = spawn_worker_runner(
        &database,
        std::slice::from_ref(&profile),
        &backends,
        test_auditor(database.clone()),
    );
    let router = build_app(
        database.clone(),
        Arc::new(config),
        vec![profile],
        test_auditor(database.clone()),
        Arc::new(Metrics::new(database.clone())),
        jobs,
    );
    (router, database, DiskDb(path))
}

/// `test_app_full`, plus the registry the app counts into.
///
/// Separate rather than widening every helper's return type: one suite asserts
/// on the counters and nineteen do not. The `Auditor` is given the same
/// registry, which is what makes an issuance counted here at all — see
/// `Auditor::with_metrics`.
pub async fn test_app_with_metrics(
    mut config: Config,
    signer: Arc<dyn SignerBackend>,
) -> (Router, Arc<Database>, Arc<Metrics>) {
    init_tracing();
    // Forced on, because `build_app` mounts the counting middleware only when
    // it is — an operator who has not asked for metrics pays nothing per
    // request. A caller of *this* helper has asked, and leaving it to each one
    // to remember would mean a test whose counters silently stay at zero while
    // every assertion about them reads as a wiring bug.
    config.metrics.enabled = true;
    let database = Arc::new(Database::connect_in_memory().await.unwrap());
    let metrics = Arc::new(Metrics::new(database.clone()));
    let (profile, backends) = one_profile(
        &database,
        &config,
        signer,
        Arc::new(FilterPolicy::default()),
        Arc::new(ChallengeRegistry::default()),
        no_notifications().await,
    )
    .await;
    let auditor = Arc::new(
        Auditor::with_resolver(
            database.clone(),
            None,
            std::time::Duration::from_millis(100),
        )
        .with_metrics(metrics.clone()),
    );
    let jobs = spawn_worker_runner(
        &database,
        std::slice::from_ref(&profile),
        &backends,
        auditor.clone(),
    );
    let router = build_app(
        database.clone(),
        Arc::new(config),
        vec![profile],
        auditor,
        metrics.clone(),
        jobs,
    );
    (router, database, metrics)
}

/// The single `default` profile both app builders mount.
///
/// Factored out rather than copied so `test_admin_app` and `test_app_full`
/// cannot drift into mounting subtly different endpoints.
async fn one_profile(
    database: &Database,
    config: &Config,
    signer: Arc<dyn SignerBackend>,
    filter: Arc<FilterPolicy>,
    challenges: Arc<ChallengeRegistry>,
    notify: Arc<NotifyDispatcher>,
) -> (Arc<Profile>, Vec<(String, Arc<dyn SignerBackend>)>) {
    met_by_a_worker(signer.as_ref()).await;
    let profile = Arc::new(Profile::new(
        PROFILE,
        &config.server.base_url,
        ProfileParts {
            signer_info: read_side(signer.as_ref(), database).await,
            filter,
            challenges,
            order: config.order.clone(),
            eab: config.eab.clone(),
            meta: config.meta.clone(),
            notify,
        },
    ));
    (profile, vec![(PROFILE.to_string(), signer)])
}

/// What a worker's first pass leaves behind for `signer`: a local CA's first
/// CRL stored.
///
/// Production serves `/crl` from the stored row through the read side, which
/// never signs, and the `worker` role's startup sweep is what stores the first
/// one. The suite has no startup sweep, so every app builder does this in its
/// place — otherwise a test reading the CRL before any revocation would see
/// the brief window a real deployment has before its worker's first tick.
pub async fn met_by_a_worker(signer: &dyn SignerBackend) {
    if let Some(refresher) = signer.crl_refresher() {
        refresher
            .refresh()
            .await
            .expect("an in-memory CA stores its first CRL");
    }
}

/// The read side a test app serves `signer` through.
///
/// `signer.info()` itself, unless the backend keeps its revocation state in a
/// database of its own — a fixture such as [`memory_ca`] built before the app's
/// database existed. A local CA's revocation from a request is a ledger row in
/// the *app's* database, which such a CA's CRL would never see; so for those
/// the read side routes revocations through the backend instead, queued for
/// the harness's worker — which is what every delegating backend does anyway,
/// and what a fixture that fails `revoke` on purpose needs to be reached at
/// all. The builders that own their CA ([`test_app_with_db`], the admin apps)
/// build it over the app's database, so the ledger route is what they exercise.
pub async fn read_side(signer: &dyn SignerBackend, database: &Database) -> Arc<dyn SignerInfo> {
    let info = signer.info();
    if let RevocationRoute::Ledger { issuer } = info.revocation_route()
        && acme_proxy_store::crl::StoredCrl::find_current(&issuer, database)
            .await
            .unwrap()
            .is_none()
    {
        return Arc::new(OwnDatabase(info));
    }
    info
}

/// A read side whose revocations go to the backend rather than to a ledger in
/// a database the backend does not keep its state in. See [`read_side`].
struct OwnDatabase(Arc<dyn SignerInfo>);

#[async_trait]
impl SignerInfo for OwnDatabase {
    async fn crl_der(&self) -> Result<Option<Vec<u8>>, SignerError> {
        self.0.crl_der().await
    }

    async fn ca_chain_pem(&self) -> Option<String> {
        self.0.ca_chain_pem().await
    }

    async fn renewal_info(&self, cert_der: &[u8]) -> Result<Option<RenewalWindow>, SignerError> {
        self.0.renewal_info(cert_der).await
    }

    fn http01_tokens(&self) -> Option<Arc<dyn Http01TokenStore>> {
        self.0.http01_tokens()
    }

    fn revocation_route(&self) -> RevocationRoute {
        RevocationRoute::Delegated
    }
}

/// The auditor every test app is built with: **no reverse lookup**.
///
/// `audit.reverse_dns` defaults to `true` in production, and deliberately not
/// here — the suite runs with no network and `Auditor::from_config` would build
/// a resolver from `/etc/resolv.conf`, making the audit path's behaviour depend
/// on the machine the tests run on. Every audit row a test sees therefore has a
/// `client_ip` and no `client_ptr`; a test that wants the PTR half drives
/// `Auditor::with_resolver` with a stub, the way the challenge and filter
/// suites drive theirs.
pub fn test_auditor(database: Arc<Database>) -> Arc<Auditor> {
    Arc::new(Auditor::with_resolver(
        database,
        None,
        std::time::Duration::from_millis(100),
    ))
}

/// The serials the CRL at `crl_path` lists, once it lists `serial_hex`.
///
/// A local CA's revocation is a ledger row that the worker then signs into the
/// CRL (`local_ca_crl_regenerate`), so the CRL follows a revocation rather than
/// changing with it. Polls the way a relying party refetching would.
pub async fn await_crl_listing(app: &Router, crl_path: &str, serial_hex: &str) -> Vec<String> {
    use x509_parser::prelude::FromDer;
    let mut serials = Vec::new();
    for _ in 0..600 {
        let response = app
            .clone()
            .oneshot(Request::get(crl_path).body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK, "GET {crl_path}");
        let der = response.into_body().collect().await.unwrap().to_bytes();
        let (_, crl) =
            x509_parser::revocation_list::CertificateRevocationList::from_der(&der).unwrap();
        serials = crl
            .iter_revoked_certificates()
            .map(|entry| entry.raw_serial_as_string().replace(':', ""))
            .collect();
        if serials.iter().any(|s| s.eq_ignore_ascii_case(serial_hex)) {
            return serials;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    panic!("expected {serial_hex} in the CRL at {crl_path}, which lists {serials:?}");
}

/// A `Config` with the web admin enabled at its defaults.
pub fn admin_config() -> Config {
    let mut config = Config::default();
    config.admin.enabled = true;
    config
}

/// Builds the **admin** router (not the ACME one) over a throwaway in-memory
/// database, plus the one `default` profile so `revoke` can resolve a signer.
pub async fn test_admin_app(config: Config) -> (Router, Arc<Database>) {
    let (router, database, _signer) = test_admin_app_with_signer(config).await;
    (router, database)
}

/// [`test_admin_app`], also handing back the signer the profile was built with.
///
/// A test that needs an order to be genuinely *issued* — so the admin API's
/// revoke path has a certificate to act on — must issue through the very same
/// backend the handler will later revoke against, or the CA-side ledger and
/// the CRL would belong to two different objects.
pub async fn test_admin_app_with_signer(
    config: Config,
) -> (Router, Arc<Database>, Arc<dyn SignerBackend>) {
    admin_app_with(config, Arc::new(FilterPolicy::default())).await
}

/// The one place the admin router is built, so the variants above and below
/// cannot drift into mounting different endpoints.
async fn admin_app_with(
    config: Config,
    filter: Arc<FilterPolicy>,
) -> (Router, Arc<Database>, Arc<dyn SignerBackend>) {
    admin_app_with_notifiers(
        config,
        filter,
        acme_proxy_jobs::notify::DispatcherMap::new().into(),
    )
    .await
}

async fn admin_app_with_notifiers(
    config: Config,
    filter: Arc<FilterPolicy>,
    notifiers: acme_proxy_jobs::notify::Notifiers,
) -> (Router, Arc<Database>, Arc<dyn SignerBackend>) {
    init_tracing();
    let database = Arc::new(Database::connect_in_memory().await.unwrap());
    let signer: Arc<dyn SignerBackend> = Arc::new(
        LocalCa::generate_in_memory("ecdsa-p256", 90, database.clone())
            .expect("an in-memory CA is always available"),
    );
    let (profile, backends) = one_profile(
        &database,
        &config,
        signer.clone(),
        filter,
        default_challenges(),
        no_notifications().await,
    )
    .await;
    let jobs = spawn_worker_runner(
        &database,
        std::slice::from_ref(&profile),
        &backends,
        test_auditor(database.clone()),
    );
    let router = acme_proxy_admin::webadmin::build_admin_app(
        database.clone(),
        Arc::new(config),
        &[profile],
        test_auditor(database.clone()),
        notifiers,
        jobs,
    );
    (router, database, signer)
}

/// [`test_admin_app_logged_in`], with a recording web-admin security
/// dispatcher wired in (under `ADMIN_DISPATCHER_KEY`) and a runner draining it,
/// for the tests whose subject is the ASVS V6.3.5 / V6.3.7 notifications.
pub async fn test_admin_app_logged_in_with_security_notify(
    config: Config,
) -> (Router, Arc<Database>, AdminSessionHandle, NotifyHarness) {
    let (harness, notifiers) = NotifyHarness::admin().await;
    let (app, database, _signer) =
        admin_app_with_notifiers(config, Arc::new(FilterPolicy::default()), notifiers).await;
    acme_proxy_admin::admin::users::create_user(
        "alice",
        ADMIN_PASSWORD,
        &PasswordContext::empty(),
        None,
        database.clone(),
    )
    .await
    .expect("the bootstrap operator must be creatable");
    acme_proxy_admin::admin::users::set_contact_email(
        "alice",
        Some("alice@example.com"),
        database.clone(),
    )
    .await
    .expect("a contact address must be settable");
    // This first login seeds `known_login_ips` from the default peer address, so
    // a later login from that same address is silent and only a *new* one fires.
    let handle = admin_login(&app, "alice", ADMIN_PASSWORD).await;
    (app, database, handle, harness)
}

/// A `[filter]` policy with two checks and two rules, built the way the server
/// builds one.
///
/// Every other admin harness mounts `FilterPolicy::default()` — no rules, so
/// `is_active()` is false — which is the honest default for a suite that does
/// not care about filtering, and exactly wrong for a surface whose whole
/// subject is the checks and the rules.
///
/// Built through `filter::from_config` rather than `FilterPolicy::new`, because
/// that is the path a running server takes: it is what decides each check's
/// kind and each rule's stage intersection, so a test over it is a test of the
/// real thing.
///
/// `corp`'s condition is written **without parentheses** on purpose. What the
/// panel and the CLI must print back is `names or (mgmt-net and names)`, and
/// that re-parenthesization is the reason either surface exists.
pub fn test_filter_policy() -> Arc<FilterPolicy> {
    use acme_proxy_core::config::CheckConfig;
    use acme_proxy_core::config::FilterConfig;
    use acme_proxy_core::config::RuleConfig;

    let filter = FilterConfig {
        rules: vec!["mgmt-bypass".to_string(), "corp".to_string()],
        default: "deny".to_string(),
        rule: [
            (
                "mgmt-bypass".to_string(),
                RuleConfig {
                    when: "mgmt-net".to_string(),
                    then: "allow".to_string(),
                    ..RuleConfig::default()
                },
            ),
            (
                "corp".to_string(),
                RuleConfig {
                    when: "names or mgmt-net and names".to_string(),
                    then: "deny".to_string(),
                    mode: "warn".to_string(),
                    ..RuleConfig::default()
                },
            ),
        ]
        .into_iter()
        .collect(),
        check: [
            (
                "mgmt-net".to_string(),
                CheckConfig {
                    r#type: "allowed_ip".to_string(),
                    allow: vec!["10.0.0.0/8".to_string()],
                    ..CheckConfig::default()
                },
            ),
            (
                "names".to_string(),
                CheckConfig {
                    r#type: "identifiers".to_string(),
                    allow: vec!["*.corp.example.com".to_string()],
                    ..CheckConfig::default()
                },
            ),
        ]
        .into_iter()
        .collect(),
        ..FilterConfig::default()
    };

    acme_proxy_policy::filter::from_config(
        &filter,
        &acme_proxy_core::config::DnsConfig::default(),
        None,
        false,
    )
    .expect("the test policy must build")
}

/// A signed-in admin session: the cookie to send back and the CSRF token every
/// mutating request needs.
pub struct AdminSessionHandle {
    pub cookie: String,
    pub csrf: String,
}

/// The password `test_admin_app_logged_in` creates its operator with.
pub const ADMIN_PASSWORD: &str = "a-long-enough-password";

/// Builds the admin router, creates one operator, signs in, and hands back the
/// credentials the rest of a test needs.
pub async fn test_admin_app_logged_in(
    config: Config,
) -> (Router, Arc<Database>, AdminSessionHandle) {
    test_admin_app_logged_in_with_filter(config, Arc::new(FilterPolicy::default())).await
}

/// [`test_admin_app_logged_in`], over a profile mounting a policy of the
/// caller's choosing — for the surfaces whose subject *is* the policy.
pub async fn test_admin_app_logged_in_with_filter(
    config: Config,
    filter: Arc<FilterPolicy>,
) -> (Router, Arc<Database>, AdminSessionHandle) {
    let (app, database, _signer) = admin_app_with(config, filter).await;
    acme_proxy_admin::admin::users::create_user(
        "alice",
        ADMIN_PASSWORD,
        &PasswordContext::empty(),
        None,
        database.clone(),
    )
    .await
    .expect("the bootstrap operator must be creatable");
    let handle = admin_login(&app, "alice", ADMIN_PASSWORD).await;
    (app, database, handle)
}

/// Signs in and extracts the cookie and CSRF token, failing loudly if the
/// login did not succeed.
pub async fn admin_login(app: &Router, username: &str, password: &str) -> AdminSessionHandle {
    let response = admin_request(
        app,
        Method::POST,
        "/api/session",
        None,
        Some(serde_json::json!({ "username": username, "password": password })),
    )
    .await;
    assert_eq!(
        response.status(),
        StatusCode::OK,
        "login must succeed to build a session handle"
    );

    let cookie = session_cookie_token(&response).expect("login must set the session cookie");
    let body = json_body(response).await;
    let csrf = body["csrfToken"]
        .as_str()
        .expect("login must return a csrfToken")
        .to_string();
    AdminSessionHandle { cookie, csrf }
}

/// Signs in as far as the password step and asserts the login is **not**
/// complete, handing back the half-authenticated cookie.
///
/// [`admin_login`]'s sibling. The CSRF token is real -- the pending response
/// carries one so the enrolment routes can be reached -- but no ordinary route
/// will accept this session.
pub async fn admin_login_pending(app: &Router, username: &str, password: &str) -> (String, String) {
    let response = admin_request(
        app,
        Method::POST,
        "/api/session",
        None,
        Some(json!({ "username": username, "password": password })),
    )
    .await;
    assert_eq!(
        response.status(),
        StatusCode::OK,
        "a correct password answers 200 even when a factor is outstanding"
    );

    let cookie = session_cookie_token(&response).expect("a pending login must set the cookie");
    let body = json_body(response).await;
    assert_eq!(
        body["mfaRequired"], true,
        "this helper is for an operator who owes a second factor"
    );
    let csrf = body["csrfToken"]
        .as_str()
        .expect("a pending login must return a csrfToken")
        .to_string();
    (cookie, csrf)
}

/// Gives `username` a confirmed second factor, through the same operation layer
/// the panel uses, and returns the raw secret so the test can compute codes.
pub async fn enrol_totp(database: Arc<Database>, username: &str) -> Vec<u8> {
    let mut user = acme_proxy_store::admin_user::AdminUser::find_by_username(username, &database)
        .await
        .unwrap()
        .expect("the operator must exist before enrolling them");

    let enrolment = acme_proxy_admin::admin::mfa::begin_totp_enrolment(
        &mut user,
        "http://localhost:3001",
        database.clone(),
    )
    .await
    .unwrap();

    let code = totp_code(&enrolment.secret, 0);
    acme_proxy_admin::admin::mfa::confirm_totp_enrolment(&mut user, &code, None, database)
        .await
        .unwrap()
        .expect("a freshly generated code must confirm its own enrolment");

    enrolment.secret
}

/// The text between two markers, for pulling a secret out of a rendered page.
///
/// Panics loudly rather than returning an `Option`: every caller is asserting
/// that the page contains it, so a miss is the test result.
pub fn between(haystack: &str, start: &str, end: &str) -> String {
    let rest = haystack
        .split_once(start)
        .unwrap_or_else(|| panic!("`{start}` not found in the rendered page"))
        .1;
    rest.split_once(end)
        .unwrap_or_else(|| panic!("`{end}` not found after `{start}`"))
        .0
        .trim()
        .to_string()
}

/// Decodes the unpadded RFC 4648 base32 the enrolment endpoint hands back.
///
/// Lives here and not in `admin::totp` on purpose: nothing in the server ever
/// reads base32 back, so a production decoder would be untested surface. The
/// tests need one because they receive the secret the way an operator does --
/// as the string they would type into an app -- and then have to compute codes
/// from it.
pub fn base32_decode(encoded: &str) -> Vec<u8> {
    const ALPHABET: &[u8; 32] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ234567";
    let mut bits = 0u32;
    let mut buffer = 0u64;
    let mut out = Vec::new();

    for character in encoded.bytes() {
        let value = ALPHABET
            .iter()
            .position(|symbol| *symbol == character)
            .unwrap_or_else(|| panic!("`{}` is not base32", character as char));
        buffer = (buffer << 5) | value as u64;
        bits += 5;
        if bits >= 8 {
            bits -= 8;
            out.push(((buffer >> bits) & 0xff) as u8);
        }
    }
    out
}

/// The code `secret` produces `steps` time steps from now.
///
/// **Prefer [`admin_login_mfa`] for a login that is only meant to succeed.**
/// Stepping forward by hand is fragile in this suite: a login costs several
/// PBKDF2 runs, so two of them routinely straddle a 30-second boundary and a
/// code computed as "now + 2" lands outside the ±1 window by the time it is
/// submitted. Reach for this only where the *code itself* is the subject.
pub fn totp_code(secret: &[u8], steps: i64) -> String {
    use acme_proxy_admin::admin::totp;
    totp::totp_at(secret, totp::step_at(now_unix()) + steps, totp::DIGITS)
}

/// Wall-clock seconds, the same value `acme_proxy_store::nonce::now_secs` returns — that
/// one is `pub(crate)`, and an integration test links this crate from outside.
pub fn now_unix() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64
}

/// Signs in through both steps and hands back a usable session.
///
/// Clears `totp_last_step` first, so the code for the current step is always
/// accepted. That is the harness stepping out of the replay guard's way rather
/// than testing around it: the guard is a `WHERE` clause with its own tests
/// (`AdminUser::claim_totp_step`, `admin::mfa`, and
/// `a_code_cannot_be_spent_twice` below), and an integration test that has to
/// win a race against a 30-second clock to sign in is a flake, not a
/// regression test.
pub async fn admin_login_mfa(
    app: &Router,
    database: Arc<Database>,
    username: &str,
    password: &str,
    secret: &[u8],
) -> AdminSessionHandle {
    sqlx::query("UPDATE admin_users SET totp_last_step = NULL WHERE username = ?;")
        .bind(username)
        .execute(database.raw_pool())
        .await
        .unwrap();

    let (cookie, csrf) = admin_login_pending(app, username, password).await;
    let pending = AdminSessionHandle { cookie, csrf };

    let response = admin_request(
        app,
        Method::POST,
        "/api/session/mfa",
        Some(&pending),
        Some(json!({ "code": totp_code(secret, 0) })),
    )
    .await;
    assert_eq!(
        response.status(),
        StatusCode::OK,
        "the second step must complete the login"
    );

    let cookie = session_cookie_token(&response).expect("promotion must rotate the cookie");
    let body = json_body(response).await;
    AdminSessionHandle {
        cookie,
        csrf: body["csrfToken"]
            .as_str()
            .expect("a promoted session must return a csrfToken")
            .to_string(),
    }
}

/// The `__Host-acme_admin_session` value from a response's `Set-Cookie`, if it
/// set one to a non-empty value.
pub fn session_cookie_token(response: &Response) -> Option<String> {
    let raw = response
        .headers()
        .get(header::SET_COOKIE)?
        .to_str()
        .ok()?
        .to_string();
    let value = raw
        .split(';')
        .next()?
        .strip_prefix("__Host-acme_admin_session=")?
        .to_string();
    (!value.is_empty()).then_some(value)
}

/// Drives one admin request, optionally authenticated and optionally with a
/// JSON body.
///
/// `session` supplies both the cookie and the `X-CSRF-Token`; a test that wants
/// to omit or corrupt the token builds the request itself.
pub async fn admin_request(
    app: &Router,
    method: Method,
    path: &str,
    session: Option<&AdminSessionHandle>,
    body: Option<serde_json::Value>,
) -> Response {
    admin_request_from(app, method, path, session, body, "127.0.0.1:40000").await
}

/// [`admin_request`], with the peer address spelled out.
///
/// The address is the `LoginLimiter`'s key, so anything asserting that a bound
/// does — or deliberately does not — survive an attacker rotating addresses has
/// to choose it per request.
pub async fn admin_request_from(
    app: &Router,
    method: Method,
    path: &str,
    session: Option<&AdminSessionHandle>,
    body: Option<serde_json::Value>,
    peer: &str,
) -> Response {
    let mut builder = Request::builder().method(method).uri(path);
    if let Some(session) = session {
        builder = builder
            .header(
                header::COOKIE,
                format!("__Host-acme_admin_session={}", session.cookie),
            )
            .header("x-csrf-token", &session.csrf);
    }
    let request = match body {
        Some(value) => builder
            .header(header::CONTENT_TYPE, "application/json")
            .body(Body::from(value.to_string()))
            .unwrap(),
        None => builder.body(Body::empty()).unwrap(),
    };
    send_from(app, request, peer).await
}

/// Drives one `/ui` page request.
///
/// `hx` sets `HX-Request: true`, which is what tells a page handler to answer
/// with the bare fragment rather than the whole document — and what turns a
/// sign-in redirect into an `HX-Redirect` header. Both branches are worth
/// exercising on every route, so it is a parameter rather than two helpers.
pub async fn admin_page(
    app: &Router,
    path: &str,
    session: Option<&AdminSessionHandle>,
    hx: bool,
) -> Response {
    let mut builder = Request::builder().method(Method::GET).uri(path);
    if let Some(session) = session {
        builder = builder.header(
            header::COOKIE,
            format!("__Host-acme_admin_session={}", session.cookie),
        );
    }
    if hx {
        builder = builder.header("hx-request", "true");
    }
    send_from(app, builder.body(Body::empty()).unwrap(), "127.0.0.1:40000").await
}

/// Drives one `/ui` mutation, with a form body rather than JSON.
///
/// The CSRF token still travels as `X-CSRF-Token` — htmx puts it there from
/// `hx-headers`, which is why the page layer needed no second CSRF path.
pub async fn admin_form_request(
    app: &Router,
    method: Method,
    path: &str,
    session: Option<&AdminSessionHandle>,
    form: Option<&[(&str, &str)]>,
) -> Response {
    let mut builder = Request::builder().method(method).uri(path);
    if let Some(session) = session {
        builder = builder
            .header(
                header::COOKIE,
                format!("__Host-acme_admin_session={}", session.cookie),
            )
            .header("x-csrf-token", &session.csrf);
    }
    let request = match form {
        Some(pairs) => builder
            .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
            .body(Body::from(encode_form(pairs)))
            .unwrap(),
        None => builder.body(Body::empty()).unwrap(),
    };
    send_from(app, request, "127.0.0.1:40000").await
}

/// `application/x-www-form-urlencoded`, hand-rolled so the test crate needs no
/// encoder dependency of its own.
fn encode_form(pairs: &[(&str, &str)]) -> String {
    fn encode(raw: &str) -> String {
        let mut out = String::with_capacity(raw.len());
        for byte in raw.bytes() {
            match byte {
                b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                    out.push(byte as char);
                }
                b' ' => out.push('+'),
                other => out.push_str(&format!("%{other:02X}")),
            }
        }
        out
    }

    pairs
        .iter()
        .map(|(key, value)| format!("{}={}", encode(key), encode(value)))
        .collect::<Vec<_>>()
        .join("&")
}

/// Reads a response body as text, for the HTML the page layer serves.
pub async fn html_body(response: Response) -> String {
    let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
        .await
        .unwrap();
    String::from_utf8(bytes.to_vec()).expect("an admin page must be UTF-8")
}

/// Parses a response body as JSON.
pub async fn json_body(response: Response) -> serde_json::Value {
    let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
        .await
        .unwrap();
    serde_json::from_slice(&bytes).unwrap_or_else(|error| {
        panic!(
            "expected a JSON body, got `{}`: {error}",
            String::from_utf8_lossy(&bytes)
        )
    })
}

/// A challenge validator that blocks until the test lets it go, and counts the
/// validations that got as far as it.
///
/// `post_challenge` awaits validation inline, so this is how a test holds an
/// ACME request open for as long as it needs — which is what the admission
/// limit has to be observed against, and the window in which a second trigger
/// for the same challenge is still in flight.
///
/// Both halves are semaphores, `GatedSigner`'s shape and for its reasons. The
/// gate is one because **a release arriving before the wait must still count**;
/// `Notify::notify_waiters` wakes only tasks already parked, so a test whose
/// spawned request had not yet reached the `.await` would drop the wake-up and
/// hang until the harness killed it. `entered` is what a test awaits instead of
/// sleeping: a permit appears the moment a validation is genuinely inside here,
/// so the assertion is against the state it is about rather than against a
/// timer, which is what this fixture's own doc comment always asked for.
pub struct BlockingValidator {
    typ: &'static str,
    /// Validations that reached this validator. What a "ran exactly once"
    /// assertion rests on.
    calls: Arc<AtomicUsize>,
    /// Zero permits until the test releases; see `first_only` for who waits.
    gate: Arc<tokio::sync::Semaphore>,
    /// A permit per caller that has entered, so a test can await arrival.
    entered: Arc<tokio::sync::Semaphore>,
    /// Whether only the first caller waits on the gate.
    first_only: bool,
}

impl BlockingValidator {
    /// Every validation blocks until the test releases it — what an
    /// admission-control test needs, since the point is to occupy slots.
    pub fn new(typ: &'static str) -> Self {
        Self {
            typ,
            calls: Arc::new(AtomicUsize::new(0)),
            gate: Arc::new(tokio::sync::Semaphore::new(0)),
            entered: Arc::new(tokio::sync::Semaphore::new(0)),
            first_only: false,
        }
    }

    /// Only the **first** validation blocks; later ones return immediately.
    ///
    /// `GatedSigner`'s reasoning, for the same reason: a test proving a second
    /// request never reaches the validator has to survive the build where it
    /// does. If that second call blocked too, an unfixed build would deadlock
    /// the test into a harness timeout — which reads like flake — instead of
    /// failing the assertion that names the bug.
    pub fn gating_the_first(typ: &'static str) -> Self {
        Self {
            first_only: true,
            ..Self::new(typ)
        }
    }

    /// Handles for the test to drive the gate with, since the validator itself
    /// is consumed by `challenges_with`: the call count, the gate to release
    /// through, and the signal that a validation has arrived.
    #[must_use]
    pub fn handles(
        &self,
    ) -> (
        Arc<AtomicUsize>,
        Arc<tokio::sync::Semaphore>,
        Arc<tokio::sync::Semaphore>,
    ) {
        (
            Arc::clone(&self.calls),
            Arc::clone(&self.gate),
            Arc::clone(&self.entered),
        )
    }
}

#[async_trait]
impl ChallengeValidator for BlockingValidator {
    fn typ(&self) -> &'static str {
        self.typ
    }

    async fn validate(&self, _context: &ValidationContext<'_>) -> Result<(), ChallengeError> {
        let first = self.calls.fetch_add(1, Ordering::SeqCst) == 0;
        self.entered.add_permits(1);
        if first || !self.first_only {
            // `forget`, so releasing N permits lets exactly N validations
            // through rather than returning one to the gate on drop.
            self.gate
                .acquire()
                .await
                .expect("the gate is never closed while a test holds it")
                .forget();
        }
        Ok(())
    }
}

/// A challenge validator answering from a canned outcome and counting its calls,
/// so a test can assert that a decided challenge was *not* re-validated.
pub struct StubValidator {
    typ: &'static str,
    outcome: Option<String>,
    calls: Arc<AtomicUsize>,
}

impl StubValidator {
    /// Passes every challenge of `typ`.
    pub fn passing(typ: &'static str) -> Self {
        Self {
            typ,
            outcome: None,
            calls: Arc::new(AtomicUsize::new(0)),
        }
    }

    /// Fails every challenge of `typ` with an `incorrectResponse`.
    pub fn failing(typ: &'static str, detail: &str) -> Self {
        Self {
            typ,
            outcome: Some(detail.to_string()),
            calls: Arc::new(AtomicUsize::new(0)),
        }
    }

    /// A handle to this validator's call counter, cloneable before the validator
    /// is moved into the registry.
    pub fn counter(&self) -> Arc<AtomicUsize> {
        self.calls.clone()
    }
}

#[async_trait]
impl ChallengeValidator for StubValidator {
    fn typ(&self) -> &'static str {
        self.typ
    }

    async fn validate(&self, _ctx: &ValidationContext<'_>) -> Result<(), ChallengeError> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        match &self.outcome {
            Some(detail) => Err(ChallengeError::IncorrectResponse(detail.clone())),
            None => Ok(()),
        }
    }
}

/// A validator that records the key authorization it was handed, so a test can
/// check the server derives it the way a real client would.
pub struct RecordingValidator {
    typ: &'static str,
    seen: Arc<Mutex<Vec<String>>>,
}

impl RecordingValidator {
    pub fn new(typ: &'static str) -> Self {
        Self {
            typ,
            seen: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub fn seen(&self) -> Arc<Mutex<Vec<String>>> {
        self.seen.clone()
    }
}

#[async_trait]
impl ChallengeValidator for RecordingValidator {
    fn typ(&self) -> &'static str {
        self.typ
    }

    async fn validate(&self, ctx: &ValidationContext<'_>) -> Result<(), ChallengeError> {
        self.seen.lock().unwrap().push(format!(
            "{}|{}|{}|{}",
            ctx.identifier, ctx.wildcard, ctx.token, ctx.key_authorization
        ));
        Ok(())
    }
}

/// A signer backend that always fails internally, for exercising the
/// finalize → `serverInternal` / order-`invalid` path.
pub struct FailingSigner;

#[async_trait]
impl SignerBackend for FailingSigner {
    async fn issue(
        &self,
        _order_id: &str,
        _csr_der: &[u8],
        _identifiers: &[Identifier],
        _validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        Err(SignerError::Internal("boom".to_string()))
    }

    async fn revoke(&self, _cert_der: &[u8], _reason: Option<u32>) -> Result<(), SignerError> {
        Err(SignerError::Internal("boom".to_string()))
    }
}

/// A real [`LocalCa`] whose **first** `issue` parks inside the signing call
/// until the test lets it go, so two finalize requests can be held in flight at
/// once.
///
/// This is the only way to reproduce the double-issuance race from outside: the
/// window it exploits is between `post_finalize` reading `status == "ready"` and
/// writing the certificate back, and nothing else in the suite can hold a
/// request open inside it. `calls` is what the assertion actually rests on — a
/// second certificate leaves no row, so counting the signing calls is the only
/// evidence that one was ever created.
///
/// Later calls return **immediately** rather than also waiting, deliberately: on
/// an unfixed build the second request must run to completion and fail an
/// assertion, not deadlock the test into a timeout that reads like flake.
pub struct GatedSigner {
    ca: Arc<LocalCa>,
    /// Signing calls that got as far as the CA. The point of the fixture.
    pub calls: Arc<AtomicUsize>,
    /// Zero permits until the test releases; only the first caller acquires.
    /// A semaphore rather than a `Notify` because a release arriving before the
    /// wait must still count.
    gate: Arc<tokio::sync::Semaphore>,
    /// Permits added as each caller enters, so the test can await the first one
    /// being genuinely inside `issue` rather than sleeping and hoping.
    entered: Arc<tokio::sync::Semaphore>,
}

impl GatedSigner {
    pub async fn new() -> Self {
        Self {
            ca: Arc::new(memory_ca().await),
            calls: Arc::new(AtomicUsize::new(0)),
            gate: Arc::new(tokio::sync::Semaphore::new(0)),
            entered: Arc::new(tokio::sync::Semaphore::new(0)),
        }
    }

    /// Handles for the test to drive the gate with, since the backend itself is
    /// consumed by `test_app_with_signer`.
    #[must_use]
    pub fn handles(
        &self,
    ) -> (
        Arc<AtomicUsize>,
        Arc<tokio::sync::Semaphore>,
        Arc<tokio::sync::Semaphore>,
    ) {
        (
            Arc::clone(&self.calls),
            Arc::clone(&self.gate),
            Arc::clone(&self.entered),
        )
    }
}

#[async_trait]
impl SignerBackend for GatedSigner {
    async fn issue(
        &self,
        order_id: &str,
        csr_der: &[u8],
        identifiers: &[Identifier],
        validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        let first = self.calls.fetch_add(1, Ordering::SeqCst) == 0;
        self.entered.add_permits(1);
        if first {
            let _permit = self
                .gate
                .acquire()
                .await
                .expect("the gate is never closed while a test holds it");
        }
        self.ca
            .issue(order_id, csr_der, identifiers, validity)
            .await
    }

    async fn revoke(&self, cert_der: &[u8], reason: Option<u32>) -> Result<(), SignerError> {
        self.ca.revoke(cert_der, reason).await
    }

    fn info(&self) -> Arc<dyn SignerInfo> {
        self.ca.info()
    }

    fn crl_refresher(&self) -> Option<Arc<dyn CrlRefresher>> {
        self.ca.crl_refresher()
    }
}

/// A real [`LocalCa`] whose read side parks inside **`crl_der`** until the test
/// lets it go — so a `GET /crl` holds a genuine request open.
///
/// What `tests/admission.rs` stands on. A request that waits on its backend is
/// the only way to fill an admission slot from outside, and after challenge
/// validation and then issuance moved into the job queue, the read side is
/// what a request still asks inline: a `custom` script's `crl` and
/// `renewal_info` hooks run in the request. Every call parks until released, so
/// a test holding one request and then releasing one permit frees exactly it.
pub struct GatedCrlSigner {
    ca: Arc<LocalCa>,
    gate: Arc<tokio::sync::Semaphore>,
    entered: Arc<tokio::sync::Semaphore>,
}

impl GatedCrlSigner {
    pub async fn new() -> Self {
        Self {
            ca: Arc::new(memory_ca().await),
            gate: Arc::new(tokio::sync::Semaphore::new(0)),
            entered: Arc::new(tokio::sync::Semaphore::new(0)),
        }
    }

    /// `(gate, entered)`: permits added to `gate` release a parked read, and
    /// `entered` gains one as each read arrives.
    #[must_use]
    pub fn handles(&self) -> (Arc<tokio::sync::Semaphore>, Arc<tokio::sync::Semaphore>) {
        (Arc::clone(&self.gate), Arc::clone(&self.entered))
    }
}

#[async_trait]
impl SignerBackend for GatedCrlSigner {
    async fn issue(
        &self,
        order_id: &str,
        csr_der: &[u8],
        identifiers: &[Identifier],
        validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        self.ca
            .issue(order_id, csr_der, identifiers, validity)
            .await
    }

    async fn revoke(&self, cert_der: &[u8], reason: Option<u32>) -> Result<(), SignerError> {
        self.ca.revoke(cert_der, reason).await
    }

    fn crl_refresher(&self) -> Option<Arc<dyn CrlRefresher>> {
        self.ca.crl_refresher()
    }

    fn info(&self) -> Arc<dyn SignerInfo> {
        Arc::new(GatedCrlInfo {
            ca: self.ca.info(),
            gate: self.gate.clone(),
            entered: self.entered.clone(),
        })
    }
}

/// [`GatedCrlSigner`]'s read side.
struct GatedCrlInfo {
    ca: Arc<dyn SignerInfo>,
    gate: Arc<tokio::sync::Semaphore>,
    entered: Arc<tokio::sync::Semaphore>,
}

#[async_trait]
impl SignerInfo for GatedCrlInfo {
    async fn crl_der(&self) -> Result<Option<Vec<u8>>, SignerError> {
        self.entered.add_permits(1);
        let _permit = self
            .gate
            .acquire()
            .await
            .expect("the gate is never closed while a test holds it");
        self.ca.crl_der().await
    }

    fn revocation_route(&self) -> RevocationRoute {
        self.ca.revocation_route()
    }
}

/// Issues for real, then refuses to revoke.
///
/// [`FailingSigner`] cannot reach the revocation path at all -- it fails
/// `issue` too, so no order ever gets a certificate to revoke. This one wraps a
/// real [`LocalCa`] so a test can genuinely issue, then drive the branch where
/// the CA-side action fails: `post_revoke_cert` calls `revoke` *before*
/// `Order::revoke`, precisely so a signer failure leaves the order un-revoked
/// and retryable rather than recording a revocation the CA never performed.
pub struct RevokeFailingSigner(pub Arc<LocalCa>);

impl RevokeFailingSigner {
    pub async fn new() -> Self {
        Self(Arc::new(memory_ca().await))
    }
}

#[async_trait]
impl SignerBackend for RevokeFailingSigner {
    async fn issue(
        &self,
        order_id: &str,
        csr_der: &[u8],
        identifiers: &[Identifier],
        validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        self.0.issue(order_id, csr_der, identifiers, validity).await
    }

    async fn revoke(&self, _cert_der: &[u8], _reason: Option<u32>) -> Result<(), SignerError> {
        Err(SignerError::Internal("the CA refused".to_string()))
    }

    fn info(&self) -> Arc<dyn SignerInfo> {
        self.0.info()
    }

    fn crl_refresher(&self) -> Option<Arc<dyn CrlRefresher>> {
        self.0.crl_refresher()
    }
}

/// A signer backend whose `revoke` **succeeds** and then closes the database
/// underneath the caller.
///
/// The sibling of [`RevokeFailingSigner`], for the opposite branch: the CA
/// withdrew trust and this server then failed to record it
/// (`certificate_revoke_persist_failed`). That branch has its own error event,
/// its own audit row and a carefully argued comment, and nothing reached it —
/// `tests/db_failures.rs` closes the pool up front, so those tests land on the
/// *lookup* long before the signer is called.
///
/// Closing the pool from inside `revoke` is what puts the failure between the
/// two writes, which is the only place this branch lives.
pub struct RevokePersistFailingSigner {
    ca: Arc<LocalCa>,
    /// Set after the app is built, since the builder is what creates the
    /// database this signer has to close.
    database: std::sync::OnceLock<Arc<Database>>,
    /// How many times `revoke` was called, so a retry can be shown to reach the
    /// signer again — the branch's whole claim is that a retry is expected.
    pub revocations: Arc<std::sync::atomic::AtomicUsize>,
}

impl RevokePersistFailingSigner {
    pub async fn new() -> Self {
        Self {
            ca: Arc::new(memory_ca().await),
            database: std::sync::OnceLock::new(),
            revocations: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
        }
    }

    /// Hands over the database to close, once the app that owns it exists.
    pub fn arm(&self, database: Arc<Database>) {
        let _ = self.database.set(database);
    }
}

#[async_trait]
impl SignerBackend for RevokePersistFailingSigner {
    async fn issue(
        &self,
        order_id: &str,
        csr_der: &[u8],
        identifiers: &[Identifier],
        validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        self.ca
            .issue(order_id, csr_der, identifiers, validity)
            .await
    }

    async fn revoke(&self, cert_der: &[u8], reason: Option<u32>) -> Result<(), SignerError> {
        self.revocations
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        // The CA-side action really happens...
        self.ca.revoke(cert_der, reason).await?;
        // ...and *then* the record of it becomes impossible.
        if let Some(database) = self.database.get() {
            database.close().await;
        }
        Ok(())
    }

    fn info(&self) -> Arc<dyn SignerInfo> {
        self.ca.info()
    }

    fn crl_refresher(&self) -> Option<Arc<dyn CrlRefresher>> {
        self.ca.crl_refresher()
    }
}

/// A signer backend carrying nothing but an `http-01` token store.
///
/// The responder route is mounted off `SignerInfo::http01_tokens`, so this
/// is how a test gets it onto the real `build_app` without standing up an
/// upstream ACME server. Issuance is deliberately unsupported: nothing that
/// uses this backend is about certificates.
pub struct TokenStoreSigner(pub Arc<DbTokenStore>);

impl TokenStoreSigner {
    /// Over its own throwaway database, with an hour before a token lapses.
    pub async fn new() -> Self {
        Self::over(Arc::new(Database::connect_in_memory().await.unwrap()))
    }

    /// Over `database` — the app's, for a test that publishes through one
    /// store and serves through another, the way two processes would.
    #[must_use]
    pub fn over(database: Arc<Database>) -> Self {
        Self(Arc::new(DbTokenStore::new(
            database,
            std::time::Duration::from_secs(3600),
        )))
    }
}

#[async_trait]
impl SignerBackend for TokenStoreSigner {
    async fn issue(
        &self,
        _order_id: &str,
        _csr_der: &[u8],
        _identifiers: &[Identifier],
        _validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        Err(SignerError::Internal("not a signing backend".to_string()))
    }

    async fn revoke(&self, _cert_der: &[u8], _reason: Option<u32>) -> Result<(), SignerError> {
        Ok(())
    }

    fn info(&self) -> Arc<dyn SignerInfo> {
        Arc::new(TokenStoreInfo(self.0.clone()))
    }
}

/// [`TokenStoreSigner`]'s read side: the store, and nothing else.
struct TokenStoreInfo(Arc<DbTokenStore>);

impl SignerInfo for TokenStoreInfo {
    fn http01_tokens(&self) -> Option<Arc<dyn Http01TokenStore>> {
        Some(self.0.clone())
    }

    fn revocation_route(&self) -> RevocationRoute {
        RevocationRoute::Delegated
    }
}

/// A signer backend that records whether `issue` was ever reached.
///
/// Exists for one property: `post_finalize` binds the CSR to its order
/// *before* handing anything to a backend. `local_ca` makes that check itself,
/// so a test using the real signer cannot tell the handler's refusal from the
/// backend's — and it is precisely the backends that do *not* check
/// (`custom` shells out to an operator script, `relay` relays to an
/// upstream that never saw the local authorizations) whose safety rests on the
/// handler getting there first. This backend refuses nothing, so if it is ever
/// called the CSR was accepted.
#[derive(Default)]
pub struct RecordingSigner {
    issued: std::sync::atomic::AtomicBool,
}

impl RecordingSigner {
    pub fn was_called(&self) -> bool {
        self.issued.load(std::sync::atomic::Ordering::SeqCst)
    }
}

#[async_trait]
impl SignerBackend for RecordingSigner {
    async fn issue(
        &self,
        _order_id: &str,
        _csr_der: &[u8],
        _identifiers: &[Identifier],
        _validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        self.issued.store(true, std::sync::atomic::Ordering::SeqCst);
        Err(SignerError::Internal("recorded".to_string()))
    }

    async fn revoke(&self, _cert_der: &[u8], _reason: Option<u32>) -> Result<(), SignerError> {
        Ok(())
    }
}

/// A signer backend that defers every issuance, the way a backend delegating
/// to an upstream CA does. It never finishes the work — a test using it is
/// asserting on what the *handler* does with `Processing`, not on any relay.
pub struct DelegatingSigner;

#[async_trait]
impl SignerBackend for DelegatingSigner {
    async fn issue(
        &self,
        _order_id: &str,
        _csr_der: &[u8],
        _identifiers: &[Identifier],
        _validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        Ok(IssueOutcome::Processing)
    }

    async fn revoke(&self, _cert_der: &[u8], _reason: Option<u32>) -> Result<(), SignerError> {
        Ok(())
    }
}

/// What a [`ScriptedAriSigner`] answers when the ARI handler asks it for a
/// renewal window.
#[derive(Clone, Copy)]
pub enum AriAnswer {
    /// The backend has an opinion — what a backend delegating to an upstream CA
    /// returns once that CA has answered.
    Window(i64, i64),
    /// The same, plus RFC 9773 §4.2's optional `explanationURL` — the case only
    /// a delegating backend can produce, since a locally computed window has
    /// nothing to explain.
    WindowWithExplanation(i64, i64, &'static str),
    /// The backend could not reach its upstream. The handler must fall back to
    /// its local estimate rather than failing the client's request.
    Unreachable,
}

/// A signer that issues real certificates through an in-memory local CA but
/// answers `renewal_info` from a script.
///
/// `LocalCa` takes the trait's default "no opinion" (`Ok(None)`), so without
/// this the two other arms of the handler's `match` are unreachable through the
/// router — there is no test app whose signer has an ARI opinion at all.
pub struct ScriptedAriSigner {
    inner: LocalCa,
    answer: AriAnswer,
}

impl ScriptedAriSigner {
    pub async fn new(answer: AriAnswer) -> Self {
        Self {
            inner: memory_ca().await,
            answer,
        }
    }
}

#[async_trait]
impl SignerBackend for ScriptedAriSigner {
    async fn issue(
        &self,
        order_id: &str,
        csr_der: &[u8],
        identifiers: &[Identifier],
        validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        self.inner
            .issue(order_id, csr_der, identifiers, validity)
            .await
    }

    async fn revoke(&self, cert_der: &[u8], reason: Option<u32>) -> Result<(), SignerError> {
        self.inner.revoke(cert_der, reason).await
    }

    fn crl_refresher(&self) -> Option<Arc<dyn CrlRefresher>> {
        self.inner.crl_refresher()
    }

    fn info(&self) -> Arc<dyn SignerInfo> {
        Arc::new(ScriptedAriInfo {
            ca: self.inner.info(),
            answer: self.answer,
        })
    }
}

/// [`ScriptedAriSigner`]'s read side: the CA's, with a scripted renewal opinion.
struct ScriptedAriInfo {
    ca: Arc<dyn SignerInfo>,
    answer: AriAnswer,
}

#[async_trait]
impl SignerInfo for ScriptedAriInfo {
    async fn crl_der(&self) -> Result<Option<Vec<u8>>, SignerError> {
        self.ca.crl_der().await
    }

    async fn ca_chain_pem(&self) -> Option<String> {
        self.ca.ca_chain_pem().await
    }

    async fn renewal_info(&self, _cert_der: &[u8]) -> Result<Option<RenewalWindow>, SignerError> {
        match self.answer {
            AriAnswer::Window(start, end) => Ok(Some(RenewalWindow::new(start, end))),
            AriAnswer::WindowWithExplanation(start, end, url) => Ok(Some(RenewalWindow {
                start,
                end,
                explanation_url: Some(url.to_string()),
            })),
            AriAnswer::Unreachable => {
                Err(SignerError::Internal("upstream unreachable".to_string()))
            }
        }
    }

    fn revocation_route(&self) -> RevocationRoute {
        self.ca.revocation_route()
    }
}

/// A signer whose `issue` succeeds but returns something that is not a PEM
/// chain — this server's own bug, since it just "issued" it. Drives the
/// finalize path's chain-parsing guards, which no real backend reaches.
pub struct GarbageChainSigner;

#[async_trait]
impl SignerBackend for GarbageChainSigner {
    async fn issue(
        &self,
        _order_id: &str,
        _csr_der: &[u8],
        _identifiers: &[Identifier],
        _validity: RequestedValidity,
    ) -> Result<IssueOutcome, SignerError> {
        Ok(IssueOutcome::Issued("not a certificate at all".to_string()))
    }

    async fn revoke(&self, _cert_der: &[u8], _reason: Option<u32>) -> Result<(), SignerError> {
        Ok(())
    }
}

/// A notify backend recording every event it receives, so a test can assert a
/// handler dispatches the right event with the right data. When built via
/// [`RecordingNotifyBackend::failing`], every `send` also returns an error —
/// used to prove a broken notify backend never affects the HTTP response.
#[derive(Default)]
pub struct RecordingNotifyBackend {
    events: Mutex<Vec<NotifyEvent>>,
    fail: bool,
}

impl RecordingNotifyBackend {
    pub fn failing() -> Self {
        Self {
            events: Mutex::new(Vec::new()),
            fail: true,
        }
    }

    /// Every event received so far, in order.
    pub fn events(&self) -> Vec<NotifyEvent> {
        self.events.lock().unwrap().clone()
    }
}

#[async_trait]
impl NotifyBackend for RecordingNotifyBackend {
    fn name(&self) -> &'static str {
        "recording"
    }

    async fn send(&self, event: &NotifyEvent) -> Result<(), NotifyError> {
        self.events.lock().unwrap().push(event.clone());
        if self.fail {
            Err(NotifyError::new("recording backend configured to fail"))
        } else {
            Ok(())
        }
    }
}

/// A check that refuses everything, at whichever hook it is configured for.
pub struct RejectingCheck {
    pub connections: bool,
    pub identifiers: bool,
    pub internal: bool,
}

impl RejectingCheck {
    /// Refuses at the connection hook.
    pub fn connections() -> Self {
        Self {
            connections: true,
            identifiers: false,
            internal: false,
        }
    }

    /// Refuses at the identifier hook.
    pub fn identifiers() -> Self {
        Self {
            connections: false,
            identifiers: true,
            internal: false,
        }
    }

    /// Fails to reach a decision at the connection hook.
    pub fn failing() -> Self {
        Self {
            connections: true,
            identifiers: false,
            internal: true,
        }
    }

    /// Fails to reach a decision at the *identifier* hook. Distinct from
    /// [`RejectingCheck::failing`]: the two hooks are mapped to problems by
    /// different callers (the middleware vs. the handlers).
    pub fn failing_identifiers() -> Self {
        Self {
            connections: false,
            identifiers: true,
            internal: true,
        }
    }

    fn refusal(&self) -> Verdict {
        if self.internal {
            Verdict::Undecided("resolver exploded".to_string())
        } else {
            Verdict::Fail("refused by test filter".to_string())
        }
    }
}

#[async_trait]
impl Check for RejectingCheck {
    fn kind(&self) -> &'static str {
        "rejecting"
    }

    /// Both, so the stage it actually refuses at is the one its constructor
    /// chose rather than one this declaration forces.
    fn stages(&self) -> StageSet {
        StageSet::both()
    }

    async fn check_connection(&self, _context: &ConnectionContext<'_>) -> Verdict {
        if self.connections {
            return self.refusal();
        }
        Verdict::Pass
    }

    async fn check_identifiers(&self, _context: &IdentifierContext<'_>) -> Verdict {
        if self.identifiers {
            return self.refusal();
        }
        Verdict::Pass
    }
}

/// A policy where **every** named check must pass — the all-must-pass shape
/// most tests want, without writing a rule per suite.
///
/// One rule per stage, each a conjunction of the checks that can decide there,
/// which is exactly what "all of these must pass" means once checks no longer
/// all answer at both hooks. A stage with no capable check gets no rule and
/// therefore allows, which is the policy engine's own law rather than a
/// special case here.
pub fn policy_of(checks: Vec<(String, Arc<dyn Check>)>) -> Arc<FilterPolicy> {
    let mut rules = Vec::new();
    for (stage, label) in [
        (Stage::Connection, "connection"),
        (Stage::Identifiers, "identifiers"),
    ] {
        let names: Vec<&str> = checks
            .iter()
            .filter(|(_, check)| check.stages().contains(stage))
            .map(|(name, _)| name.as_str())
            .collect();
        if names.is_empty() {
            continue;
        }
        rules.push(Rule {
            name: format!("all-{label}"),
            when: Condition::parse(&names.join(" and ")).expect("test condition should parse"),
            then: Effect::Allow,
            message: None,
            mode: Mode::Enforce,
        });
    }

    Arc::new(FilterPolicy::new(
        checks,
        rules,
        Effect::Deny,
        ProxyPolicy::default(),
    ))
}

/// [`policy_of`] for the common case of a single unnamed check.
pub fn policy_with(check: Arc<dyn Check>) -> Arc<FilterPolicy> {
    policy_of(vec![("only".to_string(), check)])
}

/// Sends a request as if it arrived from `peer`, by inserting the
/// `ConnectInfo` extension that `axum::serve` would normally add.
///
/// `oneshot` drives the router without a socket, so without this the filters
/// see no client address at all.
pub async fn send_from(app: &Router, request: Request<Body>, peer: &str) -> Response {
    let addr: SocketAddr = peer.parse().expect("peer must be an ip:port");
    let mut request = request;
    request.extensions_mut().insert(ConnectInfo(addr));
    app.clone().oneshot(request).await.unwrap()
}

/// POSTs a signed ACME body as if it arrived from `peer`.
pub async fn post_from(app: &Router, path: &str, body: String, peer: &str) -> Response {
    send_from(
        app,
        Request::post(path)
            .header("content-type", "application/jose+json")
            .body(Body::from(body))
            .unwrap(),
        peer,
    )
    .await
}

/// Fetches a nonce as if from `peer`, for suites where the filter would
/// otherwise refuse an address-less request.
pub async fn fetch_nonce_from(app: &Router, peer: &str) -> String {
    let res = send_from(
        app,
        Request::get(p("/newNonce")).body(Body::empty()).unwrap(),
        peer,
    )
    .await;

    res.headers()
        .get("replay-nonce")
        .expect("newNonce response must carry a Replay-Nonce header")
        .to_str()
        .unwrap()
        .to_string()
}

/// Reads a response body and parses it as JSON.
pub async fn body_json(response: Response) -> Value {
    let bytes = response.into_body().collect().await.unwrap().to_bytes();
    serde_json::from_slice(&bytes).unwrap()
}

/// Issues a real nonce by hitting the profile's `/newNonce` and returning its
/// `Replay-Nonce`
/// header value (minted and persisted by the middleware).
pub async fn fetch_nonce(app: &Router) -> String {
    let res = app
        .clone()
        .oneshot(Request::get(p("/newNonce")).body(Body::empty()).unwrap())
        .await
        .unwrap();

    res.headers()
        .get("replay-nonce")
        .expect("newNonce response must carry a Replay-Nonce header")
        .to_str()
        .unwrap()
        .to_string()
}

/// Assembles a flattened JWS body from an already-built `protected` header and
/// an already-encoded payload, signing the exact `protected.payload` input.
///
/// The three `build_jws*` wrappers below differ only in what goes in the header
/// (`jwk` vs `kid`) and whether the payload is encoded or empty, so the encoding
/// and signing live here once.
fn build_jws_parts(
    protected: Value,
    payload_b64: &str,
    sign: impl FnOnce(&[u8]) -> Vec<u8>,
) -> String {
    let protected_b64 = BASE64_URL_SAFE_NO_PAD.encode(serde_json::to_vec(&protected).unwrap());
    let signing_input = format!("{protected_b64}.{payload_b64}");
    let signature = sign(signing_input.as_bytes());
    flattened_jws(&protected_b64, payload_b64, &signature)
}

/// Base64url-encodes a JSON payload the way a JWS payload segment carries it.
fn encode_payload(payload: &Value) -> String {
    BASE64_URL_SAFE_NO_PAD.encode(serde_json::to_vec(payload).unwrap())
}

/// Assembles a flattened JWS body (`{protected, payload, signature}`) with an
/// embedded `jwk` — the `newAccount` form. Public so tests can craft off-nominal
/// envelopes (e.g. a JWK whose `alg` disagrees with the key type).
pub fn build_jws(
    alg: &str,
    jwk: Value,
    url: &str,
    nonce: &str,
    payload: &Value,
    sign: impl FnOnce(&[u8]) -> Vec<u8>,
) -> String {
    let protected = json!({ "alg": alg, "jwk": jwk, "nonce": nonce, "url": url });
    build_jws_parts(protected, &encode_payload(payload), sign)
}

/// Like [`build_jws`] but uses the `kid` (account URL) form of the protected
/// header instead of an embedded `jwk` — the shape RFC 8555 §6.2 requires for
/// requests to an existing account URL (e.g. account update).
pub fn build_jws_kid(
    alg: &str,
    kid: &str,
    url: &str,
    nonce: &str,
    payload: &Value,
    sign: impl FnOnce(&[u8]) -> Vec<u8>,
) -> String {
    let protected = json!({ "alg": alg, "kid": kid, "nonce": nonce, "url": url });
    build_jws_parts(protected, &encode_payload(payload), sign)
}

/// Like [`build_jws_kid`] but with an **empty** payload, the shape RFC 8555 §6.3
/// POST-as-GET requires. The signature covers the `protected.` input (empty
/// payload segment).
pub fn build_jws_kid_empty(
    alg: &str,
    kid: &str,
    url: &str,
    nonce: &str,
    sign: impl FnOnce(&[u8]) -> Vec<u8>,
) -> String {
    let protected = json!({ "alg": alg, "kid": kid, "nonce": nonce, "url": url });
    build_jws_parts(protected, "", sign)
}

/// Like [`build_jws`] but with no `nonce` in the protected header — the shape
/// RFC 8555 §7.3.5 requires for the *inner* JWS of a `keyChange` request
/// (whose protected header MUST omit `nonce`: it is bound into a request the
/// outer JWS already replay-protects, not a second signed request in its own
/// right).
pub fn build_jws_no_nonce(
    alg: &str,
    jwk: Value,
    url: &str,
    payload: &Value,
    sign: impl FnOnce(&[u8]) -> Vec<u8>,
) -> String {
    let protected = json!({ "alg": alg, "jwk": jwk, "url": url });
    build_jws_parts(protected, &encode_payload(payload), sign)
}

/// Assembles the flattened JWS JSON string from already-encoded parts. Lets
/// tests supply a raw (possibly invalid) `payload_b64` while keeping a signature
/// that matches the exact `protected.payload` signing input.
pub fn flattened_jws(protected_b64: &str, payload_b64: &str, signature: &[u8]) -> String {
    json!({
        "protected": protected_b64,
        "payload": payload_b64,
        "signature": BASE64_URL_SAFE_NO_PAD.encode(signature),
    })
    .to_string()
}

/// Builds the inner EAB JWS object (RFC 8555 §7.3.4): protected header
/// `{alg: "HS256", kid, url}` (no nonce), payload = the account's own JWK (as
/// returned by [`TestSigner::jwk`]), HMAC-signed with `hmac_secret`. Returns
/// the JSON value to embed as `externalAccountBinding` in a `newAccount`
/// payload.
pub fn build_eab(kid: &str, hmac_secret: &[u8], url: &str, account_jwk: &Value) -> Value {
    let protected = json!({ "alg": "HS256", "kid": kid, "url": url });
    let protected_b64 = BASE64_URL_SAFE_NO_PAD.encode(serde_json::to_vec(&protected).unwrap());
    let payload_b64 = BASE64_URL_SAFE_NO_PAD.encode(serde_json::to_vec(account_jwk).unwrap());
    let signing_input = format!("{protected_b64}.{payload_b64}");

    let key = hmac::Key::new(hmac::HMAC_SHA256, hmac_secret);
    let signature = hmac::sign(&key, signing_input.as_bytes());

    json!({
        "protected": protected_b64,
        "payload": payload_b64,
        "signature": BASE64_URL_SAFE_NO_PAD.encode(signature.as_ref()),
    })
}

/// What a test needs from a key to drive a signed ACME request.
///
/// `EcSigner` and `RsaSigner` differ only in their `alg`, how they expose a JWK,
/// and how they sign bytes; every envelope shape is built the same way from
/// those three. Keeping the shapes here rather than on each signer means a test
/// helper can be generic over the key type — which is what lets the order
/// lifecycle and the account-update tests run against both EC and RSA without
/// being written twice.
pub trait TestSigner {
    /// The JWS `alg` this key signs with.
    fn alg(&self) -> &'static str;
    /// The public key as a JWK, for the embedded-`jwk` (newAccount) form.
    fn jwk(&self) -> Value;
    /// Raw signature over `input`, for crafting custom JWS envelopes.
    fn sign_input(&self, input: &[u8]) -> Vec<u8>;

    /// A flattened JWS body with an embedded `jwk`.
    fn sign(&self, url: &str, nonce: &str, payload: &Value) -> String {
        build_jws(self.alg(), self.jwk(), url, nonce, payload, |input| {
            self.sign_input(input)
        })
    }

    /// A flattened JWS body in the `kid` (account URL) form.
    fn sign_kid(&self, kid: &str, url: &str, nonce: &str, payload: &Value) -> String {
        build_jws_kid(self.alg(), kid, url, nonce, payload, |input| {
            self.sign_input(input)
        })
    }

    /// A `kid`-form POST-as-GET body (empty payload).
    fn sign_kid_empty(&self, kid: &str, url: &str, nonce: &str) -> String {
        build_jws_kid_empty(self.alg(), kid, url, nonce, |input| self.sign_input(input))
    }

    /// A flattened JWS body with an embedded `jwk` and **no `nonce`** — the
    /// `keyChange` (RFC 8555 §7.3.5) inner JWS shape, self-signed by this key
    /// (the *new* account key in a rollover).
    fn sign_inner(&self, url: &str, payload: &Value) -> String {
        build_jws_no_nonce(self.alg(), self.jwk(), url, payload, |input| {
            self.sign_input(input)
        })
    }
}

/// A freshly generated EC (P-256 / ES256) signer.
pub struct EcSigner {
    key_pair: EcdsaKeyPair,
    rng: SystemRandom,
}

impl EcSigner {
    pub fn new() -> Self {
        let rng = SystemRandom::new();
        let pkcs8 = EcdsaKeyPair::generate_pkcs8(&signature::ECDSA_P256_SHA256_FIXED_SIGNING, &rng)
            .unwrap();
        Self::from_pkcs8(pkcs8.as_ref())
    }

    /// Builds a signer from existing PKCS8 EC key material — used when a test
    /// needs the same key wrapped both as a JWS signer (this) and as an
    /// `rcgen::KeyPair` (to self-sign a CSR), e.g. "revoke via the
    /// certificate's own keypair". See [`make_csr_and_keypair`].
    pub fn from_pkcs8(pkcs8_der: &[u8]) -> Self {
        let rng = SystemRandom::new();
        let key_pair =
            EcdsaKeyPair::from_pkcs8(&signature::ECDSA_P256_SHA256_FIXED_SIGNING, pkcs8_der, &rng)
                .unwrap();
        Self { key_pair, rng }
    }
}

impl TestSigner for EcSigner {
    fn alg(&self) -> &'static str {
        "ES256"
    }

    /// The JWK (EC public key) derived from this signer's SEC1 point `04 || x || y`.
    fn jwk(&self) -> Value {
        let sec1 = self.key_pair.public_key().as_ref();
        json!({
            "kty": "EC",
            "crv": "P-256",
            "x": BASE64_URL_SAFE_NO_PAD.encode(&sec1[1..33]),
            "y": BASE64_URL_SAFE_NO_PAD.encode(&sec1[33..65]),
        })
    }

    fn sign_input(&self, input: &[u8]) -> Vec<u8> {
        self.key_pair
            .sign(&self.rng, input)
            .unwrap()
            .as_ref()
            .to_vec()
    }
}

/// An RSA (RS256) signer backed by the shared 2048-bit test key fixture.
pub struct RsaSigner {
    key_pair: RsaKeyPair,
    rng: SystemRandom,
}

impl RsaSigner {
    pub fn new() -> Self {
        let pkcs8 = include_bytes!("../fixtures/rsa_test_key.pk8");
        let key_pair = RsaKeyPair::from_pkcs8(pkcs8).unwrap();
        Self {
            key_pair,
            rng: SystemRandom::new(),
        }
    }
}

impl TestSigner for RsaSigner {
    fn alg(&self) -> &'static str {
        "RS256"
    }

    /// The JWK (RSA public key), deriving `n`/`e` from the PKCS#1 public key DER
    /// `SEQUENCE { INTEGER n, INTEGER e }`.
    fn jwk(&self) -> Value {
        let blocks = simple_asn1::from_der(self.key_pair.public_key().as_ref()).unwrap();
        let (n, e) = match &blocks[0] {
            ASN1Block::Sequence(_, items) => {
                let int = |block: &ASN1Block| match block {
                    ASN1Block::Integer(_, v) => v.to_bytes_be().1,
                    _ => panic!("expected INTEGER in RSA public key"),
                };
                (int(&items[0]), int(&items[1]))
            }
            _ => panic!("unexpected RSA public key DER structure"),
        };
        json!({
            "kty": "RSA",
            "n": BASE64_URL_SAFE_NO_PAD.encode(&n),
            "e": BASE64_URL_SAFE_NO_PAD.encode(&e),
        })
    }

    /// Raw RS256 signature over `input` (256-byte, 2048-bit key).
    fn sign_input(&self, input: &[u8]) -> Vec<u8> {
        let mut sig = vec![0u8; 256];
        self.key_pair
            .sign(&signature::RSA_PKCS1_SHA256, &self.rng, input, &mut sig)
            .unwrap();
        sig
    }
}

/// A scratch directory that removes itself on drop, so a failing assertion
/// cannot leave files behind.
///
/// The library has its own copy under `crates/signer/src/testutil.rs`; an integration test
/// cannot see a `#[cfg(test)]` item of the crate it links against, and making
/// that one a real `pub mod` would ship test scaffolding to every consumer.
/// Two copies, deliberately — down from seven.
pub struct TempDir(std::path::PathBuf);

impl TempDir {
    pub fn new(label: &str) -> Self {
        let path =
            std::env::temp_dir().join(format!("acme-proxy-{label}-{}", uuid::Uuid::now_v7()));
        std::fs::create_dir_all(&path).unwrap();
        Self(path)
    }

    pub fn path(&self) -> &std::path::Path {
        &self.0
    }

    pub fn join(&self, name: &str) -> std::path::PathBuf {
        self.0.join(name)
    }
}

impl Drop for TempDir {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}

/// Writes an executable script and returns its path.
///
/// See `crates/signer/src/testutil.rs::write_script` for why this suite must run under
/// `cargo nextest` rather than `cargo test`: every caller exec's a file it has
/// just written, which intermittently hits `ETXTBSY` when tests share one
/// process.
#[cfg(unix)]
pub fn write_script(dir: &TempDir, name: &str, body: &str) -> std::path::PathBuf {
    use std::os::unix::fs::PermissionsExt;

    let path = dir.join(name);
    std::fs::write(&path, body).unwrap();
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
    path
}

/// The ACME ladder every suite climbs: register, order, prove, finalize, read.
///
/// Nineteen suites had each grown their own copy of these — `post` in
/// twenty-two files, `register` in eighteen, `issue_certificate` in four at
/// roughly fifty lines apiece. They were byte-identical or differed only in
/// whitespace, which is the worst kind of duplication: it reads as deliberate.
///
/// The cost was not the lines. `assert_problem` existed in **two** of the
/// nineteen, so most suites asserted a bare status code on a refusal — and a
/// regression that returned the right status with the wrong `type` passed.
/// With the helper one `use` away, checking the problem type is the cheap
/// option rather than the thorough one.
pub mod acme {
    use super::{BASE, HOST, TestSigner, body_json, fetch_nonce, p};
    use axum::Router;
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use axum::response::Response;
    use http_body_util::BodyExt;
    use serde_json::{Value, json};
    use tower::ServiceExt;

    /// The absolute URL of a resource under the test profile — what a JWS
    /// `url` header must name (RFC 8555 §6.4).
    #[must_use]
    pub fn url(path: &str) -> String {
        format!("{BASE}{path}")
    }

    /// POSTs a signed body, with the media type §6.2 requires.
    pub async fn post(app: &Router, path: &str, body: String) -> Response {
        app.clone()
            .oneshot(
                Request::post(path)
                    .header("content-type", "application/jose+json")
                    .body(Body::from(body))
                    .unwrap(),
            )
            .await
            .unwrap()
    }

    /// The response body as text, for the assertions that are about wording.
    pub async fn body_text(response: Response) -> String {
        let bytes = response.into_body().collect().await.unwrap().to_bytes();
        String::from_utf8(bytes.to_vec()).unwrap()
    }

    /// Asserts an RFC 8555 problem document: the status **and** the `type`.
    ///
    /// Both halves matter. A refusal that kept its status and changed its type
    /// is a different answer to the client — `unauthorized` and
    /// `rejectedIdentifier` are both `403` and mean opposite things about
    /// whether retrying could ever work.
    pub async fn assert_problem(response: Response, status: StatusCode, typ: &str) {
        assert_eq!(response.status(), status);
        let problem = body_json(response).await;
        assert_eq!(problem["type"], typ, "problem type");
    }

    /// The same, plus a substring the `detail` must carry — for the refusals
    /// whose wording is the point (an operator's `filter.rule.message`, a name
    /// the server is meant to quote back).
    pub async fn assert_problem_detail(
        response: Response,
        status: StatusCode,
        typ: &str,
        detail: &str,
    ) {
        assert_eq!(response.status(), status);
        let problem = body_json(response).await;
        assert_eq!(problem["type"], typ, "problem type");
        let got = problem["detail"].as_str().unwrap_or_default();
        assert!(
            got.contains(detail),
            "detail {got:?} does not contain {detail:?}"
        );
    }

    /// Registers an account and returns its URL — the `kid` of every request
    /// after this one.
    pub async fn register(app: &Router, signer: &impl TestSigner) -> String {
        let nonce = fetch_nonce(app).await;
        let payload = json!({ "termsOfServiceAgreed": true });
        let response = post(
            app,
            &p("/newAccount"),
            signer.sign(&url("/newAccount"), &nonce, &payload),
        )
        .await;
        assert_eq!(response.status(), StatusCode::CREATED);
        location(&response)
    }

    /// The `Location` header, which `newAccount` and `newOrder` both set.
    pub fn location(response: &Response) -> String {
        response
            .headers()
            .get("location")
            .and_then(|value| value.to_str().ok())
            .expect("this response must set a Location header")
            .to_string()
    }

    /// An absolute URL the server handed back, as a request path.
    ///
    /// Strips [`HOST`] and not [`BASE`]: the `/profile/<name>` prefix is part
    /// of the route, so removing it would ask the root router instead.
    #[must_use]
    pub fn path_of(absolute: &str) -> String {
        absolute
            .strip_prefix(HOST)
            .unwrap_or_else(|| panic!("{absolute} is not under {HOST}"))
            .to_string()
    }

    /// Places an order for `names`, returning `(order_url, order_object)`.
    pub async fn new_order(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        names: &[&str],
    ) -> (String, Value) {
        let identifiers: Vec<Value> = names
            .iter()
            .map(|name| json!({ "type": "dns", "value": name }))
            .collect();
        let nonce = fetch_nonce(app).await;
        let body = signer.sign_kid(
            account_url,
            &url("/newOrder"),
            &nonce,
            &json!({ "identifiers": identifiers }),
        );
        let response = post(app, &p("/newOrder"), body).await;
        assert_eq!(response.status(), StatusCode::CREATED);
        let order_url = location(&response);
        (order_url, body_json(response).await)
    }

    /// Reads any resource by signed POST-as-GET (§6.3), asserting a `200`.
    pub async fn post_as_get(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        absolute_url: &str,
    ) -> Value {
        let response = post_as_get_raw(app, signer, account_url, absolute_url).await;
        assert_eq!(response.status(), StatusCode::OK);
        body_json(response).await
    }

    /// The same without the status assertion, for the tests that are *about*
    /// the refusal.
    pub async fn post_as_get_raw(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        absolute_url: &str,
    ) -> Response {
        let nonce = fetch_nonce(app).await;
        let body = signer.sign_kid_empty(account_url, absolute_url, &nonce);
        post(app, &path_of(absolute_url), body).await
    }

    /// Triggers a challenge with the `{}` payload §7.5.1 defines.
    pub async fn trigger(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        challenge_url: &str,
    ) -> Response {
        let nonce = fetch_nonce(app).await;
        let body = signer.sign_kid(account_url, challenge_url, &nonce, &json!({}));
        post(app, &path_of(challenge_url), body).await
    }

    /// Polls a challenge until the server has decided it, and returns it.
    ///
    /// Validation is queued work: the trigger claims the challenge and answers
    /// `processing` (§7.1.6), and the verdict lands moments later in the job
    /// runner. A real client polls exactly this way — §8.2 pairs `processing`
    /// with the `Retry-After` the trigger carries — and so does the suite,
    /// rather than reaching into the queue, so what is exercised is the path a
    /// certbot run takes.
    ///
    /// Polled by re-POSTing `{}`, not POST-as-GET: §7.5.1 defines the challenge
    /// resource's POST body as the trigger and this server accepts no empty
    /// payload there, and the same section makes a retry explicitly *not* a
    /// state change — the claim simply fails and the object comes back as it
    /// stands.
    ///
    /// Bounded, so a challenge that never settles fails the test with a
    /// diagnosis instead of hanging it.
    pub async fn await_challenge(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        challenge_url: &str,
    ) -> Value {
        for _ in 0..600 {
            let response = trigger(app, signer, account_url, challenge_url).await;
            let challenge = crate::common::body_json(response).await;
            if challenge["status"] != "processing" {
                return challenge;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        panic!("`{challenge_url}` never left `processing`");
    }

    /// Triggers a challenge and waits for its verdict, returning the challenge.
    ///
    /// What a suite proving domain control wants: [`trigger`] alone now only
    /// starts the work.
    pub async fn trigger_and_settle(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        challenge_url: &str,
    ) -> Value {
        let response = trigger(app, signer, account_url, challenge_url).await;
        // Both outcomes are `200` + the challenge object (§7.5.1), so this
        // asserts the transport, not the verdict.
        assert_eq!(response.status(), StatusCode::OK);
        await_challenge(app, signer, account_url, challenge_url).await
    }

    /// Drives an order from `pending` to `ready` by proving every
    /// authorization, and returns the order object.
    ///
    /// Assumes the bypassing registry every suite's default app carries, where
    /// triggering a challenge is what makes it `valid`.
    pub async fn drive_to_ready(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        order: &Value,
    ) {
        let authorizations: Vec<String> = order["authorizations"]
            .as_array()
            .expect("an order carries its authorizations")
            .iter()
            .map(|value| value.as_str().unwrap().to_string())
            .collect();

        for authz_url in &authorizations {
            let authz = post_as_get(app, signer, account_url, authz_url).await;
            let challenge_url = authz["challenges"][0]["url"].as_str().unwrap().to_string();
            trigger_and_settle(app, signer, account_url, &challenge_url).await;
        }
    }

    /// Registers, orders `names` and proves every authorization, returning
    /// `(account_url, order_url, order)` with the order in `ready`.
    pub async fn ready_order(
        app: &Router,
        signer: &impl TestSigner,
        names: &[&str],
    ) -> (String, String, Value) {
        let account_url = register(app, signer).await;
        let (order_url, order) = new_order(app, signer, &account_url, names).await;
        drive_to_ready(app, signer, &account_url, &order).await;
        let order = post_as_get(app, signer, &account_url, &order_url).await;
        (account_url, order_url, order)
    }

    /// POST-as-GETs `order_url` until it leaves `processing`, returning the
    /// order — what a client does after finalizing, RFC 8555 §7.4.
    pub async fn await_order(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        order_url: &str,
    ) -> Value {
        for _ in 0..600 {
            let order = post_as_get(app, signer, account_url, order_url).await;
            if order["status"] != "processing" {
                return order;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        panic!("`{order_url}` never left `processing`");
    }

    /// Finalizes an order and waits for the issuance it queued, returning the
    /// settled order — `valid` with a certificate URL, or `invalid` with the
    /// reason as its `error`.
    ///
    /// Asserts the finalize request itself was accepted: a synchronous refusal
    /// is [`finalize`]'s to test.
    pub async fn finalize_and_settle(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        order_url: &str,
        names: &[&str],
    ) -> Value {
        let response = finalize(
            app,
            signer,
            account_url,
            &format!("{order_url}/finalize"),
            names,
        )
        .await;
        assert_eq!(response.status(), StatusCode::OK, "finalize");
        await_order(app, signer, account_url, order_url).await
    }

    /// Finalizes an order with a CSR for `names`, returning the raw response.
    ///
    /// A `200` here answers `processing`: the certificate is issued by the
    /// worker, and [`await_order`] is how a test waits for it.
    pub async fn finalize(
        app: &Router,
        signer: &impl TestSigner,
        account_url: &str,
        finalize_url: &str,
        names: &[&str],
    ) -> Response {
        let csr = super::make_csr_for(names);
        let nonce = fetch_nonce(app).await;
        let body = signer.sign_kid(account_url, finalize_url, &nonce, &json!({ "csr": csr }));
        post(app, &path_of(finalize_url), body).await
    }

    /// The whole flow: register, order, prove, finalize, fetch the chain.
    ///
    /// Returns `(account_url, order_url, certificate_pem)`.
    pub async fn issue_certificate(
        app: &Router,
        signer: &impl TestSigner,
        names: &[&str],
    ) -> (String, String, String) {
        let (account_url, order_url, order) = ready_order(app, signer, names).await;
        let finalize_url = order["finalize"].as_str().unwrap().to_string();

        let response = finalize(app, signer, &account_url, &finalize_url, names).await;
        assert_eq!(response.status(), StatusCode::OK, "finalize");

        let order = await_order(app, signer, &account_url, &order_url).await;
        let certificate_url = order["certificate"]
            .as_str()
            .expect("a valid order carries a certificate URL")
            .to_string();

        let nonce = fetch_nonce(app).await;
        let body = signer.sign_kid_empty(&account_url, &certificate_url, &nonce);
        let response = post(app, &path_of(&certificate_url), body).await;
        assert_eq!(response.status(), StatusCode::OK, "certificate");
        let pem = body_text(response).await;

        (account_url, order_url, pem)
    }
}

/// An order under `account` holding a certificate that expires at `not_after`
/// (`None`: never stamped, which counts as live).
pub async fn certified_order(
    database: &acme_proxy_store::db::Database,
    account: uuid::Uuid,
    not_after: Option<i64>,
) -> acme_proxy_store::order::Order {
    use acme_proxy_core::identifier::Identifier;
    use acme_proxy_store::order::Order;

    let mut order = Order::create(
        "default",
        account,
        vec![Identifier::dns("live.example.com")],
        2_000_000_000,
        None,
        None,
        database,
    )
    .await
    .unwrap();
    order
        .finalize(
            "-----BEGIN CERTIFICATE-----\n...".to_string(),
            order.id.simple().to_string(),
            vec![1],
            not_after,
            database,
        )
        .await
        .unwrap();
    order
}

/// A credential and `count` accounts registered with it, each holding a
/// certificate that expires at `not_after`.
pub async fn bound_eab(
    database: &acme_proxy_store::db::Database,
    count: u8,
    not_after: Option<i64>,
) -> (String, Vec<uuid::Uuid>) {
    use acme_proxy_store::account::Account;
    use acme_proxy_store::eab::Eab;

    let eab = Eab::create(Some("tenant".to_string()), None, database)
        .await
        .unwrap();
    let mut accounts = Vec::new();
    for index in 0..count {
        let (mut account, _) = Account::find_or_create(
            "default",
            &[eab.kid.as_bytes()[15], index, 99],
            vec![],
            &acme_proxy_core::audit::ClientContext::default(),
            database,
        )
        .await
        .unwrap();
        account.set_eab_kid(eab.kid, database).await.unwrap();
        certified_order(database, account.id, not_after).await;
        accounts.push(account.id);
    }
    (eab.kid.to_string(), accounts)
}