didbot_dns/

route53.rs

//! [`Route53Dns`]: the production backend that writes a real hosted zone.
//!
//! # Why hand-rolled SigV4 over `aws-sdk-route53`
//!
//! The official SDK is async-only (it is generated from Smithy models built
//! on `hyper` and `tokio`), which conflicts with this crate's synchronous
//! trait — see the crate-level docs for why that is deliberate. Using it
//! would mean embedding a private Tokio runtime inside every provider just to
//! `block_on` each call, and pulling in `aws-config`, `aws-runtime`,
//! `aws-smithy-runtime`, `aws-smithy-xml`, `aws-sigv4` and their own copies of
//! `hyper`/`rustls` — none of which are in `Cargo.lock` today.
//!
//! Route53's control-plane API is a small, stable REST/XML surface: five
//! operations, one signing scheme. `Cargo.lock` already carries everything
//! signing it needs — `reqwest` (with `rustls-tls`, already selected), `hmac`
//! and `sha2` (both used elsewhere in this workspace for other purposes) —
//! so the addition for this feature is `reqwest`'s `blocking` client feature,
//! `hex`, `time` and `quick-xml`. The request bodies and the response shapes
//! this provider reads are serde types on the private `xml` module this file
//! defines, so the XML itself is `quick-xml`'s to escape and to parse.
//!
//! # Eventual consistency: a challenge publish waits, a withdrawal does not
//!
//! Every Route53 change is accepted as `PENDING` and becomes `INSYNC` some
//! seconds later, once it has propagated to every edge name server. The
//! caller of [`DnsProvider::publish_txt`] is about to ask a certificate
//! authority to look for the value it just published, so this provider
//! polls
//! [`GetChange`](https://docs.aws.amazon.com/Route53/latest/APIReference/API_GetChange.html)
//! until the change reports `INSYNC`, up to a configurable timeout, and only
//! returns `Ok` once it does — never optimistically. A caller that gets `Ok`
//! back has a value a resolver can see.
//!
//! Withdrawal makes the opposite call. A challenge value still visible for a
//! few more seconds after the order is done costs nothing, so
//! [`DnsProvider::withdraw_txt`] and [`DnsProvider::publish_caa`] return as
//! soon as Route53 accepts the change, not once it has propagated.
//!
//! # Rate limits and retries
//!
//! Route53 throttles the change APIs (`ChangeResourceRecordSets`) far below
//! the read APIs, and returns `Throttling` or `PriorRequestNotComplete` when
//! a caller exceeds it. Every call in this module retries those, and a
//! `5xx`, with capped exponential backoff (see [`RetryPolicy`]); every other
//! API error — including the one this provider maps to
//! [`DnsError::NotPublished`] — is not retried, because retrying a real
//! conflict cannot make it stop being one.
//!
//! # Hosted-zone record limits
//!
//! A Route53 hosted zone defaults to a 10,000-record-set quota. Under the
//! wildcard shape `docs/names.md` draws, the zone holds a handful of record
//! sets whatever the account count, and the one this provider creates and
//! deletes on every renewal is the `_acme-challenge.<zone>` `TXT`. Nothing
//! in this module enforces the quota locally — `ChangeResourceRecordSets`
//! reports it as a normal API error, which surfaces as
//! [`DnsError::Backend`].
//!
//! # The credential
//!
//! Scope the IAM credential this provider is given to the records it
//! writes in the one hosted zone:
//!
//! ```text
//! {
//!   "Version": "2012-10-17",
//!   "Statement": [
//!     { "Effect": "Allow",
//!       "Action": ["route53:ChangeResourceRecordSets"],
//!       "Resource": "arn:aws:route53:::hostedzone/<HOSTED_ZONE_ID>",
//!       "Condition": { "ForAllValues:StringEquals": {
//!         "route53:ChangeResourceRecordSetsRecordTypes":
//!           ["TXT", "CAA"],
//!         "route53:ChangeResourceRecordSetsNormalizedRecordNames":
//!           ["<zone>", "_acme-challenge.<zone>"] } } },
//!     { "Effect": "Allow",
//!       "Action": ["route53:GetChange"],
//!       "Resource": "arn:aws:route53:::change/*" }
//!   ]
//! }
//! ```
//!
//! Both condition keys match per change batch — every record name and type
//! in the batch must be in the sets — and the names Route53 compares are
//! normalized: lowercase, no trailing dot. [`Route53Dns`] sends one record
//! set per batch, each one a pair from those sets.
//!
//! `docs/deployment.md` already says why: control of the zone is control of
//! every DID under it, so this credential is as powerful as the signing
//! keys, and it should be scoped, rotated and kept off the host that holds
//! those keys the same way.

use crate::{CaaRecords, CaaValue, DnsError, DnsProvider, TxtRecords};
use didbot_identity::hostname_is_at_or_below;
use hmac::{Hmac, Mac};
use sha2::Sha256;
use std::collections::BTreeSet;
use std::thread;
use std::time::{Duration, Instant};

const DEFAULT_ENDPOINT: &str = "https://route53.amazonaws.com";
const API_VERSION: &str = "2013-04-01";

/// Long-lived AWS credentials, scoped to one hosted zone.
///
/// See the module docs for the IAM policy this credential should carry.
/// Nothing here reads one from the environment or a file —
/// [`InstanceCredentials`] is the one credential path this crate offers a
/// deployment, and it is ambient rather than configured: whatever role is
/// attached to the running instance, read from IMDS and re-read as it
/// expires.
#[derive(Clone)]
pub struct Route53Credentials {
    /// The `AKIA…`/`ASIA…` access key id.
    pub access_key_id: String,
    /// The paired secret key. Never logged, never included in a `Debug`.
    pub secret_access_key: String,
    /// The session token, when the credential is temporary (an assumed
    /// role or an instance profile).
    pub session_token: Option<String>,
}

impl std::fmt::Debug for Route53Credentials {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Route53Credentials")
            .field("access_key_id", &self.access_key_id)
            .field("secret_access_key", &"...")
            .field("session_token", &self.session_token.as_ref().map(|_| "..."))
            .finish()
    }
}

/// Why credentials could not be fetched from the instance metadata service.
#[derive(Debug, thiserror::Error)]
pub enum ImdsError {
    /// A request to IMDS itself failed — no route to
    /// `169.254.169.254` (not running on EC2, or the link-local route is
    /// firewalled), a timeout, or a non-2xx status.
    #[error("could not reach the instance metadata service: {0}")]
    Request(String),
    /// IMDS answered, but not with something this reads as credentials —
    /// most often no IAM role attached to the instance at all.
    #[error("instance metadata service response was not usable: {0}")]
    Response(String),
}

/// A credential read from IMDS, with the moment IMDS says it stops being
/// accepted.
#[derive(Clone, Debug)]
pub(crate) struct TimedCredentials {
    pub(crate) credentials: Route53Credentials,
    /// `None` when the response carried no `Expiration` — a credential that
    /// states no end, and is signed with until the process ends.
    pub(crate) expires_at: Option<time::OffsetDateTime>,
}

/// The IMDS read [`InstanceCredentials`] repeats, isolated the same way
/// [`Transport`] is and for the same reason: there is no instance metadata
/// service in this environment, so tests drive the holder against a fake
/// that expires on command.
pub(crate) trait CredentialFetch: Send + Sync {
    fn fetch(&self) -> Result<TimedCredentials, ImdsError>;
}

/// The real one: three IMDSv2 round trips against `169.254.169.254`.
struct Imds;

impl CredentialFetch for Imds {
    fn fetch(&self) -> Result<TimedCredentials, ImdsError> {
        let client = imds_client().map_err(|err| ImdsError::Request(err.to_string()))?;

        let token = imds_request(
            client
                .put(format!("{IMDS_ENDPOINT}/latest/api/token"))
                .header("X-aws-ec2-metadata-token-ttl-seconds", "21600"),
        )?;

        let role = imds_request(
            client
                .get(format!(
                    "{IMDS_ENDPOINT}/latest/meta-data/iam/security-credentials/"
                ))
                .header("X-aws-ec2-metadata-token", &token),
        )?;
        let role = role.trim();
        if role.is_empty() {
            return Err(ImdsError::Response(
                "instance has no IAM role attached".to_string(),
            ));
        }

        let body = imds_request(
            client
                .get(format!(
                    "{IMDS_ENDPOINT}/latest/meta-data/iam/security-credentials/{role}"
                ))
                .header("X-aws-ec2-metadata-token", &token),
        )?;
        parse_imds_credentials(&body)
    }
}

/// How far ahead of a credential's stated expiry a refresh is made.
///
/// AWS rotates an instance profile's credential well before the one it
/// handed out expires, so a read this far ahead gets the next one. The
/// margin also covers the clock skew between this host and the signing
/// service, and leaves room for the three round trips the refresh itself
/// costs.
pub const REFRESH_MARGIN: Duration = Duration::from_secs(300);

/// The credential every Route53 request in this process is signed with,
/// kept current against the instance profile that issues it.
///
/// An instance profile's credential is temporary: IMDS states an
/// `Expiration`, and a signature made with a credential past it is rejected
/// by Route53 — which on this deployment stops agent hostnames being
/// published and stops ACME DNS-01 renewing the certificate, since
/// `terraform-aws-didbot-pds/security_groups.tf` opens no port 80 for an HTTP-01 order.
/// This holder reads IMDS again once the credential it holds is within
/// [`REFRESH_MARGIN`] of that expiry, and hands out the new one.
///
/// A refresh that fails is reported at `error` and tried again on the next
/// request rather than ending the process: while the credential in hand is
/// still valid it keeps signing with it, and once it is not,
/// [`Self::current`] returns the [`ImdsError`], which reaches a caller as a
/// [`DnsError::Backend`] through the same retry path as any other transport
/// failure.
///
/// The refresh happens under the lock on the held credential, so callers
/// that arrive during one wait for it and then sign with what it produced:
/// a hosted zone being written by several threads makes one IMDS read
/// between them, not one each.
pub struct InstanceCredentials {
    fetch: Box<dyn CredentialFetch>,
    held: std::sync::Mutex<TimedCredentials>,
}

impl InstanceCredentials {
    /// Reads the credential of whatever IAM role is attached to this EC2
    /// instance, over IMDSv2, and keeps reading it as it expires.
    ///
    /// IMDSv2 specifically, not v1: v1's tokenless `GET` is what let the
    /// 2019 Capital One breach turn an unrelated request-smuggling bug into
    /// stolen instance-role credentials, and `infra/pds/`'s launch template
    /// requires the token hop (`HttpTokens = required`) for exactly that
    /// reason — see `docs/deployment.md`. This is the one credential path
    /// this crate offers a production deployment: there is no environment
    /// variable or file this reads instead, because inventing a second path
    /// is a second thing to misconfigure and a second place a credential can
    /// leak from.
    ///
    /// Three round trips, each against a 5-second timeout: the session
    /// token, the attached role's name, and that role's current
    /// credentials. Blocking, like the rest of this provider — see the
    /// module docs for why. The first read happens here, so a deployment
    /// whose instance profile is wrong fails at startup rather than at the
    /// first publish.
    pub fn from_instance_metadata() -> Result<Self, ImdsError> {
        Self::with_fetch(Box::new(Imds))
    }

    pub(crate) fn with_fetch(fetch: Box<dyn CredentialFetch>) -> Result<Self, ImdsError> {
        let held = fetch.fetch()?;
        Ok(Self {
            fetch,
            held: std::sync::Mutex::new(held),
        })
    }

    /// The credential to sign the next request with, refreshed first if the
    /// one held is inside [`REFRESH_MARGIN`] of its expiry.
    pub fn current(&self) -> Result<Route53Credentials, ImdsError> {
        // Recovering from a poisoned mutex for the same reason
        // [`Route53Dns::record_sets_slot`] does: what is under it is one
        // value replaced in one operation, so a panic elsewhere cannot have
        // left it half-written.
        let mut held = self
            .held
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        let now = time::OffsetDateTime::now_utc();
        let Some(expires_at) = held.expires_at else {
            return Ok(held.credentials.clone());
        };
        if now + REFRESH_MARGIN < expires_at {
            return Ok(held.credentials.clone());
        }

        match self.fetch.fetch() {
            Ok(fresh) => {
                tracing::info!(
                    expires_at = fresh.expires_at.map(|at| at.to_string()),
                    "refreshed the route53 credential from instance metadata"
                );
                let credentials = fresh.credentials.clone();
                *held = fresh;
                Ok(credentials)
            }
            Err(err) => {
                if now < expires_at {
                    tracing::error!(
                        %err,
                        expires_at = expires_at.to_string(),
                        "could not refresh the route53 credential; signing with the one held \
                         until it expires"
                    );
                    Ok(held.credentials.clone())
                } else {
                    tracing::error!(
                        %err,
                        expires_at = expires_at.to_string(),
                        "could not refresh the route53 credential and the one held has \
                         expired; route53 calls will fail until instance metadata answers"
                    );
                    Err(err)
                }
            }
        }
    }
}

/// The one lever [`crate::aws_fake`] has on a credential holder that is
/// otherwise driven entirely by the wall clock.
///
/// Nothing outside this crate's own tests can reach it: it is `pub(crate)`
/// and behind the `aws-fake` feature, which no production build turns on.
#[cfg(feature = "aws-fake")]
impl InstanceCredentials {
    /// Makes the credential in hand expired as of now, so the next
    /// [`Self::current`] takes the refresh path rather than answering from
    /// what it holds.
    pub(crate) fn expire_held(&self) {
        let mut held = self
            .held
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        held.expires_at = Some(time::OffsetDateTime::now_utc() - time::Duration::seconds(1));
    }
}

const IMDS_ENDPOINT: &str = "http://169.254.169.254";

/// How long one IMDS round trip may take. Shorter than didbot-http's shared
/// default: 169.254.169.254 is a link-local address on the same host, so a
/// slow answer means IMDS is not there at all rather than a congested route.
const IMDS_TIMEOUT: Duration = Duration::from_secs(5);

/// The client IMDS is reached through.
fn imds_client() -> reqwest::Result<reqwest::blocking::Client> {
    didbot_http::blocking_builder()
        .timeout(IMDS_TIMEOUT)
        .build()
}

/// The client the Route53 API is reached through. Without a timeout, a
/// stalled call would hang whichever DNS operation asked for it, which is
/// why the fallback to `reqwest`'s own defaults says so at `warn` rather
/// than leaving that hang with nothing to explain it.
fn api_client() -> reqwest::blocking::Client {
    didbot_http::blocking_builder()
        .build()
        .unwrap_or_else(|error| {
            tracing::warn!(
                %error,
                "building the Route53 API client failed; falling back to reqwest's own \
                 defaults, which set no deadline on a Route53 call"
            );
            reqwest::blocking::Client::default()
        })
}

fn imds_request(request: reqwest::blocking::RequestBuilder) -> Result<String, ImdsError> {
    let response = request
        .send()
        .map_err(|err| ImdsError::Request(err.to_string()))?
        .error_for_status()
        .map_err(|err| ImdsError::Request(err.to_string()))?;
    let body = didbot_http::read_bounded_blocking(response, didbot_http::MAX_JSON_BODY)
        .map_err(|err| ImdsError::Request(err.to_string()))?;
    Ok(didbot_http::body_text(body))
}

/// What
/// `GET /latest/meta-data/iam/security-credentials/<role>` answers with.
/// `Code`, `LastUpdated` and `Type` are in the response and are not read.
#[derive(serde::Deserialize)]
struct ImdsCredentials {
    #[serde(rename = "AccessKeyId")]
    access_key_id: String,
    #[serde(rename = "SecretAccessKey")]
    secret_access_key: String,
    #[serde(rename = "Token")]
    token: Option<String>,
    #[serde(rename = "Expiration")]
    expiration: Option<String>,
}

/// Reads the credentials out of a
/// `GET /latest/meta-data/iam/security-credentials/<role>` response.
fn parse_imds_credentials(body: &str) -> Result<TimedCredentials, ImdsError> {
    let parsed: ImdsCredentials = serde_json::from_str(body).map_err(|err| {
        ImdsError::Response(format!("the credentials response does not read: {err}"))
    })?;
    let expires_at = match &parsed.expiration {
        Some(stated) => Some(
            time::OffsetDateTime::parse(stated, &time::format_description::well_known::Rfc3339)
                .map_err(|err| {
                    ImdsError::Response(format!("Expiration {stated:?} is not a timestamp: {err}"))
                })?,
        ),
        None => None,
    };
    Ok(TimedCredentials {
        credentials: Route53Credentials {
            access_key_id: parsed.access_key_id,
            secret_access_key: parsed.secret_access_key,
            session_token: parsed.token,
        },
        expires_at,
    })
}

/// How this provider retries a throttled or transiently failed call.
#[derive(Debug, Clone, Copy)]
pub struct RetryPolicy {
    /// Total attempts, including the first. `1` disables retrying.
    pub max_attempts: u32,
    /// Delay before the first retry; doubles on every subsequent one.
    pub base_delay: Duration,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 5,
            base_delay: Duration::from_millis(200),
        }
    }
}

/// The production [`DnsProvider`]: writes and removes the `TXT` and `CAA`
/// records in one Route53 hosted zone. See the module docs for the operating
/// rules — propagation, retries, quotas and the credential's scope.
pub struct Route53Dns {
    zone: String,
    hosted_zone_id: String,
    transport: Box<dyn Transport>,
    retry: RetryPolicy,
    wait_for_propagation: bool,
    propagation_timeout: Duration,
    poll_interval: Duration,
    /// The TTL on the `CAA` record this provider writes, matching the
    /// address records beside it at the apex.
    ttl: u32,
    /// The TTL on the `TXT` records this provider writes. Short, because an
    /// ACME DNS-01 challenge value is only needed for the length of one
    /// validation and a short TTL keeps a withdrawn or replaced value from
    /// lingering in downstream caches.
    txt_ttl: u32,
    txt: TxtRecords,
    caa: CaaRecords,
}

impl Route53Dns {
    /// A provider for `zone`, writing to hosted zone `hosted_zone_id`, over
    /// the real network.
    ///
    /// Starts with empty local bookkeeping — the same starting state every
    /// provider in this crate has. Every write this provider makes is an
    /// `UPSERT` of the whole set it holds for a name, so a value the zone
    /// carries from a previous process is replaced by the first publish at
    /// that name rather than added to.
    pub fn new(
        zone: impl Into<String>,
        hosted_zone_id: impl Into<String>,
        credentials: std::sync::Arc<InstanceCredentials>,
    ) -> Self {
        Self::with_transport(
            zone,
            hosted_zone_id,
            Box::new(HttpTransport {
                client: api_client(),
                endpoint: DEFAULT_ENDPOINT.to_string(),
                credentials,
            }),
        )
    }

    pub(crate) fn with_transport(
        zone: impl Into<String>,
        hosted_zone_id: impl Into<String>,
        transport: Box<dyn Transport>,
    ) -> Self {
        Self {
            zone: zone.into(),
            hosted_zone_id: hosted_zone_id.into(),
            transport,
            retry: RetryPolicy::default(),
            wait_for_propagation: true,
            propagation_timeout: Duration::from_secs(30),
            poll_interval: Duration::from_millis(500),
            ttl: 300,
            txt_ttl: 60,
            txt: TxtRecords::default(),
            caa: CaaRecords::default(),
        }
    }

    /// Overrides the retry policy for throttled or transient failures.
    #[must_use]
    pub fn with_retry_policy(mut self, retry: RetryPolicy) -> Self {
        self.retry = retry;
        self
    }

    /// Whether [`DnsProvider::publish_txt`] waits for the change to reach
    /// `INSYNC` before returning. Default `true`; see the module docs for
    /// why that is the honest default. Turning it off trades that guarantee
    /// for lower publish latency.
    #[must_use]
    pub fn with_propagation_wait(mut self, wait: bool) -> Self {
        self.wait_for_propagation = wait;
        self
    }

    /// How long [`DnsProvider::publish_txt`] will wait for `INSYNC` before
    /// giving up and returning [`DnsError::Backend`]. Default 30 seconds.
    #[must_use]
    pub fn with_propagation_timeout(mut self, timeout: Duration) -> Self {
        self.propagation_timeout = timeout;
        self
    }

    /// How long [`DnsProvider::publish_txt`] waits between `GetChange` polls
    /// while a change is `PENDING`. Default 500 milliseconds.
    ///
    /// A Route53 change reaches `INSYNC` in tens of seconds, so the default
    /// is sized for that rather than for how fast a caller can learn the
    /// answer. Lowering it costs one API call per poll.
    #[must_use]
    pub fn with_poll_interval(mut self, interval: Duration) -> Self {
        self.poll_interval = interval;
        self
    }

    /// Whether `host` is under this provider's zone.
    ///
    /// A deployment answers this zone with a single-label wildcard DNS
    /// record (`*.<zone>`), and RFC 1034 wildcards match exactly one label.
    /// So `host` must be the zone itself (the apex, for `CAA`) or exactly
    /// one label below it (`_acme-challenge.<zone>`) — never two or more.
    pub fn accepts(&self, host: &str) -> bool {
        if !hostname_is_at_or_below(host, &self.zone) {
            return false;
        }
        let zone_labels = self.zone.matches('.').count() + 1;
        let host_labels = host.matches('.').count() + 1;
        host_labels <= zone_labels + 1
    }

    fn check(&self, host: &str) -> Result<(), DnsError> {
        if self.accepts(host) {
            Ok(())
        } else {
            Err(DnsError::NotUnderZone {
                host: host.to_owned(),
            })
        }
    }

    fn backend_error(&self, host: &str, message: String) -> DnsError {
        DnsError::Backend {
            host: host.to_string(),
            message,
        }
    }

    /// Issues one API call, retrying on throttling or a server error per
    /// [`Self::retry`].
    fn call_with_retry(
        &self,
        method: &str,
        path: &str,
        query: &[(String, String)],
        body: String,
        host_for_errors: &str,
    ) -> Result<String, DnsError> {
        let mut attempt = 0;
        let mut delay = self.retry.base_delay;
        loop {
            attempt += 1;
            match self.transport.call(method, path, query, &body) {
                Ok(response) if response.status < 300 => return Ok(response.body),
                Ok(response) => {
                    let error = xml::parse_error(&response.body);
                    let retryable = response.status == 500
                        || response.status == 503
                        || error.as_ref().is_some_and(|e| {
                            e.code == "Throttling" || e.code == "PriorRequestNotComplete"
                        });
                    if retryable && attempt < self.retry.max_attempts {
                        thread::sleep(delay);
                        delay *= 2;
                        continue;
                    }
                    let message = error
                        .map(|e| format!("{}: {}", e.code, e.message))
                        .unwrap_or_else(|| format!("http {}", response.status));
                    return Err(self.classify(host_for_errors, response.status, &message));
                }
                Err(message) => {
                    if attempt < self.retry.max_attempts {
                        thread::sleep(delay);
                        delay *= 2;
                        continue;
                    }
                    return Err(self.backend_error(host_for_errors, message));
                }
            }
        }
    }

    /// Maps a non-retried API error to the right [`DnsError`] variant.
    fn classify(&self, host: &str, status: u16, message: &str) -> DnsError {
        let lower = message.to_ascii_lowercase();
        if status == 400 && (lower.contains("not found") || lower.contains("not match")) {
            DnsError::NotPublished {
                host: host.to_string(),
            }
        } else {
            self.backend_error(host, message.to_string())
        }
    }

    /// Polls `GetChange` until `change_id` reaches `INSYNC`, or gives up
    /// after [`Self::propagation_timeout`].
    fn await_propagation(&self, change_id: &str) -> Result<(), DnsError> {
        let deadline = Instant::now() + self.propagation_timeout;
        loop {
            let path = format!("/{API_VERSION}/change/{change_id}");
            let body = self.call_with_retry("GET", &path, &[], String::new(), change_id)?;
            match xml::parse_status(&body) {
                Some(status) if status == "INSYNC" => return Ok(()),
                _ if Instant::now() >= deadline => {
                    return Err(self.backend_error(
                        change_id,
                        format!(
                            "change did not reach INSYNC within {:?}",
                            self.propagation_timeout
                        ),
                    ))
                }
                _ => thread::sleep(self.poll_interval),
            }
        }
    }

    fn submit_change(
        &self,
        host: &str,
        action: &str,
        rtype: &str,
        ttl: u32,
        values: &[String],
    ) -> Result<String, DnsError> {
        let body = xml::change_request("didbot", action, host, rtype, ttl, values)
            .map_err(|err| self.backend_error(host, format!("cannot build the change: {err}")))?;
        let path = format!("/{API_VERSION}/hostedzone/{}/rrset", self.hosted_zone_id);
        let response = self.call_with_retry("POST", &path, &[], body, host)?;
        xml::parse_change_id(&response).ok_or_else(|| {
            self.backend_error(
                host,
                "change accepted but no change id in response".to_string(),
            )
        })
    }
}

impl std::fmt::Debug for Route53Dns {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Route53Dns")
            .field("zone", &self.zone)
            .field("hosted_zone_id", &self.hosted_zone_id)
            .finish_non_exhaustive()
    }
}

impl DnsProvider for Route53Dns {
    fn name(&self) -> &str {
        "route53"
    }

    fn publish_txt(&self, host: &str, value: &str) -> Result<(), DnsError> {
        self.check(host)?;
        let before = self.txt.values(host);
        if before.contains(&value.to_string()) {
            return Ok(());
        }
        self.txt.insert(host, value);
        let after: BTreeSet<String> = self.txt.values(host).into_iter().collect();
        let values: Vec<String> = after.iter().map(|v| xml::quote_txt(v)).collect();

        // The caller's next step is to tell a certificate authority to look
        // for this value, so the change has to be visible before this
        // returns; see the module docs.
        let outcome = self
            .submit_change(host, "UPSERT", "TXT", self.txt_ttl, &values)
            .and_then(|change_id| {
                if self.wait_for_propagation {
                    self.await_propagation(&change_id)
                } else {
                    Ok(())
                }
            });
        if let Err(error) = outcome {
            self.txt.remove(host, value).ok();
            return Err(error);
        }
        tracing::info!(
            provider = self.name(),
            host = host,
            value = value,
            "published txt value"
        );
        Ok(())
    }

    fn withdraw_txt(&self, host: &str, value: &str) -> Result<(), DnsError> {
        self.check(host)?;
        let before: Vec<String> = self.txt.values(host);
        if !before.iter().any(|v| v == value) {
            return Err(DnsError::NotPublished {
                host: host.to_string(),
            });
        }
        self.txt.remove(host, value)?;
        let after = self.txt.values(host);

        let outcome = if after.is_empty() {
            // Nothing left: the whole record set must be deleted, and a
            // DELETE must name the exact values Route53 currently has,
            // which is `before`.
            let quoted: Vec<String> = before.iter().map(|v| xml::quote_txt(v)).collect();
            self.submit_change(host, "DELETE", "TXT", self.txt_ttl, &quoted)
        } else {
            let quoted: Vec<String> = after.iter().map(|v| xml::quote_txt(v)).collect();
            self.submit_change(host, "UPSERT", "TXT", self.txt_ttl, &quoted)
        };

        if let Err(error) = outcome {
            self.txt.insert(host, value);
            return Err(error);
        }
        tracing::info!(
            provider = self.name(),
            host = host,
            value = value,
            "withdrew txt value"
        );
        Ok(())
    }

    fn txt_values(&self, host: &str) -> Vec<String> {
        self.txt.values(host)
    }

    fn publish_caa(&self, host: &str, values: &[CaaValue]) -> Result<(), DnsError> {
        self.check(host)?;
        let before = self.caa.values(host);
        if !self.caa.replace(host, values) {
            // Byte-for-byte what this zone already says. A restart restating
            // its own policy is the common case and must not spend a write
            // against Route53's change quota.
            return Ok(());
        }
        let after = self.caa.values(host);

        let outcome = if after.is_empty() {
            // A DELETE must name the exact values Route53 currently holds,
            // which is `before` — the same rule `withdraw_txt` follows.
            let wire: Vec<String> = before.iter().map(CaaValue::to_wire).collect();
            self.submit_change(host, "DELETE", "CAA", self.ttl, &wire)
        } else {
            let wire: Vec<String> = after.iter().map(CaaValue::to_wire).collect();
            self.submit_change(host, "UPSERT", "CAA", self.ttl, &wire)
        };

        if let Err(error) = outcome {
            self.caa.replace(host, &before);
            return Err(error);
        }
        crate::log_caa(self.name(), host, &after, "published caa record set");
        Ok(())
    }

    fn caa_values(&self, host: &str) -> Vec<CaaValue> {
        self.caa.values(host)
    }
}

/// One HTTP response, isolated from the `reqwest` type so tests can build
/// one without a client.
pub(crate) struct Response {
    pub(crate) status: u16,
    pub(crate) body: String,
}

/// The network boundary this provider drives, isolated so tests can replace
/// it with something that records calls and returns canned XML instead of
/// reaching the real Route53 API — which there is no AWS account to test
/// against in this environment. `HttpTransport` is the only real
/// implementation. The tests in `sigv4` check each step of the signature it
/// sends against AWS's published signing test suite.
pub(crate) trait Transport: Send + Sync {
    fn call(
        &self,
        method: &str,
        path: &str,
        query: &[(String, String)],
        body: &str,
    ) -> Result<Response, String>;
}

struct HttpTransport {
    client: reqwest::blocking::Client,
    endpoint: String,
    credentials: std::sync::Arc<InstanceCredentials>,
}

impl Transport for HttpTransport {
    fn call(
        &self,
        method: &str,
        path: &str,
        query: &[(String, String)],
        body: &str,
    ) -> Result<Response, String> {
        // Per call, not per provider: an instance profile's credential
        // expires, and this is where the holder notices.
        let credentials = self.credentials.current().map_err(|err| err.to_string())?;
        let now = time::OffsetDateTime::now_utc();
        // One value for both the signed `host` header and the sent one: a
        // signature over a host the request does not carry cannot verify.
        let host = sigv4::host_header(&self.endpoint);
        let request = sigv4::Request {
            method,
            host: &host,
            path,
            query,
            body,
        };
        let signed = sigv4::sign(&credentials, sigv4::ROUTE53, &request, now);

        let mut url = format!("{}{}", self.endpoint, path);
        if !query.is_empty() {
            url.push('?');
            url.push_str(&sigv4::canonical_query(query));
        }

        let method = match method {
            "GET" => reqwest::Method::GET,
            "POST" => reqwest::Method::POST,
            other => return Err(format!("unsupported method {other}")),
        };

        let mut request = self
            .client
            .request(method, &url)
            .header("host", host)
            .header("x-amz-date", signed.amz_date.clone())
            .header("authorization", signed.authorization)
            .header("content-type", "text/xml; charset=utf-8");
        if let Some(token) = &credentials.session_token {
            request = request.header("x-amz-security-token", token.clone());
        }
        if !body.is_empty() {
            request = request.body(body.to_string());
        }

        let response = request.send().map_err(|error| error.to_string())?;
        let status = response.status().as_u16();
        let body = didbot_http::read_bounded_blocking(response, didbot_http::MAX_JSON_BODY)
            .map_err(|error| error.to_string())?;
        Ok(Response {
            status,
            body: didbot_http::body_text(body),
        })
    }
}

mod sigv4;

/// The handful of request bodies and response shapes this provider needs.
///
/// Route53's XML surface is large; this reads and writes only the elements
/// listed below; a response element this module does not recognise is
/// ignored rather than rejected, which is deliberate — Route53 is free to
/// add fields to a response without breaking a client that never looked for
/// them. This is a purpose-built extractor for those specific shapes, not a
/// conformant general XML parser: it does not handle attributes, namespaces,
/// CDATA or non-well-formed input, none of which Route53's own responses use
/// in the elements this module reads.
mod xml {
    use serde::{Deserialize, Serialize};

    /// One `ChangeResourceRecordSets` request body.
    #[derive(Serialize)]
    #[serde(rename = "ChangeResourceRecordSetsRequest")]
    struct ChangeRequest {
        #[serde(rename = "@xmlns")]
        xmlns: String,
        #[serde(rename = "ChangeBatch")]
        batch: ChangeBatch,
    }

    #[derive(Serialize)]
    struct ChangeBatch {
        #[serde(rename = "Comment")]
        comment: String,
        #[serde(rename = "Changes")]
        changes: Changes,
    }

    #[derive(Serialize)]
    struct Changes {
        #[serde(rename = "Change")]
        change: Vec<Change>,
    }

    #[derive(Serialize)]
    struct Change {
        #[serde(rename = "Action")]
        action: String,
        #[serde(rename = "ResourceRecordSet")]
        record_set: ResourceRecordSet,
    }

    #[derive(Serialize)]
    struct ResourceRecordSet {
        #[serde(rename = "Name")]
        name: String,
        #[serde(rename = "Type")]
        rtype: String,
        #[serde(rename = "TTL")]
        ttl: u32,
        #[serde(rename = "ResourceRecords")]
        records: ResourceRecords,
    }

    #[derive(Serialize)]
    struct ResourceRecords {
        #[serde(rename = "ResourceRecord")]
        record: Vec<ResourceRecord>,
    }

    #[derive(Serialize)]
    struct ResourceRecord {
        #[serde(rename = "Value")]
        value: String,
    }

    /// The body of a one-change `ChangeResourceRecordSets` call.
    pub(super) fn change_request(
        comment: &str,
        action: &str,
        host: &str,
        rtype: &str,
        ttl: u32,
        values: &[String],
    ) -> Result<String, quick_xml::SeError> {
        let request = ChangeRequest {
            xmlns: format!("https://route53.amazonaws.com/doc/{}/", super::API_VERSION),
            batch: ChangeBatch {
                comment: comment.to_string(),
                changes: Changes {
                    change: vec![Change {
                        action: action.to_string(),
                        record_set: ResourceRecordSet {
                            name: host.to_string(),
                            rtype: rtype.to_string(),
                            ttl,
                            records: ResourceRecords {
                                record: values
                                    .iter()
                                    .map(|value| ResourceRecord {
                                        value: value.clone(),
                                    })
                                    .collect(),
                            },
                        },
                    }],
                },
            },
        };
        quick_xml::se::to_string(&request)
    }

    /// Wraps a TXT value in the quoted, backslash-escaped form the DNS
    /// presentation format requires for a `TXT` `ResourceRecord` value.
    pub(super) fn quote_txt(value: &str) -> String {
        let mut out = String::with_capacity(value.len() + 2);
        out.push('"');
        for c in value.chars() {
            if c == '"' || c == '\\' {
                out.push('\\');
            }
            out.push(c);
        }
        out.push('"');
        out
    }

    /// What `ChangeResourceRecordSets` and `GetChange` both answer with.
    #[derive(Deserialize)]
    struct ChangeInfoResponse {
        #[serde(rename = "ChangeInfo")]
        change_info: ChangeInfo,
    }

    #[derive(Deserialize)]
    struct ChangeInfo {
        #[serde(rename = "Id")]
        id: String,
        #[serde(rename = "Status")]
        status: String,
    }

    #[derive(Deserialize)]
    struct ErrorResponse {
        #[serde(rename = "Error")]
        error: ApiError,
    }

    /// What an API error says about itself. `Message` is optional because a
    /// few of Route53's errors carry only a code.
    #[derive(Deserialize)]
    pub(super) struct ApiError {
        #[serde(rename = "Code")]
        pub(super) code: String,
        #[serde(rename = "Message", default)]
        pub(super) message: String,
    }

    pub(super) fn parse_change_id(xml: &str) -> Option<String> {
        let response: ChangeInfoResponse = quick_xml::de::from_str(xml).ok()?;
        Some(
            response
                .change_info
                .id
                .trim_start_matches("/change/")
                .to_string(),
        )
    }

    pub(super) fn parse_status(xml: &str) -> Option<String> {
        let response: ChangeInfoResponse = quick_xml::de::from_str(xml).ok()?;
        Some(response.change_info.status)
    }

    pub(super) fn parse_error(xml: &str) -> Option<ApiError> {
        let response: ErrorResponse = quick_xml::de::from_str(xml).ok()?;
        Some(response.error)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The host [`DEFAULT_ENDPOINT`] sends, and so the host a signature
    /// against it covers.
    const DEFAULT_HOST: &str = "route53.amazonaws.com";
    use std::sync::Mutex;

    /// Records every call it receives and answers from a scripted list of
    /// responses, so the provider's request-building, retry and parsing
    /// logic is tested without a network or an AWS account — which this
    /// environment has neither of.
    struct FakeTransport {
        responses: Mutex<Vec<Result<Response, String>>>,
        calls: Mutex<Vec<(String, String, String)>>, // method, path, body
    }

    impl FakeTransport {
        fn new(responses: Vec<Result<Response, String>>) -> Self {
            Self {
                responses: Mutex::new(responses),
                calls: Mutex::new(Vec::new()),
            }
        }

        fn calls(&self) -> Vec<(String, String, String)> {
            self.calls.lock().unwrap().clone()
        }
    }

    impl Transport for FakeTransport {
        fn call(
            &self,
            method: &str,
            path: &str,
            _query: &[(String, String)],
            body: &str,
        ) -> Result<Response, String> {
            self.calls.lock().unwrap().push((
                method.to_string(),
                path.to_string(),
                body.to_string(),
            ));
            let mut responses = self.responses.lock().unwrap();
            if responses.is_empty() {
                // GetChange polling loop after the first canned response ran
                // out: keep reporting INSYNC so a test that does not script
                // every poll does not hang.
                return Ok(Response {
                    status: 200,
                    body: "<GetChangeResponse><ChangeInfo><Id>/change/C1</Id><Status>INSYNC</Status></ChangeInfo></GetChangeResponse>".to_string(),
                });
            }
            responses.remove(0)
        }
    }

    // Lets a test keep a handle to the fake (to inspect `calls()`) while
    // still handing `Route53Dns` an owned `Box<dyn Transport>`.
    impl Transport for std::sync::Arc<FakeTransport> {
        fn call(
            &self,
            method: &str,
            path: &str,
            query: &[(String, String)],
            body: &str,
        ) -> Result<Response, String> {
            (**self).call(method, path, query, body)
        }
    }

    fn ok(body: &str) -> Result<Response, String> {
        Ok(Response {
            status: 200,
            body: body.to_string(),
        })
    }

    fn change_response() -> Result<Response, String> {
        ok("<ChangeResourceRecordSetsResponse><ChangeInfo><Id>/change/C1</Id><Status>PENDING</Status></ChangeInfo></ChangeResourceRecordSetsResponse>")
    }

    fn insync_response() -> Result<Response, String> {
        ok("<GetChangeResponse><ChangeInfo><Id>/change/C1</Id><Status>INSYNC</Status></ChangeInfo></GetChangeResponse>")
    }

    fn provider(responses: Vec<Result<Response, String>>) -> Route53Dns {
        let transport = Box::new(FakeTransport::new(responses));
        Route53Dns::with_transport("agents.example.com", "Z123", transport)
            .with_propagation_timeout(Duration::from_millis(200))
    }

    #[test]
    fn a_txt_publish_waits_for_insync_before_returning() {
        let fake = std::sync::Arc::new(FakeTransport::new(vec![
            change_response(),
            insync_response(),
        ]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", Box::new(fake.clone()))
            .with_propagation_timeout(Duration::from_millis(200));
        assert!(dns
            .publish_txt("_acme-challenge.agents.example.com", "token")
            .is_ok());
        // The UPSERT, then the GetChange that answered INSYNC.
        let calls = fake.calls();
        assert_eq!(calls.len(), 2, "{calls:?}");
        assert!(calls[1].1.contains("/change/"), "{calls:?}");
    }

    #[test]
    fn a_txt_withdrawal_does_not_wait_for_propagation() {
        // Exactly two scripted responses, the UPSERT's and the DELETE's: if
        // the withdrawal also polled GetChange the way a publish does, the
        // fake would run out of responses and fall back to a canned INSYNC
        // body this test never asserts on -- so assert on the call count.
        let fake = std::sync::Arc::new(FakeTransport::new(vec![
            change_response(),
            change_response(),
        ]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", Box::new(fake.clone()))
            .with_propagation_wait(false);
        let name = "_acme-challenge.agents.example.com";
        assert!(dns.publish_txt(name, "token").is_ok());
        assert!(dns.withdraw_txt(name, "token").is_ok());
        assert_eq!(fake.calls().len(), 2);
    }

    #[test]
    fn a_txt_publish_refuses_a_host_outside_the_zone() {
        // No transport call is queued, so this would also panic on an
        // unexpected HTTP call if `accepts` failed to refuse before the
        // network was reached.
        let dns = provider(vec![]);
        assert_eq!(
            dns.publish_txt("_acme-challenge.other.com", "token"),
            Err(DnsError::NotUnderZone {
                host: "_acme-challenge.other.com".to_string()
            })
        );
        assert_eq!(
            dns.publish_txt("_acme-challenge.deep.agents.example.com", "token"),
            Err(DnsError::NotUnderZone {
                host: "_acme-challenge.deep.agents.example.com".to_string()
            })
        );
    }

    #[test]
    fn caa_is_upserted_as_one_record_set_at_the_apex() {
        let fake = std::sync::Arc::new(FakeTransport::new(vec![change_response()]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", Box::new(fake.clone()));
        assert!(dns
            .publish_caa(
                "agents.example.com",
                &[
                    CaaValue::issue_for_account("letsencrypt.org", "https://acme.example/acct/1"),
                    CaaValue::iodef("mailto:ops@example.com"),
                ],
            )
            .is_ok());

        let (_, _, body) = fake.calls()[0].clone();
        assert!(body.contains("<Action>UPSERT</Action>"), "{body}");
        assert!(body.contains("<Type>CAA</Type>"), "{body}");
        assert!(
            body.contains("0 issue \"letsencrypt.org; accounturi=https://acme.example/acct/1\""),
            "{body}"
        );
        assert!(
            body.contains("0 iodef \"mailto:ops@example.com\""),
            "{body}"
        );
    }

    #[test]
    fn restating_the_same_caa_set_spends_no_change() {
        let fake = std::sync::Arc::new(FakeTransport::new(vec![change_response()]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", Box::new(fake.clone()));
        let values = [CaaValue::issue("letsencrypt.org")];
        assert!(dns.publish_caa("agents.example.com", &values).is_ok());
        // Every start restates the policy; only the first one is a write.
        assert!(dns.publish_caa("agents.example.com", &values).is_ok());
        assert_eq!(fake.calls().len(), 1);
    }

    #[test]
    fn txt_publish_is_additive_and_uses_upsert() {
        let fake = std::sync::Arc::new(FakeTransport::new(vec![
            change_response(),
            insync_response(),
            change_response(),
            insync_response(),
        ]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", Box::new(fake.clone()))
            .with_propagation_timeout(Duration::from_millis(200));
        let name = "_acme-challenge.agents.example.com";
        assert!(dns.publish_txt(name, "token-a").is_ok());
        assert!(dns.publish_txt(name, "token-b").is_ok());
        for (_, _, body) in fake
            .calls()
            .iter()
            .filter(|(_, path, _)| path.ends_with("/rrset"))
        {
            assert!(body.contains("<Action>UPSERT</Action>"), "{body}");
        }
        // The second write carries both values, not just the new one.
        let (_, _, body) = fake.calls()[2].clone();
        assert!(
            body.contains("token-a") && body.contains("token-b"),
            "{body}"
        );
        assert_eq!(dns.txt_values(name), vec!["token-a", "token-b"]);
    }

    #[test]
    fn txt_publish_same_value_twice_makes_no_second_call() {
        let dns = provider(vec![change_response()]);
        let name = "_acme-challenge.agents.example.com";
        assert!(dns.publish_txt(name, "token-a").is_ok());
        assert!(dns.publish_txt(name, "token-a").is_ok());
        assert_eq!(dns.txt_values(name), vec!["token-a"]);
    }

    #[test]
    fn withdrawing_last_txt_value_issues_a_delete() {
        let transport = Box::new(FakeTransport::new(vec![
            change_response(),
            change_response(),
        ]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", transport);
        let name = "_acme-challenge.agents.example.com";
        assert!(dns.publish_txt(name, "only").is_ok());
        assert!(dns.withdraw_txt(name, "only").is_ok());
        assert!(dns.txt_values(name).is_empty());
    }

    #[test]
    fn retries_on_throttling_then_succeeds() {
        let throttled = Ok(Response {
            status: 400,
            body: "<ErrorResponse><Error><Code>Throttling</Code><Message>Rate exceeded</Message></Error></ErrorResponse>".to_string(),
        });
        let transport = Box::new(FakeTransport::new(vec![
            throttled,
            change_response(),
            insync_response(),
        ]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", transport)
            .with_retry_policy(RetryPolicy {
                max_attempts: 3,
                base_delay: Duration::from_millis(0),
            });
        assert!(dns
            .publish_txt("_acme-challenge.agents.example.com", "token")
            .is_ok());
    }

    #[test]
    fn exhausting_retries_on_throttling_is_a_backend_error() {
        let throttled = || {
            Ok(Response {
                status: 400,
                body: "<ErrorResponse><Error><Code>Throttling</Code><Message>Rate exceeded</Message></Error></ErrorResponse>".to_string(),
            })
        };
        let transport = Box::new(FakeTransport::new(vec![
            throttled(),
            throttled(),
            throttled(),
        ]));
        let dns = Route53Dns::with_transport("agents.example.com", "Z123", transport)
            .with_retry_policy(RetryPolicy {
                max_attempts: 3,
                base_delay: Duration::from_millis(0),
            });
        assert!(matches!(
            dns.publish_txt("_acme-challenge.agents.example.com", "token"),
            Err(DnsError::Backend { .. })
        ));
        // The failed publish must not linger as a value this process
        // believes is live.
        assert!(dns
            .txt_values("_acme-challenge.agents.example.com")
            .is_empty());
    }

    #[test]
    fn quote_txt_escapes_quotes_and_backslashes() {
        assert_eq!(xml::quote_txt("plain"), "\"plain\"");
        assert_eq!(xml::quote_txt("a\"b"), "\"a\\\"b\"");
        assert_eq!(xml::quote_txt("a\\b"), "\"a\\\\b\"");
    }

    #[test]
    fn parses_change_id_and_status() {
        let body = change_response().unwrap().body;
        assert_eq!(xml::parse_change_id(&body).as_deref(), Some("C1"));
        let body = insync_response().unwrap().body;
        assert_eq!(xml::parse_status(&body).as_deref(), Some("INSYNC"));
    }

    /// The `Authorization` header of a Route53 `GET` of `path` to `host`.
    fn route53_get(credentials: &Route53Credentials, host: &str, path: &str) -> String {
        let request = sigv4::Request {
            method: "GET",
            host,
            path,
            query: &[],
            body: "",
        };
        let now = time::macros::datetime!(2015-08-30 12:36:00 UTC);
        sigv4::sign(credentials, sigv4::ROUTE53, &request, now).authorization
    }

    #[test]
    fn a_custom_endpoint_signs_the_host_it_sends() {
        let credentials = Route53Credentials {
            access_key_id: "AKIDEXAMPLE".to_string(),
            secret_access_key: "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".to_string(),
            session_token: None,
        };

        // The endpoint a test double or a private link would be pointed at.
        let endpoint = "https://route53.us-gov.amazonaws.com/";
        let host = sigv4::host_header(endpoint);
        assert_eq!(host, "route53.us-gov.amazonaws.com");

        let elsewhere = route53_get(&credentials, &host, "/");
        let default = route53_get(&credentials, DEFAULT_HOST, "/");

        // The host is signed, so two endpoints cannot share one signature.
        // Equal signatures here would mean the host was not covered at all.
        assert_ne!(elsewhere, default);
        assert_eq!(sigv4::host_header(DEFAULT_ENDPOINT), DEFAULT_HOST);
    }

    #[test]
    fn parses_imds_security_credentials() {
        // A representative
        // `GET .../iam/security-credentials/<role>` response body; field
        // order and the extra fields this module does not read
        // (`Code`, `LastUpdated`, `Type`) match a real one.
        let body = r#"{
            "Code" : "Success",
            "LastUpdated" : "2024-01-01T00:00:00Z",
            "Type" : "AWS-HMAC",
            "AccessKeyId" : "ASIAEXAMPLE",
            "SecretAccessKey" : "secretexamplevalue",
            "Token" : "tokenexamplevalue",
            "Expiration" : "2024-01-01T06:00:00Z"
        }"#;
        let timed = parse_imds_credentials(body).expect("parses");
        assert_eq!(timed.credentials.access_key_id, "ASIAEXAMPLE");
        assert_eq!(timed.credentials.secret_access_key, "secretexamplevalue");
        assert_eq!(
            timed.credentials.session_token.as_deref(),
            Some("tokenexamplevalue")
        );
        assert_eq!(
            timed.expires_at,
            Some(time::macros::datetime!(2024-01-01 06:00:00 UTC))
        );
    }

    #[test]
    fn imds_credentials_with_an_unreadable_expiration_are_a_response_error() {
        // Signing with a credential whose end this cannot read would mean
        // treating it as endless, which is the failure this holder exists to
        // stop.
        let body = r#"{"AccessKeyId":"A","SecretAccessKey":"s","Expiration":"six hours"}"#;
        assert!(matches!(
            parse_imds_credentials(body),
            Err(ImdsError::Response(_))
        ));
    }

    #[test]
    fn imds_credentials_without_a_token_field_still_parse() {
        let body = r#"{"AccessKeyId":"AKIAEXAMPLE","SecretAccessKey":"secret"}"#;
        let timed = parse_imds_credentials(body).expect("parses");
        assert_eq!(timed.credentials.access_key_id, "AKIAEXAMPLE");
        assert_eq!(timed.credentials.session_token, None);
        assert_eq!(timed.expires_at, None);
    }

    #[test]
    fn imds_credentials_missing_a_required_field_is_a_response_error() {
        let body = r#"{"Code":"Success"}"#;
        assert!(matches!(
            parse_imds_credentials(body),
            Err(ImdsError::Response(_))
        ));
    }

    /// Answers [`InstanceCredentials`] from a scripted list and counts what
    /// it was asked, so a test can drive a credential through its expiry
    /// without an instance metadata service — which this environment, like
    /// the AWS account behind `Transport`, does not have. The last answer
    /// repeats once the list runs out.
    struct FakeImds {
        answers: Mutex<Vec<Result<TimedCredentials, String>>>,
        reads: std::sync::atomic::AtomicUsize,
        /// Long enough for other threads to reach `current()` while a
        /// refresh is in flight.
        delay: Duration,
    }

    impl FakeImds {
        fn new(answers: Vec<Result<TimedCredentials, String>>) -> std::sync::Arc<Self> {
            std::sync::Arc::new(Self {
                answers: Mutex::new(answers),
                reads: std::sync::atomic::AtomicUsize::new(0),
                delay: Duration::ZERO,
            })
        }

        fn with_delay(
            answers: Vec<Result<TimedCredentials, String>>,
            delay: Duration,
        ) -> std::sync::Arc<Self> {
            std::sync::Arc::new(Self {
                answers: Mutex::new(answers),
                reads: std::sync::atomic::AtomicUsize::new(0),
                delay,
            })
        }

        fn reads(&self) -> usize {
            self.reads.load(std::sync::atomic::Ordering::SeqCst)
        }
    }

    impl CredentialFetch for FakeImds {
        fn fetch(&self) -> Result<TimedCredentials, ImdsError> {
            let read = self.reads.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            if !self.delay.is_zero() {
                thread::sleep(self.delay);
            }
            let answers = self.answers.lock().unwrap();
            let answer = answers[read.min(answers.len() - 1)].clone();
            answer.map_err(ImdsError::Request)
        }
    }

    impl CredentialFetch for std::sync::Arc<FakeImds> {
        fn fetch(&self) -> Result<TimedCredentials, ImdsError> {
            (**self).fetch()
        }
    }

    /// A credential whose secret is `secret`, valid for `valid_for` from
    /// now; a negative `valid_for` is one that has already expired.
    fn expiring(secret: &str, valid_for: time::Duration) -> Result<TimedCredentials, String> {
        Ok(TimedCredentials {
            credentials: Route53Credentials {
                access_key_id: format!("ASIA{}", secret.to_uppercase()),
                secret_access_key: secret.to_string(),
                session_token: Some(format!("{secret}-token")),
            },
            expires_at: Some(time::OffsetDateTime::now_utc() + valid_for),
        })
    }

    fn holder(fetch: &std::sync::Arc<FakeImds>) -> InstanceCredentials {
        InstanceCredentials::with_fetch(Box::new(fetch.clone())).expect("the first read answers")
    }

    #[test]
    fn holds_one_credential_while_its_expiry_is_further_off_than_the_margin() {
        let imds = FakeImds::new(vec![
            expiring("first", time::Duration::hours(6)),
            expiring("second", time::Duration::hours(6)),
        ]);
        let credentials = holder(&imds);

        for _ in 0..3 {
            assert_eq!(
                credentials.current().expect("held").secret_access_key,
                "first"
            );
        }
        assert_eq!(imds.reads(), 1, "one read at construction and no more");
    }

    #[test]
    fn refreshes_inside_the_margin_before_the_credential_expires() {
        let imds = FakeImds::new(vec![
            expiring(
                "first",
                REFRESH_MARGIN.try_into().expect("a positive margin"),
            ),
            expiring("second", time::Duration::hours(6)),
        ]);
        let credentials = holder(&imds);

        assert_eq!(
            credentials.current().expect("refreshed").secret_access_key,
            "second",
            "a credential inside the margin is replaced before it expires, not after"
        );
        assert_eq!(imds.reads(), 2);
        assert_eq!(
            credentials.current().expect("held").secret_access_key,
            "second",
            "and the replacement is then held"
        );
        assert_eq!(imds.reads(), 2);
    }

    #[test]
    fn a_signature_made_after_expiry_carries_the_refreshed_secret() {
        let expired = expiring("first", time::Duration::seconds(-10));
        let fresh = expiring("second", time::Duration::hours(6));
        let imds = FakeImds::new(vec![expired.clone(), fresh.clone()]);
        let credentials = holder(&imds);

        let path = "/2013-04-01/hostedzone/Z1/rrset";
        let signed = route53_get(
            &credentials.current().expect("refreshed"),
            DEFAULT_HOST,
            path,
        );
        let with_fresh = route53_get(
            &fresh.expect("a credential").credentials,
            DEFAULT_HOST,
            path,
        );
        let with_expired = route53_get(
            &expired.expect("a credential").credentials,
            DEFAULT_HOST,
            path,
        );

        assert_eq!(signed, with_fresh);
        assert_ne!(signed, with_expired);
    }

    #[test]
    fn a_refresh_failure_keeps_signing_while_the_credential_is_still_valid() {
        let imds = FakeImds::new(vec![
            expiring("first", time::Duration::seconds(60)),
            Err("connection refused".to_string()),
        ]);
        let credentials = holder(&imds);

        assert_eq!(
            credentials
                .current()
                .expect("still valid")
                .secret_access_key,
            "first"
        );
        assert_eq!(imds.reads(), 2);
        assert_eq!(
            credentials
                .current()
                .expect("still valid")
                .secret_access_key,
            "first"
        );
        assert_eq!(imds.reads(), 3, "and the next request tries again");
    }

    #[test]
    fn a_refresh_failure_past_expiry_reaches_the_caller() {
        let imds = FakeImds::new(vec![
            expiring("first", time::Duration::seconds(-10)),
            Err("connection refused".to_string()),
        ]);
        let credentials = holder(&imds);

        assert!(
            matches!(credentials.current(), Err(ImdsError::Request(_))),
            "an expired credential and an unreachable IMDS is an error, not a signature \
             route53 will reject"
        );
        assert!(
            matches!(credentials.current(), Err(ImdsError::Request(_))),
            "and it stays retryable rather than latching"
        );
    }

    #[test]
    fn concurrent_signers_make_one_refresh_between_them() {
        // The delay is long enough that every thread below reaches
        // `current()` while the refresh is still in flight.
        let imds = FakeImds::with_delay(
            vec![
                expiring("first", time::Duration::seconds(10)),
                expiring("second", time::Duration::hours(6)),
            ],
            Duration::from_millis(50),
        );
        let credentials = std::sync::Arc::new(holder(&imds));

        let signers: Vec<_> = (0..8)
            .map(|_| {
                let credentials = credentials.clone();
                thread::spawn(move || credentials.current().expect("refreshed").secret_access_key)
            })
            .collect();
        for signer in signers {
            assert_eq!(signer.join().expect("the thread signs"), "second");
        }

        assert_eq!(
            imds.reads(),
            2,
            "one read at construction and one refresh shared by every waiter"
        );
    }

    /// Why every caller of `publish`, `provision` or `delete` has to reach
    /// this file's blocking client through `tokio::task::spawn_blocking`
    /// rather than inline on an async task.
    ///
    /// Building a `reqwest::blocking::Client`, as `api_client` does, builds
    /// a shell `current_thread` runtime purely to check it is safe to block, then drops it; dropping
    /// a runtime blocks the thread until its worker shuts down, and — in a
    /// debug build only, per reqwest's own `wait::enter` — that panics when
    /// the thread already belongs to a runtime that forbids blocking, which
    /// is exactly what an async task on `#[tokio::main]`'s two-worker
    /// runtime is. `spawn_blocking` runs on a thread the runtime itself
    /// marks safe to block, so the same call there does not panic.
    ///
    /// This is not a hazard specific to `didbot-dns`: it is why `routes.rs`
    /// crosses `spawn_blocking` for every call into this file's provider
    /// rather than calling `Registry::provision`/`delete` inline.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn building_the_blocking_client_inline_on_an_async_task_is_the_hazard_spawn_blocking_avoids(
    ) {
        let inline = std::panic::catch_unwind(api_client);
        #[cfg(debug_assertions)]
        assert!(
            inline.is_err(),
            "building the blocking client no longer panics when done inline on an async \
             task; if a reqwest upgrade fixed this, every caller that crosses spawn_blocking \
             for it may not need to any more"
        );
        #[cfg(not(debug_assertions))]
        assert!(
            inline.is_ok(),
            "outside a debug build this should not panic, only silently risk blocking a worker"
        );

        let via_spawn_blocking =
            tokio::task::spawn_blocking(|| std::panic::catch_unwind(api_client))
                .await
                .expect("the blocking task itself must not panic");
        assert!(
            via_spawn_blocking.is_ok(),
            "spawn_blocking should isolate the same construction from the panic the inline \
             case hits in a debug build"
        );
    }

    /// A Route53 endpoint that accepts the connection and never answers
    /// costs a call `didbot_http::DEFAULT_TIMEOUT`, then a timeout error.
    /// Real time: the blocking client runs its own runtime.
    #[test]
    fn the_api_client_gives_up_on_an_endpoint_that_never_answers() {
        let addr = didbot_http::test_support::hung_server();
        let started = Instant::now();
        let err = api_client()
            .get(format!("http://{addr}/"))
            .send()
            .expect_err("nothing ever answers");
        assert!(err.is_timeout(), "{err:?}");
        assert!(started.elapsed() >= didbot_http::DEFAULT_TIMEOUT);
    }

    /// The same for IMDS, at its own shorter timeout.
    #[test]
    fn the_imds_client_gives_up_on_an_endpoint_that_never_answers() {
        let addr = didbot_http::test_support::hung_server();
        let started = Instant::now();
        let err = imds_client()
            .expect("the client builds")
            .get(format!("http://{addr}/"))
            .send()
            .expect_err("nothing ever answers");
        assert!(err.is_timeout(), "{err:?}");
        let elapsed = started.elapsed();
        assert!(
            elapsed >= IMDS_TIMEOUT && elapsed < didbot_http::DEFAULT_TIMEOUT,
            "{elapsed:?}"
        );
    }
}