fakecloud_core/

protocol.rs

use bytes::Bytes;
use http::HeaderMap;
use std::collections::HashMap;

/// The wire protocol used by an AWS service.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AwsProtocol {
    /// Query protocol: form-encoded body, Action param, XML response.
    /// Used by: SQS, SNS, IAM, STS.
    Query,
    /// EC2 Query protocol: a variant of Query. Requests are wire-identical to
    /// `Query` (form-encoded body, `Action` param), but responses — including
    /// errors — use EC2's distinct XML envelope. Where `Query`/awsQuery emits
    /// `<ErrorResponse><Error><Type/><Code/><Message/></Error><RequestId/>`,
    /// ec2Query emits `<Response><Errors><Error><Code/><Message/></Error></Errors><RequestID/></Response>`
    /// (root `<Response>`, an `<Errors>` wrapper, no `<Type>`, and capital-ID
    /// `<RequestID>`). The AWS SDKs (aws-sdk-go-v2/Terraform, aws-sdk-js v3,
    /// aws-sdk-rust, aws-sdk-java v2) parse EC2 error codes strictly against
    /// this shape. Used by: EC2.
    Ec2Query,
    /// JSON protocol: JSON body, X-Amz-Target header, JSON response.
    /// Used by: SSM, EventBridge, DynamoDB, SecretsManager, KMS, CloudWatch Logs.
    Json,
    /// REST protocol: HTTP method + path-based routing, XML responses.
    /// Used by: S3, API Gateway, Route53.
    Rest,
    /// REST-JSON protocol: HTTP method + path-based routing, JSON responses.
    /// Used by: Lambda, SES v2.
    RestJson,
    /// Smithy RPC v2 CBOR: `POST /service/{Service}/operation/{Operation}`
    /// with `smithy-protocol: rpc-v2-cbor` and CBOR bodies. The request body is
    /// decoded to the awsJson document and handled on the service's JSON path;
    /// see [`crate::cbor`]. Used by: CloudWatch (current aws-sdk-rust).
    RpcV2Cbor,
}

/// Pick the Query-family protocol for a resolved service. EC2 speaks the
/// `ec2Query` variant (distinct error/response envelope); every other
/// Action-param service uses plain `awsQuery`.
fn query_protocol_for(service: &str) -> AwsProtocol {
    if service == "ec2" {
        AwsProtocol::Ec2Query
    } else {
        AwsProtocol::Query
    }
}

/// Services that use REST protocol with XML responses (detected from SigV4 credential scope).
const REST_XML_SERVICES: &[&str] = &["s3", "cloudfront", "route53"];

/// Services that use REST protocol with JSON responses (detected from SigV4 credential scope).
const REST_JSON_SERVICES: &[&str] = &[
    "managedblockchain",
    "lambda",
    "ses",
    "apigateway",
    "bedrock",
    "bedrock-agent",
    "bedrock-agent-runtime",
    "scheduler",
    "batch",
    "pipes",
    "rds-data",
    "dsql",
    "resource-groups",
    "eks",
    "glacier",
    "backup",
    // AWS Resource Access Manager: restJson1 control plane (single-segment
    // `@http` POST/DELETE URIs + JSON bodies).
    "ram",
    // Amazon S3 Tables: restJson1 control plane (path-labelled `@http` URIs
    // over the table bucket -> namespace -> table hierarchy + JSON bodies).
    "s3tables",
    // AWS Lake Formation: restJson1 governance control plane over Glue
    // (single-segment `@http` `POST /<Op>` URIs + JSON bodies).
    "lakeformation",
    // Amazon OpenSearch Service + Amazon Elasticsearch Service both sign as
    // `es` and speak restJson1; one service handles both, splitting on the
    // URL path version prefix.
    "es",
    "account",
    // AWS AppConfig control plane + AppConfig Data plane both sign as
    // `appconfig` and speak restJson1; one service handles both, splitting on
    // the URL path (control `/applications/...` vs data `/configuration...`).
    "appconfig",
    // AWS CodeArtifact: restJson1 control plane (`@http` method + path routing,
    // multi-value query params, JSON bodies).
    "codeartifact",
    // Amazon EFS: restJson1 control plane (path-labelled `@http` URIs over file
    // systems, mount targets, access points; JSON bodies). Signs as
    // `elasticfilesystem`.
    "elasticfilesystem",
    // Amazon MQ: restJson1 control plane (path-labelled `@http` URIs over
    // brokers, configurations, users, and tags; JSON bodies). Signs as `mq`.
    "mq",
    // Amazon MSK (Managed Streaming for Apache Kafka): restJson1 control plane
    // (path-labelled `@http` URIs over clusters, configurations, operations,
    // replicators, VPC connections, and topics; JSON bodies). Signs as `kafka`.
    "kafka",
    // Amazon MWAA (Managed Workflows for Apache Airflow): restJson1 control
    // plane (path-labelled `@http` URIs over environments, access tokens, and
    // tags; JSON bodies). Signs as `airflow`, normalized to `mwaa`.
    "mwaa",
    // AWS Fault Injection Simulator: restJson1 control plane (path-labelled
    // `@http` URIs over experiment templates, experiments, the actions and
    // target-resource-type catalogs, target-account configurations, safety
    // levers, and tags; JSON bodies). Signs as `fis`.
    "fis",
    // AWS X-Ray: restJson1 control plane + trace data plane (fixed `@http`
    // `POST /<Op>` URIs over trace segments, the derived service graph,
    // sampling rules, groups, encryption config, and tags; JSON bodies).
    // Signs as `xray`.
    "xray",
    // AWS AppSync: restJson1 control plane + schema state (RESTful `@http`
    // method + path routing over GraphQL APIs, data sources, resolvers,
    // functions, types, caches, domain names, the Event-API surface, and
    // source-API associations; JSON bodies). Signs as `appsync`.
    "appsync",
    // AWS Amplify: restJson1 hosting control plane (path-labelled `@http` URIs
    // over apps, branches, domain associations, webhooks, backend
    // environments, jobs/deployments, artifacts, and tags; JSON bodies).
    // Signs as `amplify`.
    "amplify",
    // AWS Elemental MediaConvert: restJson1 video-transcoding control plane
    // (path-labelled `@http` URIs under `/2017-08-29` over queues, presets, job
    // templates, jobs, policy, endpoints, and tags; JSON bodies). Signs as
    // `mediaconvert`.
    "mediaconvert",
    // AWS Serverless Application Repository: restJson1 control plane
    // (path-labelled `@http` URIs under `/applications` over applications,
    // versions, sharing policies, CloudFormation change sets/templates, and
    // dependencies; JSON bodies). Signs as `serverlessrepo`.
    "serverlessrepo",
    // AWS IoT Data Plane: restJson1 device-shadow + retained-message data plane
    // (path-labelled `@http` URIs over `/things/{thingName}/shadow`,
    // `/topics/{topic}`, `/retainedMessage`, and `/connections`; raw
    // `@httpPayload` shadow documents). Signs as `iotdata`.
    "iotdata",
    // Amazon Pinpoint: restJson1 control plane (RESTful `@http` method + path
    // routing under `/v1/apps/...`, `/v1/templates/...`, `/v1/recommenders`,
    // `/v1/tags/...` over apps, campaigns, segments, endpoints, channels,
    // journeys, templates, jobs, event streams, recommenders, and tags; JSON
    // bodies). Signs as `mobiletargeting`, normalized to `pinpoint`.
    "pinpoint",
    // AWS IoT Core control plane: restJson1 registry / jobs / rules / security
    // control plane (path-labelled `@http` URIs over `/things/{thingName}`,
    // `/policies/{policyName}`, `/jobs/{jobId}`, `/rules/{ruleName}`, ...).
    // Signs as `iot`.
    "iot",
    // AWS IoT Wireless control plane: restJson1 LoRaWAN / Sidewalk registry
    // (collection-POST creates + path-labelled reads over `/destinations`,
    // `/wireless-devices/{Identifier}`, `/fuota-tasks/{Id}`, ...). Signs as
    // `iotwireless`.
    "iotwireless",
];

/// Whether the request is an S3 or API Gateway request on which a
/// query-string `Action` is only a parameter, never a Query call: its `Host`
/// is an S3 endpoint (path-style or a bucket's virtual host) or an
/// execute-api host, its SigV4 scope (header or presign) is `s3` or
/// `execute-api`, or it is an S3 SigV2 presign (`AWSAccessKeyId`,
/// `Signature` and `Expires`; a SigV2 Query signature carries
/// `SignatureVersion` and `Timestamp` instead). Dispatch adds the cases it
/// alone can tell: an unsigned plain-host path addressing an existing bucket,
/// and an API Gateway custom-domain host.
pub fn query_action_is_a_parameter(
    headers: &HeaderMap,
    query_params: &HashMap<String, String>,
) -> bool {
    const PARAMETER_SERVICES: &[&str] = &["s3", "execute-api"];
    let host = parse_routing_host_from_headers(headers)
        .is_some_and(|h| PARAMETER_SERVICES.contains(&h.service.as_str()) || h.bucket.is_some())
        || headers
            .get(http::header::HOST)
            .and_then(|v| v.to_str().ok())
            .is_some_and(|h| h.contains(".execute-api."));
    let scope = extract_service_from_auth(headers)
        .or_else(|| {
            let credential = query_params.get("X-Amz-Credential")?;
            Some(normalize_service_name(credential.split('/').nth(3)?).to_string())
        })
        .is_some_and(|s| PARAMETER_SERVICES.contains(&s.as_str()));
    let sigv2_presign = query_params.contains_key("AWSAccessKeyId")
        && query_params.contains_key("Signature")
        && query_params.contains_key("Expires")
        && !query_params.contains_key("SignatureVersion");
    host || scope || sigv2_presign
}

/// A query-string `Action` as a Query call: of the service the signature
/// names, else the one the action names, else the one the `Host` names --
/// unless the request is S3's or API Gateway's
/// ([`query_action_is_a_parameter`]).
fn query_string_action(
    headers: &HeaderMap,
    query_params: &HashMap<String, String>,
) -> Option<DetectedRequest> {
    let action = query_params.get("Action")?;
    if query_action_is_a_parameter(headers, query_params) {
        return None;
    }
    let service = extract_service_from_auth(headers)
        .or_else(|| infer_service_from_action(action))
        .or_else(|| parse_routing_host_from_headers(headers).map(|h| h.service))?;
    let protocol = query_protocol_for(&service);
    Some(DetectedRequest {
        service,
        action: action.clone(),
        protocol,
    })
}

/// Detected service name and action from an incoming HTTP request.
#[derive(Debug, Clone)]
pub struct DetectedRequest {
    pub service: String,
    pub action: String,
    pub protocol: AwsProtocol,
}

/// Header-only service detection. Skips the form-encoded body sniff so
/// the dispatch path can decide whether to stream or buffer the body
/// without first reading it. Returns `None` when only a body sniff
/// would succeed; the caller must then fall back to [`detect_service`]
/// after buffering. Used to opt streaming routes (S3 PutObject /
/// UploadPart, ECR OCI v2 blob upload) out of the global body cap.
pub fn detect_service_headers_only(
    headers: &HeaderMap,
    query_params: &HashMap<String, String>,
) -> Option<DetectedRequest> {
    // Mirrors `detect_service` minus step 3 (form-body sniff).
    if let Some(target) = headers.get("x-amz-target").and_then(|v| v.to_str().ok()) {
        return parse_amz_target(target);
    }
    if let Some(detected) = query_string_action(headers, query_params) {
        return Some(detected);
    }
    if let Some(service) = extract_service_from_auth(headers) {
        if let Some(protocol) = rest_protocol_for(&service) {
            return Some(DetectedRequest {
                service,
                action: String::new(),
                protocol,
            });
        }
    }
    if let Some(credential) = query_params.get("X-Amz-Credential") {
        let parts: Vec<&str> = credential.split('/').collect();
        if parts.len() >= 4 {
            let service = normalize_service_name(parts[3]).to_string();
            if let Some(protocol) = rest_protocol_for(&service) {
                return Some(DetectedRequest {
                    service,
                    action: String::new(),
                    protocol,
                });
            }
        }
    }
    if query_params.contains_key("AWSAccessKeyId")
        && query_params.contains_key("Signature")
        && query_params.contains_key("Expires")
    {
        return Some(DetectedRequest {
            service: "s3".to_string(),
            action: String::new(),
            protocol: AwsProtocol::Rest,
        });
    }
    if let Some(host_info) = parse_routing_host_from_headers(headers) {
        if let Some(protocol) = rest_protocol_for(&host_info.service) {
            return Some(DetectedRequest {
                service: host_info.service,
                action: String::new(),
                protocol,
            });
        }
    }
    None
}

/// Whether [`detect_service_headers_only`] names the service
/// [`detect_service`] will route the request to, whatever its body: the
/// header detection rests on an `X-Amz-Target`, a query `Action` that names
/// a service, or a SigV4 `Authorization` scope, each of which `detect_service`
/// honors ahead of (or, for the scope, inside) its form-body sniff. A
/// detection resting on a presigned URL or the Host header alone is final
/// only without a form body ([`is_form_content_type`]): a form body naming an
/// `Action` routes the request to that action's service instead.
pub fn headers_name_final_service(
    headers: &HeaderMap,
    query_params: &HashMap<String, String>,
) -> bool {
    !is_form_content_type(headers)
        || headers.contains_key("x-amz-target")
        || query_params.get("Action").is_some_and(|action| {
            extract_service_from_auth(headers)
                .or_else(|| infer_service_from_action(action))
                .or_else(|| parse_routing_host_from_headers(headers).map(|h| h.service))
                .is_some()
        })
        || extract_service_from_auth(headers).is_some()
}

/// Whether the request declares a form body
/// (`application/x-www-form-urlencoded`, parameters such as `charset`
/// allowed): the only body Query detection reads an `Action` from, before or
/// after the body is read.
pub fn is_form_content_type(headers: &HeaderMap) -> bool {
    headers
        .get(http::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .and_then(|ct| ct.split(';').next())
        .is_some_and(|media| {
            media
                .trim()
                .eq_ignore_ascii_case("application/x-www-form-urlencoded")
        })
}

/// Detect the target service and action from HTTP request components.
pub fn detect_service(
    headers: &HeaderMap,
    query_params: &HashMap<String, String>,
    body: &Bytes,
) -> Option<DetectedRequest> {
    // 1. Check X-Amz-Target header (JSON protocol)
    if let Some(target) = headers.get("x-amz-target").and_then(|v| v.to_str().ok()) {
        return parse_amz_target(target);
    }

    // 2. A Query `Action` in the query string, for the service the request
    //    names (see [`query_string_action`]).
    if let Some(detected) = query_string_action(headers, query_params) {
        return Some(detected);
    }

    // 3. Try a form-encoded body. Only a body declared as a form is one: real
    //    AWS does not read `Action=...` from a `text/plain` (or untyped) body
    //    as an API call at all.
    if is_form_content_type(headers) {
        let form_params = decode_form_urlencoded(body);

        if let Some(action) = form_params.get("Action") {
            let service = extract_service_from_auth(headers)
                .or_else(|| infer_service_from_action(action))
                .or_else(|| parse_routing_host_from_headers(headers).map(|h| h.service));
            if let Some(service) = service {
                let protocol = query_protocol_for(&service);
                return Some(DetectedRequest {
                    service,
                    action: action.clone(),
                    protocol,
                });
            }
        }
    }

    // 4. Fallback: check auth header for REST-style services (S3, Lambda, SES, etc.)
    if let Some(service) = extract_service_from_auth(headers) {
        if let Some(protocol) = rest_protocol_for(&service) {
            return Some(DetectedRequest {
                service,
                action: String::new(), // REST services determine action from method+path
                protocol,
            });
        }
    }

    // 5. Check query params for presigned URL auth (X-Amz-Credential for SigV4)
    if let Some(credential) = query_params.get("X-Amz-Credential") {
        // Format: AKID/date/region/service/aws4_request
        let parts: Vec<&str> = credential.split('/').collect();
        if parts.len() >= 4 {
            let service = normalize_service_name(parts[3]).to_string();
            if let Some(protocol) = rest_protocol_for(&service) {
                return Some(DetectedRequest {
                    service,
                    action: String::new(),
                    protocol,
                });
            }
        }
    }

    // 6. Check for SigV2-style presigned URL (AWSAccessKeyId + Signature + Expires)
    //    Only match when all three SigV2 presigned-URL parameters are present so
    //    we don't accidentally claim non-S3 requests.
    if query_params.contains_key("AWSAccessKeyId")
        && query_params.contains_key("Signature")
        && query_params.contains_key("Expires")
    {
        return Some(DetectedRequest {
            service: "s3".to_string(),
            action: String::new(),
            protocol: AwsProtocol::Rest,
        });
    }

    // 7. Fallback: unsigned REST-style request carrying a LocalStack-shaped
    //    Host header. Lets fixtures and curl-style probes reach the right
    //    service without SigV4; signed requests were already handled in step 4.
    if let Some(host_info) = parse_routing_host_from_headers(headers) {
        if let Some(protocol) = rest_protocol_for(&host_info.service) {
            return Some(DetectedRequest {
                service: host_info.service,
                action: String::new(),
                protocol,
            });
        }
    }

    None
}

/// Service + region (and optional bucket) decoded from a `Host` header.
/// Covers both the LocalStack hostname convention
/// (`<service>.<region>.localhost.localstack.cloud[:port]`,
/// `<bucket>.s3.<region>.localhost.localstack.cloud[:port]`) and real AWS
/// service hostnames (`<service>.<region>.amazonaws.com`, S3 path-style
/// and virtual-hosted-style including the legacy no-region
/// `s3.amazonaws.com` / `<bucket>.s3.amazonaws.com` forms, the older
/// dash-separated `s3-<region>.amazonaws.com` form, the dualstack, FIPS and
/// static-website endpoints), under every AWS partition's DNS suffix.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoutingHost {
    pub service: String,
    pub region: String,
    /// Set only for virtual-hosted-style S3 hostnames.
    pub bucket: Option<String>,
}

const LOCALSTACK_SUFFIX: &str = ".localhost.localstack.cloud";

/// The service an endpoint's leading label names. SES (v1 and v2) is served
/// at `email.<region>`; every other label is the service's own name.
fn host_label_service(label: &str) -> String {
    match label {
        "email" => "ses".to_string(),
        other => other.to_string(),
    }
}

/// Services whose commercial-partition endpoint is the region-less
/// `<service>.amazonaws.com` (`s3.amazonaws.com` is handled with S3's other
/// hosts).
const GLOBAL_ENDPOINT_SERVICES: &[&str] = &["iam", "sts", "route53", "cloudfront"];

/// Parse a `Host` header value for a LocalStack- or AWS-shaped hostname.
/// Returns `None` for anything that doesn't match — callers fall through
/// to their existing detection path.
pub fn parse_routing_host(host: &str) -> Option<RoutingHost> {
    let hostname = host.split(':').next()?;
    if hostname.is_empty() {
        return None;
    }
    let hostname = hostname.to_ascii_lowercase();
    if let Some(prefix) = hostname.strip_suffix(LOCALSTACK_SUFFIX) {
        return parse_localstack_prefix(prefix);
    }
    // The commercial partition's global endpoints carry no region
    // (`route53.amazonaws.com`); they sign for us-east-1.
    if let Some(service) = hostname
        .strip_suffix(".amazonaws.com")
        .filter(|s| GLOBAL_ENDPOINT_SERVICES.contains(s))
    {
        return Some(RoutingHost {
            service: service.to_string(),
            region: "us-east-1".to_string(),
            bucket: None,
        });
    }
    // Endpoint hostnames share one shape across partitions; only the DNS
    // suffix differs, so accept every partition's.
    fakecloud_aws::endpoint::DNS_SUFFIXES
        .iter()
        .find_map(|suffix| hostname.strip_suffix(suffix)?.strip_suffix('.'))
        .and_then(parse_aws_prefix)
}

/// Whether the request is addressed to an S3 Control endpoint
/// (`[<account-id>.]s3-control.<region>...`). S3 Control is served by the
/// `s3` handler, so the host is the only thing telling the two apart. The
/// `s3-control` label must lead the host (after the SDK's account-ID
/// prefix), so a bucket whose name merely contains `s3-control` stays S3.
pub fn is_s3_control_host(headers: &HeaderMap) -> bool {
    let Some(host) = headers.get("host").and_then(|v| v.to_str().ok()) else {
        return false;
    };
    let hostname = host.split(':').next().unwrap_or("").to_ascii_lowercase();
    let labels: Vec<&str> = hostname.split('.').collect();
    let control_label = match labels.as_slice() {
        ["s3-control", ..] => true,
        [account, "s3-control", ..] => {
            account.len() == 12 && account.bytes().all(|b| b.is_ascii_digit())
        }
        _ => false,
    };
    control_label && parse_routing_host(host).is_none_or(|h| h.bucket.is_none())
}

/// Normalize a `Host` header value for comparing against configured domain
/// names: drop the port (`api.example.com:4566`), any trailing root dot, and
/// lowercase it (DNS names are case-insensitive). A bracketed IPv6 literal
/// keeps its brackets.
pub fn normalize_host(host: &str) -> String {
    let host = host.trim();
    let without_port = if let Some(rest) = host.strip_prefix('[') {
        match rest.find(']') {
            Some(end) => &host[..end + 2],
            None => host,
        }
    } else {
        match host.rsplit_once(':') {
            Some((name, port)) if port.chars().all(|c| c.is_ascii_digit()) => name,
            _ => host,
        }
    };
    without_port.trim_end_matches('.').to_ascii_lowercase()
}

/// The normalized `Host` header (see [`normalize_host`]), if present.
pub fn normalized_host_from_headers(headers: &HeaderMap) -> Option<String> {
    let host = headers.get("host")?.to_str().ok()?;
    let host = normalize_host(host);
    (!host.is_empty()).then_some(host)
}

/// Pull the `Host` header and parse it with [`parse_routing_host`].
pub fn parse_routing_host_from_headers(headers: &HeaderMap) -> Option<RoutingHost> {
    let host = headers.get("host")?.to_str().ok()?;
    parse_routing_host(host)
}

fn parse_localstack_prefix(prefix: &str) -> Option<RoutingHost> {
    if prefix.is_empty() {
        return None;
    }
    let labels: Vec<&str> = prefix.split('.').collect();
    if labels.iter().any(|l| l.is_empty()) {
        return None;
    }
    // API Gateway execute-api hosts lead with the API id, not a service:
    // `<api-id>.execute-api[.<region>]`. Without a region label the host
    // carries no region at all, so let the caller fall back to its default
    // rather than reading `execute-api` as one.
    if let Some(host) = parse_execute_api_labels(&labels) {
        return host;
    }
    match labels.len() {
        2 => Some(RoutingHost {
            service: host_label_service(labels[0]),
            region: labels[1].to_string(),
            bucket: None,
        }),
        n if n >= 3 && labels[n - 2] == "s3" => {
            let bucket = labels[..n - 2].join(".");
            Some(RoutingHost {
                service: "s3".to_string(),
                region: labels[n - 1].to_string(),
                bucket: Some(bucket),
            })
        }
        n if n >= 3 && labels[n - 2] == "s3-accesspoint" => {
            let bucket = labels[..n - 2].join(".");
            Some(RoutingHost {
                service: "s3".to_string(),
                region: labels[n - 1].to_string(),
                bucket: Some(bucket),
            })
        }
        n if n >= 3 && labels[n - 2] == "s3-control" => Some(RoutingHost {
            service: "s3".to_string(),
            region: labels[n - 1].to_string(),
            bucket: None,
        }),
        _ => None,
    }
}

/// Classify an API Gateway execute-api host prefix (`<api-id>.execute-api`,
/// optionally followed by `.<region>`). Returns `None` when the labels are not
/// an execute-api host, `Some(None)` for one without a region, and
/// `Some(Some(host))` for one naming its region.
fn parse_execute_api_labels(labels: &[&str]) -> Option<Option<RoutingHost>> {
    match labels {
        [_api_id, "execute-api"] => Some(None),
        [_api_id, "execute-api", region] => Some(Some(RoutingHost {
            service: "execute-api".to_string(),
            region: region.to_string(),
            bucket: None,
        })),
        _ => None,
    }
}

/// Parse the prefix before an AWS partition DNS suffix (`.amazonaws.com`, ...).
///
/// Handles every variant AWS has shipped for the common REST/Query services:
///
/// - `<service>.<region>` — modern regional endpoint (most services).
/// - `s3.<region>` — modern path-style S3.
/// - `<bucket>.s3.<region>` — modern virtual-hosted S3 (bucket may contain dots).
/// - `s3` — legacy S3 global endpoint (implicitly `us-east-1`).
/// - `<bucket>.s3` — legacy virtual-hosted S3 (implicitly `us-east-1`).
/// - `s3-<region>` — older dash-separated path-style S3.
/// - `<bucket>.s3-<region>` — older dash-separated virtual-hosted S3.
/// - `[<bucket>.]s3.dualstack.<region>`, `[<bucket>.]s3-fips[.dualstack].<region>`
///   — dualstack and FIPS S3 endpoints.
/// - `<bucket>.s3-website-<region>` / `<bucket>.s3-website.<region>` — S3
///   static-website endpoints.
fn parse_aws_prefix(prefix: &str) -> Option<RoutingHost> {
    if prefix.is_empty() {
        return None;
    }
    let labels: Vec<&str> = prefix.split('.').collect();
    if labels.iter().any(|l| l.is_empty()) {
        return None;
    }
    let last = *labels.last()?;

    // `<bucket>.s3-website-<region>`: dash-separated S3 static-website
    // endpoint. Checked before the generic `s3-<region>` form below, which
    // would otherwise read the region as `website-<region>`.
    if let Some(region) = last.strip_prefix("s3-website-") {
        if !region.is_empty() && labels.len() >= 2 {
            return Some(RoutingHost {
                service: "s3".to_string(),
                region: region.to_string(),
                bucket: Some(labels[..labels.len() - 1].join(".")),
            });
        }
    }

    // `s3-<region>` as the last label: dash-separated S3. Bucket, if any,
    // is whatever precedes it.
    if let Some(region) = last.strip_prefix("s3-") {
        if !region.is_empty() {
            let bucket = if labels.len() >= 2 {
                Some(labels[..labels.len() - 1].join("."))
            } else {
                None
            };
            return Some(RoutingHost {
                service: "s3".to_string(),
                region: region.to_string(),
                bucket,
            });
        }
    }

    // Legacy global S3: last label is `s3`, no region present. `s3` on its
    // own is the path-style global endpoint; anything preceding it is the
    // bucket (including dotted names like `a.b.s3.amazonaws.com`).
    if last == "s3" {
        if labels.len() == 1 {
            return Some(RoutingHost {
                service: "s3".to_string(),
                region: "us-east-1".to_string(),
                bucket: None,
            });
        }
        return Some(RoutingHost {
            service: "s3".to_string(),
            region: "us-east-1".to_string(),
            bucket: Some(labels[..labels.len() - 1].join(".")),
        });
    }

    // `s3-accesspoint.<region>` — path-style access point endpoint.
    // `{alias}-{account-id}.s3-accesspoint.<region>` — virtual-hosted access point.
    if last == "s3-accesspoint" {
        if labels.len() == 2 {
            return Some(RoutingHost {
                service: "s3".to_string(),
                region: labels[0].to_string(),
                bucket: None,
            });
        }
        // Virtual-hosted form needs at least {alias}.{region}.s3-accesspoint, i.e.
        // 3+ labels. A bare "s3-accesspoint" host (1 label) must not reach the
        // `len() - 2` slice, which would underflow and panic.
        if labels.len() >= 3 {
            let bucket = labels[..labels.len() - 2].join(".");
            return Some(RoutingHost {
                service: "s3".to_string(),
                region: labels[labels.len() - 1].to_string(),
                bucket: Some(bucket),
            });
        }
    }

    // `s3-control.<region>` or `{account-id}.s3-control.<region>` — S3
    // Control endpoint (access point management).
    if labels.len() >= 2 && labels[labels.len() - 2] == "s3-control" {
        return Some(RoutingHost {
            service: "s3".to_string(),
            region: last.to_string(),
            bucket: None,
        });
    }

    // Region-last S3 endpoints: `[<bucket>.]<endpoint>.<region>` where the
    // endpoint is `s3`, `s3-fips`, either of those followed by `.dualstack`,
    // or `s3-website` (the dot-separated static-website endpoint). Whatever
    // precedes the endpoint is the bucket (dotted names included).
    let before_region = &labels[..labels.len() - 1];
    let (endpoint_labels, s3_endpoint) = match before_region {
        [rest @ .., s3, "dualstack"] if matches!(*s3, "s3" | "s3-fips") => (rest, true),
        [rest @ .., endpoint] if matches!(*endpoint, "s3" | "s3-fips" | "s3-website") => {
            (rest, true)
        }
        _ => (before_region, false),
    };
    if s3_endpoint {
        return Some(RoutingHost {
            service: "s3".to_string(),
            region: last.to_string(),
            bucket: (!endpoint_labels.is_empty()).then(|| endpoint_labels.join(".")),
        });
    }

    // `<api-id>.execute-api.<region>` — API Gateway's default endpoint.
    if let Some(host) = parse_execute_api_labels(&labels) {
        return host;
    }

    // `<service>.<region>` — the common case for every other service.
    match labels.as_slice() {
        [service, region] => Some(RoutingHost {
            service: host_label_service(service),
            region: region.to_string(),
            bucket: None,
        }),
        _ => None,
    }
}

/// Detect a Smithy RPC v2 CBOR request: the `smithy-protocol: rpc-v2-cbor`
/// header plus a `/service/{ServiceName}/operation/{Operation}` path. The
/// service shape name is the same identifier awsJson services use as their
/// `X-Amz-Target` prefix, so it resolves through the same table.
pub fn detect_rpc_v2_cbor(headers: &HeaderMap, path: &str) -> Option<DetectedRequest> {
    let is_cbor = headers
        .get(crate::cbor::SMITHY_PROTOCOL_HEADER)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|v| v.trim().eq_ignore_ascii_case(crate::cbor::RPC_V2_CBOR));
    if !is_cbor {
        return None;
    }
    let (service_shape, operation) = crate::cbor::parse_rpc_v2_path(path)?;
    let detected = parse_amz_target(&format!("{service_shape}.{operation}"))?;
    Some(DetectedRequest {
        protocol: AwsProtocol::RpcV2Cbor,
        ..detected
    })
}

/// Parse `X-Amz-Target: AWSEvents.PutEvents` -> service=events, action=PutEvents
/// Parse `X-Amz-Target: AmazonSSM.GetParameter` -> service=ssm, action=GetParameter
pub(crate) fn parse_amz_target(target: &str) -> Option<DetectedRequest> {
    let (prefix, action) = target.rsplit_once('.')?;

    let service = match prefix {
        "AWSEvents" => "events",
        "AmazonSSM" => "ssm",
        "AmazonSQS" => "sqs",
        "AmazonSNS" => "sns",
        "DynamoDB_20120810" => "dynamodb",
        "DynamoDBStreams_20120810" => "dynamodbstreams",
        "Logs_20140328" => "logs",
        s if s.starts_with("secretsmanager") => "secretsmanager",
        s if s.starts_with("TrentService") => "kms",
        s if s.starts_with("AWSCognitoIdentityProviderService") => "cognito-idp",
        s if s.starts_with("AWSCognitoIdentityService") => "cognito-identity",
        s if s.starts_with("Kinesis_20131202") => "kinesis",
        s if s.starts_with("AmazonEC2ContainerRegistry_V") => "ecr",
        s if s.starts_with("AmazonEC2ContainerServiceV") => "ecs",
        s if s.starts_with("AWSStepFunctions") => "states",
        s if s.starts_with("AWSOrganizationsV") => "organizations",
        "CertificateManager" => "acm",
        "ACMPrivateCA" => "acm-pca",
        // Amazon Route 53 Resolver (resolver endpoints/rules, query logging, DNS
        // Firewall): awsJson1_1. The service shape short name is the target
        // prefix. Distinct from Route 53 (`route53`, a REST-XML service).
        "Route53Resolver" => "route53resolver",
        // AWS Config (config recorder / rules / compliance): awsJson1_1. The
        // service shape short name is the target prefix.
        "StarlingDoveService" => "config",
        "AnyScaleFrontendService" => "application-autoscaling",
        // Match the WAFv2 target version exactly so legacy WAF Classic
        // (`AWSWAF_*` without the `_20190729` suffix) doesn't get routed here.
        "AWSWAF_20190729" => "wafv2",
        "AmazonAthena" => "athena",
        s if s.starts_with("Firehose_") => "firehose",
        "AWSGlue" => "glue",
        // Amazon EMR (Elastic MapReduce): awsJson1.1. The service shape short
        // name is the target prefix (`ElasticMapReduce.<Operation>`).
        "ElasticMapReduce" => "emr",
        // Amazon Textract (document text/analysis extraction): awsJson1_1. The
        // service shape short name is the target prefix (`Textract.<Operation>`).
        "Textract" => "textract",
        // Amazon Transcribe: awsJson1.1. The service shape short name is the
        // target prefix (`Transcribe.<Operation>`).
        "Transcribe" => "transcribe",
        // Amazon Translate: awsJson1_1. The service shape short name
        // (`AWSShineFrontendService_20170701.<Operation>`) is the target prefix.
        "AWSShineFrontendService_20170701" => "translate",
        // AWS Shield / Shield Advanced: awsJson1_1. The service shape short
        // name (`AWSShield_20160616.<Operation>`) is the target prefix.
        "AWSShield_20160616" => "shield",
        // Amazon Comprehend (NLP): awsJson1.1. The service shape name carries the
        // dated version (`Comprehend_20171127.<Operation>`).
        "Comprehend_20171127" => "comprehend",
        // Amazon SWF (Simple Workflow Service): awsJson1_0. The service shape
        // short name is the target prefix (`SimpleWorkflowService.<Operation>`).
        "SimpleWorkflowService" => "swf",
        // Amazon Timestream (Write + Query): awsJson1_0. BOTH the write and
        // query SDK clients carry the SAME dated target prefix
        // (`Timestream_20181101.<Operation>`); one fakecloud crate serves both.
        "Timestream_20181101" => "timestream",
        // AWS Support: awsJson1.1. The service shape carries the dated version
        // (`AWSSupport_20130415.<Operation>`).
        "AWSSupport_20130415" => "support",
        "CloudApiService" => "cloudcontrolapi",
        "ResourceGroupsTaggingAPI_20170126" => "tagging",
        "AmazonMemoryDB" => "memorydb",
        // Amazon Managed Service for Apache Flink (formerly Kinesis Data
        // Analytics v2): awsJson1.1. The SigV4 signing name is
        // `kinesisanalytics`; the internal fakecloud service key is
        // `kinesisanalyticsv2`.
        s if s.starts_with("KinesisAnalytics_20180523") => "kinesisanalyticsv2",
        // Cloud Map (servicediscovery): awsJson1.1, target prefix carries the
        // dated Route53 Auto Naming service version.
        "Route53AutoNaming_v20170314" => "servicediscovery",
        // Database Migration Service: awsJson1.1.
        "AmazonDMSv20160101" => "dms",
        // CloudTrail: awsJson1.1.
        // aws-sdk-go-v2 / smithy clients (and terraform) send the short shape
        // name; aws-sdk-go-v1 sends the fully-qualified form. Accept both.
        "CloudTrail_20131101" => "cloudtrail",
        "com.amazonaws.cloudtrail.v20131101.CloudTrail_20131101" => "cloudtrail",
        // Cost Explorer: awsJson1.1. aws-sdk / smithy clients (and terraform)
        // send the short service-shape name; older clients may send the
        // fully-qualified form. Accept both.
        "AWSInsightsIndexService" => "ce",
        "com.amazonaws.costexplorer.v20171025.AWSInsightsIndexService" => "ce",
        // Transfer Family: awsJson1.1.
        "TransferService" => "transfer",
        // AWS CodeBuild: awsJson1.1, target prefix is the dated service shape.
        "CodeBuild_20161006" => "codebuild",
        // AWS CodeCommit: awsJson1.1, target prefix is the dated service shape.
        "CodeCommit_20150413" => "codecommit",
        // Service Quotas: awsJson1.1, target prefix is the dated service shape.
        "ServiceQuotasV20190624" => "servicequotas",
        // IAM Identity Center Identity Store: awsJson1.1.
        "AWSIdentityStore" => "identitystore",
        // IAM Identity Center SSO Admin: awsJson1.1.
        "SWBExternalService" => "sso",
        // Verified Permissions: awsJson1.0.
        "VerifiedPermissions" => "verifiedpermissions",
        // CodeConnections (successor to CodeStar Connections): awsJson1.0.
        "CodeConnections_20231201" => "codeconnections",
        // Legacy CodeStar Connections API (same operations as CodeConnections);
        // the terraform `aws_codestarconnections_connection` resource still
        // signs with this dated prefix (note the lowercase `connections`, as
        // emitted by the aws-sdk-go-v2 codestarconnections client), so route it
        // to the same handler.
        "CodeStar_connections_20191201" => "codeconnections",
        // AWS CodeDeploy: awsJson1.1, target prefix is the dated service shape.
        "CodeDeploy_20141006" => "codedeploy",
        // AWS CodePipeline: awsJson1.1, target prefix is the dated service shape.
        "CodePipeline_20150709" => "codepipeline",
        // CloudWatch advertises awsJson1_0 (target service shape
        // `GraniteServiceVersion20100801`) alongside the legacy awsQuery
        // protocol. Newer SDKs (aws-sdk-rust / js-v3 / go-v2) POST with
        // `X-Amz-Target: GraniteServiceVersion20100801.<Operation>` and a JSON
        // body. The service registry key is `monitoring`.
        s if s.starts_with("GraniteServiceVersion") => "monitoring",
        // Amazon SageMaker: awsJson1.1. The service shape short name is the
        // target prefix (`SageMaker.<Operation>`).
        "SageMaker" => "sagemaker",
        _ => return None,
    };

    Some(DetectedRequest {
        service: service.to_string(),
        action: action.to_string(),
        protocol: AwsProtocol::Json,
    })
}

/// Returns the REST protocol variant for a service, or None if not a REST service.
/// Whether `service`'s API is a REST protocol (restXml or restJson1), so a
/// REST request to its API host is one of its operations. ECR's API is
/// awsJson: a REST request routed to ECR is its OCI registry.
pub fn is_rest_api_service(service: &str) -> bool {
    rest_protocol_for(service).is_some()
}

fn rest_protocol_for(service: &str) -> Option<AwsProtocol> {
    if REST_XML_SERVICES.contains(&service) {
        Some(AwsProtocol::Rest)
    } else if REST_JSON_SERVICES.contains(&service) {
        Some(AwsProtocol::RestJson)
    } else {
        None
    }
}

/// Infer service from the action name when no SigV4 auth is present.
/// Some AWS operations (e.g., AssumeRoleWithSAML, AssumeRoleWithWebIdentity)
/// do not require authentication and won't have an Authorization header.
fn infer_service_from_action(action: &str) -> Option<String> {
    match action {
        "AssumeRole"
        | "AssumeRoleWithSAML"
        | "AssumeRoleWithWebIdentity"
        | "GetCallerIdentity"
        | "GetSessionToken"
        | "GetFederationToken"
        | "GetAccessKeyInfo"
        | "DecodeAuthorizationMessage"
        | "AssumeRoot"
        | "GetWebIdentityToken"
        | "GetDelegatedAccessToken" => Some("sts".to_string()),
        "CreateUser" | "DeleteUser" | "GetUser" | "ListUsers" | "CreateRole" | "DeleteRole"
        | "GetRole" | "ListRoles" | "CreatePolicy" | "DeletePolicy" | "GetPolicy"
        | "ListPolicies" | "AttachRolePolicy" | "DetachRolePolicy" | "CreateAccessKey"
        | "DeleteAccessKey" | "ListAccessKeys" | "ListRolePolicies" => Some("iam".to_string()),
        // SES v1 (Query protocol)
        "VerifyEmailIdentity"
        | "VerifyDomainIdentity"
        | "VerifyDomainDkim"
        | "ListIdentities"
        | "GetIdentityVerificationAttributes"
        | "GetIdentityDkimAttributes"
        | "DeleteIdentity"
        | "SetIdentityDkimEnabled"
        | "SetIdentityNotificationTopic"
        | "SetIdentityFeedbackForwardingEnabled"
        | "GetIdentityNotificationAttributes"
        | "GetIdentityMailFromDomainAttributes"
        | "SetIdentityMailFromDomain"
        | "SendEmail"
        | "SendRawEmail"
        | "SendTemplatedEmail"
        | "SendBulkTemplatedEmail"
        | "CreateTemplate"
        | "GetTemplate"
        | "ListTemplates"
        | "DeleteTemplate"
        | "UpdateTemplate"
        | "CreateConfigurationSet"
        | "DeleteConfigurationSet"
        | "DescribeConfigurationSet"
        | "ListConfigurationSets"
        | "CreateConfigurationSetEventDestination"
        | "UpdateConfigurationSetEventDestination"
        | "DeleteConfigurationSetEventDestination"
        | "GetSendQuota"
        | "GetSendStatistics"
        | "GetAccountSendingEnabled"
        | "CreateReceiptRuleSet"
        | "DeleteReceiptRuleSet"
        | "DescribeReceiptRuleSet"
        | "ListReceiptRuleSets"
        | "CloneReceiptRuleSet"
        | "SetActiveReceiptRuleSet"
        | "ReorderReceiptRuleSet"
        | "CreateReceiptRule"
        | "DeleteReceiptRule"
        | "DescribeReceiptRule"
        | "UpdateReceiptRule"
        | "CreateReceiptFilter"
        | "DeleteReceiptFilter"
        | "ListReceiptFilters" => Some("ses".to_string()),
        // SNS subscription handshake: the SubscribeURL / UnsubscribeUrl that SNS
        // hands to HTTP/S and email subscribers are unsigned bare GETs (no auth
        // header), so the service must be inferred from the action alone.
        "ConfirmSubscription" | "Unsubscribe" => Some("sns".to_string()),
        // Any other Query operation name exactly one modeled Query service
        // has (generated from aws-models).
        _ => query_actions::QUERY_ACTION_SERVICES
            .binary_search_by(|(a, _)| (*a).cmp(action))
            .ok()
            .map(|i| query_actions::QUERY_ACTION_SERVICES[i].1.to_string()),
    }
}

#[path = "query_actions_gen.rs"]
mod query_actions;

/// Extract service name from the SigV4 Authorization header credential scope.
pub(crate) fn extract_service_from_auth(headers: &HeaderMap) -> Option<String> {
    let auth = headers.get("authorization")?.to_str().ok()?;
    let info = fakecloud_aws::sigv4::parse_sigv4(auth)?;
    Some(normalize_service_name(&info.service).to_string())
}

/// Map AWS service-name aliases that share path namespace and handlers
/// to the canonical form used by fakecloud's service registry.
///
/// AWS uses `bedrock-runtime` in the SigV4 credential scope of runtime
/// API calls (`InvokeModel`, `ApplyGuardrail`, etc.) but the REST paths
/// (e.g. `POST /guardrail/{id}/version/{ver}/apply`) live under the same
/// `BedrockService` handler that owns the control-plane `bedrock` paths.
/// Without normalization, `detect_service` returns `None` for
/// `bedrock-runtime` (not in `REST_JSON_SERVICES`), the central
/// dispatcher falls back to API Gateway, and `/guardrail/...` 404s with
/// `NotFoundException: Stage not found: guardrail`. See issue #1232.
fn normalize_service_name(service: &str) -> &str {
    match service {
        "bedrock-runtime" => "bedrock",
        // Real AWS API Gateway V2 SDK signs with `apigateway` as the SigV4
        // service (per the model's `aws.api#service.endpointPrefix`), but
        // tools driven by the Smithy service shape name (including our own
        // conformance probe) may send `apigatewayv2`. Both refer to the
        // same fakecloud service registry entry — the v2 handler is path-
        // routed under `/v2/...` and the v1 handler under `/restapis/...`,
        // both reachable behind the `apigateway` SigV4 service.
        "apigatewayv2" => "apigateway",
        // Amazon OpenSearch Service has no dedicated SigV4 signing scope: its
        // SDK signs with `es` (the shared Elasticsearch Service scope), so a
        // real OpenSearch request already arrives as `es` and needs no
        // normalization. The conformance probe, however, signs with the
        // Smithy service shape name `opensearch`; alias it to `es` so both the
        // 2015 (Elasticsearch) and 2021 (OpenSearch) probes resolve to the one
        // registry entry, which then routes on the URL path version prefix.
        "opensearch" => "es",
        // AWS AppConfig Data signs with the shared `appconfig` scope, so a real
        // request already arrives as `appconfig`. The conformance probe signs
        // the data-plane operations with the Smithy service shape name
        // `appconfigdata`; alias it to `appconfig` so both model-services
        // resolve to the one registry entry, which routes on the URL path.
        "appconfigdata" => "appconfig",
        // Amazon MWAA signs SigV4 with the `airflow` scope (its ARN namespace),
        // so a real SDK request arrives as `airflow`. Alias it to the `mwaa`
        // registry entry (the conformance probe signs with the Smithy service
        // shape name `mwaa`, which already resolves).
        "airflow" => "mwaa",
        // Amazon Pinpoint signs SigV4 with the `mobiletargeting` scope (its ARN
        // namespace), so a real SDK request arrives as `mobiletargeting`. Alias
        // it to the `pinpoint` registry entry (the conformance probe signs with
        // the service-map `service_name`, `pinpoint`, which already resolves).
        "mobiletargeting" => "pinpoint",
        other => other,
    }
}

/// Parse form-encoded body into key-value pairs.
pub fn parse_query_body(body: &Bytes) -> HashMap<String, String> {
    decode_form_urlencoded(body)
}

/// Flatten an awsJson request body into the flat `awsQuery` key form that
/// query-protocol handlers consume.
///
/// CloudWatch is served by handlers written against the awsQuery flat-key map
/// (`MetricData.member.1.MetricName`, `StatisticValues.Sum`,
/// `Dimensions.member.2.Value`, ...). Its Smithy model also advertises
/// `awsJson1_0`, so modern SDKs send a nested JSON body instead. Rather than
/// duplicate every parser, we flatten the JSON into the same map the awsQuery
/// handlers already read:
///
/// - object field `K` -> key `K` (or `<parent>.K` when nested in a struct)
/// - array element `i` (1-based) -> `<K>.member.<i>` (matching the awsQuery
///   list wire convention)
/// - scalars -> their string form (numbers/booleans stringified)
///
/// A body that is not a JSON object yields an empty map.
pub fn flatten_json_to_query(body: &Bytes) -> HashMap<String, String> {
    let mut out = HashMap::new();
    let Ok(value) = serde_json::from_slice::<serde_json::Value>(body) else {
        return out;
    };
    if value.is_object() {
        flatten_json_value("", &value, &mut out);
    }
    out
}

fn flatten_json_value(prefix: &str, value: &serde_json::Value, out: &mut HashMap<String, String>) {
    match value {
        serde_json::Value::Object(map) => {
            for (k, v) in map {
                let child = if prefix.is_empty() {
                    k.clone()
                } else {
                    format!("{prefix}.{k}")
                };
                flatten_json_value(&child, v, out);
            }
        }
        serde_json::Value::Array(items) => {
            for (i, v) in items.iter().enumerate() {
                let child = format!("{prefix}.member.{}", i + 1);
                flatten_json_value(&child, v, out);
            }
        }
        serde_json::Value::Null => {}
        serde_json::Value::String(s) => {
            out.insert(prefix.to_string(), s.clone());
        }
        serde_json::Value::Bool(b) => {
            out.insert(prefix.to_string(), b.to_string());
        }
        serde_json::Value::Number(n) => {
            out.insert(prefix.to_string(), n.to_string());
        }
    }
}

/// Decode a query / form-urlencoded string into ordered `(key, value)` pairs,
/// **preserving repeated keys**. The [`HashMap`] form
/// ([`decode_form_urlencoded`]) collapses `a=1&a=2` to a single entry, which
/// loses multi-value `@httpQuery` params; this variant keeps every occurrence
/// in wire order for callers that need them (e.g. list-style query params).
pub(crate) fn form_urlencoded_pairs(input: &str) -> Vec<(String, String)> {
    let mut pairs = Vec::new();
    for pair in input.split('&') {
        if pair.is_empty() {
            continue;
        }
        let (key, value) = match pair.find('=') {
            Some(pos) => (&pair[..pos], &pair[pos + 1..]),
            None => (pair, ""),
        };
        pairs.push((url_decode(key), url_decode(value)));
    }
    pairs
}

fn decode_form_urlencoded(input: &[u8]) -> HashMap<String, String> {
    let s = std::str::from_utf8(input).unwrap_or("");
    let mut result = HashMap::new();
    for pair in s.split('&') {
        if pair.is_empty() {
            continue;
        }
        let (key, value) = match pair.find('=') {
            Some(pos) => (&pair[..pos], &pair[pos + 1..]),
            None => (pair, ""),
        };
        result.insert(url_decode(key), url_decode(value));
    }
    result
}

/// Decode one `application/x-www-form-urlencoded` component (a query-string
/// key or value, or a form-body field): `+` is a space and `%XX` escapes are
/// decoded on the raw bytes, then read as UTF-8 (lossily). For URI path
/// segments, where `+` is literal, use [`crate::path::percent_decode_segment`].
pub fn url_decode(input: &str) -> String {
    // Accumulate the decoded RAW BYTES first, then interpret the whole buffer
    // as UTF-8. Decoding each `%XX` byte straight into a `char` would treat it
    // as a Unicode codepoint (Latin-1), which corrupts multi-byte UTF-8
    // sequences (e.g. "caf%C3%A9" -> "café" instead of "café"). Reassembling
    // the bytes lets multi-byte sequences round-trip correctly.
    let mut buf: Vec<u8> = Vec::with_capacity(input.len());
    let mut bytes = input.bytes();
    while let Some(b) = bytes.next() {
        match b {
            b'+' => buf.push(b' '),
            b'%' => {
                let high = bytes.next().and_then(from_hex);
                let low = bytes.next().and_then(from_hex);
                // A well-formed `%XX` escape decodes to a single raw byte.
                // A malformed escape is dropped (best-effort, panic-free),
                // matching the prior behaviour.
                if let (Some(h), Some(l)) = (high, low) {
                    buf.push((h << 4) | l);
                }
            }
            _ => buf.push(b),
        }
    }
    String::from_utf8_lossy(&buf).into_owned()
}

fn from_hex(b: u8) -> Option<u8> {
    match b {
        b'0'..=b'9' => Some(b - b'0'),
        b'a'..=b'f' => Some(b - b'a' + 10),
        b'A'..=b'F' => Some(b - b'A' + 10),
        _ => None,
    }
}

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

    #[test]
    fn headers_settle_the_service_only_when_no_form_body_can_change_it() {
        // An unsigned POST whose Host says S3 but whose form body names an
        // STS action is routed to STS: header detection alone is not final.
        let mut plain_host = HeaderMap::new();
        plain_host.insert("host", "b.s3.us-east-1.amazonaws.com".parse().unwrap());
        let mut host_only = plain_host.clone();
        host_only.insert(
            "content-type",
            "application/x-www-form-urlencoded; charset=utf-8"
                .parse()
                .unwrap(),
        );
        let none = HashMap::new();
        let body = Bytes::from("Action=GetCallerIdentity&Version=2011-06-15");
        assert_eq!(
            detect_service_headers_only(&host_only, &none).map(|d| d.service),
            Some("s3".to_string())
        );
        assert_eq!(
            detect_service(&host_only, &none, &body).map(|d| d.service),
            Some("sts".to_string())
        );
        assert!(!headers_name_final_service(&host_only, &none));
        // Without a form Content-Type the body is no Query call (real AWS does
        // not treat it as one), so the Host settles it.
        assert_eq!(
            detect_service(&plain_host, &none, &body).map(|d| d.service),
            Some("s3".to_string())
        );
        assert!(headers_name_final_service(&plain_host, &none));
        let mut text = plain_host.clone();
        text.insert("content-type", "text/plain".parse().unwrap());
        assert!(headers_name_final_service(&text, &none));
        // Likewise a presigned URL: its scope is not the Authorization header.
        let presigned: HashMap<String, String> = [(
            "X-Amz-Credential".to_string(),
            "AKID/20240101/us-east-1/s3/aws4_request".to_string(),
        )]
        .into();
        let mut form = HeaderMap::new();
        form.insert(
            "content-type",
            "application/x-www-form-urlencoded".parse().unwrap(),
        );
        assert!(!headers_name_final_service(&form, &presigned));

        // A SigV4 scope, a target or a resolvable query Action settles it.
        let mut signed = host_only.clone();
        signed.insert(
            "authorization",
            "AWS4-HMAC-SHA256 Credential=AKID/20240101/us-east-1/s3/aws4_request, \
             SignedHeaders=host, Signature=00"
                .parse()
                .unwrap(),
        );
        assert!(headers_name_final_service(&signed, &none));
        assert_eq!(
            detect_service(&signed, &none, &body).map(|d| d.service),
            Some("s3".to_string())
        );
        let mut target = HeaderMap::new();
        target.insert(
            "x-amz-target",
            "DynamoDB_20120810.ListTables".parse().unwrap(),
        );
        assert!(headers_name_final_service(&target, &none));
        let action: HashMap<String, String> =
            [("Action".to_string(), "GetCallerIdentity".to_string())].into();
        assert!(headers_name_final_service(&HeaderMap::new(), &action));
    }

    #[test]
    fn s3_control_host_requires_leading_control_label() {
        let is_control = |host: &str| {
            let mut headers = HeaderMap::new();
            headers.insert("host", host.parse().unwrap());
            is_s3_control_host(&headers)
        };
        assert!(is_control("s3-control.us-east-1.amazonaws.com"));
        assert!(is_control(
            "123456789012.s3-control.us-east-1.amazonaws.com"
        ));
        assert!(is_control(
            "000000000000.s3-control.us-east-1.localhost.localstack.cloud:4566"
        ));
        assert!(is_control("123456789012.s3-control.localhost"));
        assert!(!is_control("my-s3-control-logs.s3.us-east-1.amazonaws.com"));
        assert!(!is_control(
            "my-s3-control-logs.s3.us-east-1.localhost.localstack.cloud:4566"
        ));
        assert!(!is_control("a.s3-control.b.s3.us-east-1.amazonaws.com"));
        assert!(!is_control("localhost:4566"));
    }

    #[test]
    fn form_urlencoded_pairs_preserves_repeated_keys() {
        // The HashMap decoder collapses repeats; the ordered-pairs variant must
        // keep every occurrence in wire order (1.45 multi-value @httpQuery).
        let pairs = form_urlencoded_pairs("Id=a&Id=b&Id=c&Other=x");
        let ids: Vec<&str> = pairs
            .iter()
            .filter(|(k, _)| k == "Id")
            .map(|(_, v)| v.as_str())
            .collect();
        assert_eq!(ids, vec!["a", "b", "c"]);
        // Sanity: the HashMap form keeps only the last value.
        assert_eq!(
            decode_form_urlencoded(b"Id=a&Id=b&Id=c").get("Id").unwrap(),
            "c"
        );
    }

    #[test]
    fn form_urlencoded_pairs_decodes_percent_and_plus() {
        let pairs = form_urlencoded_pairs("q=caf%C3%A9+bar&empty=");
        assert_eq!(pairs[0], ("q".to_string(), "café bar".to_string()));
        assert_eq!(pairs[1], ("empty".to_string(), String::new()));
    }

    #[test]
    fn parse_amz_target_events() {
        let result = parse_amz_target("AWSEvents.PutEvents").unwrap();
        assert_eq!(result.service, "events");
        assert_eq!(result.action, "PutEvents");
        assert_eq!(result.protocol, AwsProtocol::Json);
    }

    #[test]
    fn parse_amz_target_ssm() {
        let result = parse_amz_target("AmazonSSM.GetParameter").unwrap();
        assert_eq!(result.service, "ssm");
        assert_eq!(result.action, "GetParameter");
    }

    #[test]
    fn parse_amz_target_kinesis() {
        let result = parse_amz_target("Kinesis_20131202.ListStreams").unwrap();
        assert_eq!(result.service, "kinesis");
        assert_eq!(result.action, "ListStreams");
        assert_eq!(result.protocol, AwsProtocol::Json);
    }

    #[test]
    fn parse_query_body_basic() {
        let body = Bytes::from(
            "Action=SendMessage&QueueUrl=http%3A%2F%2Flocalhost%3A4566%2Fqueue&MessageBody=hello",
        );
        let params = parse_query_body(&body);
        assert_eq!(params.get("Action").unwrap(), "SendMessage");
        assert_eq!(params.get("MessageBody").unwrap(), "hello");
    }

    #[test]
    fn parse_query_body_empty_returns_empty_map() {
        let body = Bytes::from("");
        let params = parse_query_body(&body);
        assert!(params.is_empty());
    }

    #[test]
    fn parse_query_body_duplicate_keys_last_wins() {
        let body = Bytes::from("key=a&key=b");
        let params = parse_query_body(&body);
        assert_eq!(params.get("key").unwrap(), "b");
    }

    #[test]
    fn parse_query_body_single_key() {
        let body = Bytes::from("key=value");
        let params = parse_query_body(&body);
        assert_eq!(params.get("key").unwrap(), "value");
    }

    #[test]
    fn url_decode_plain_ascii() {
        assert_eq!(url_decode("hello"), "hello");
        assert_eq!(url_decode("Action=SendMessage"), "Action=SendMessage");
    }

    #[test]
    fn url_decode_plus_is_space() {
        assert_eq!(url_decode("hello+world"), "hello world");
        assert_eq!(url_decode("a+b+c"), "a b c");
    }

    #[test]
    fn url_decode_multibyte_utf8_accents() {
        // "café" -> the é is UTF-8 0xC3 0xA9, two %-escapes for one codepoint.
        assert_eq!(url_decode("caf%C3%A9"), "café");
    }

    #[test]
    fn url_decode_multibyte_utf8_cjk() {
        // "日本" (each char is 3 UTF-8 bytes).
        assert_eq!(url_decode("%E6%97%A5%E6%9C%AC"), "日本");
    }

    #[test]
    fn url_decode_multibyte_utf8_emoji() {
        // "🚀" is a 4-byte UTF-8 sequence (F0 9F 9A 80).
        assert_eq!(url_decode("%F0%9F%9A%80"), "🚀");
    }

    #[test]
    fn url_decode_mixed_ascii_and_multibyte() {
        assert_eq!(url_decode("Tag+%3D+caf%C3%A9%21"), "Tag = café!");
    }

    #[test]
    fn url_decode_malformed_percent_is_graceful() {
        // A malformed escape is dropped best-effort and must never panic.
        assert_eq!(url_decode("100%"), "100");
        assert_eq!(url_decode("a%zz"), "a");
        assert_eq!(url_decode("a%4"), "a");
        // Preceding and following ASCII are preserved either side of the drop.
        assert_eq!(url_decode("x%y"), "x");
    }

    #[test]
    fn url_decode_invalid_utf8_bytes_are_lossy_no_panic() {
        // 0xFF is not valid UTF-8; must not panic, replaced lossily.
        let out = url_decode("bad%FFbyte");
        assert!(out.starts_with("bad"));
        assert!(out.ends_with("byte"));
    }

    #[test]
    fn parse_query_body_multibyte_value_round_trips() {
        let body = Bytes::from("Tag.Value=caf%C3%A9&Name=%E6%97%A5%E6%9C%AC");
        let params = parse_query_body(&body);
        assert_eq!(params.get("Tag.Value").unwrap(), "café");
        assert_eq!(params.get("Name").unwrap(), "日本");
    }

    #[test]
    fn parse_amz_target_ecs() {
        let result = parse_amz_target("AmazonEC2ContainerServiceV20141113.ListClusters").unwrap();
        assert_eq!(result.service, "ecs");
        assert_eq!(result.action, "ListClusters");
        assert_eq!(result.protocol, AwsProtocol::Json);
    }

    #[test]
    fn parse_amz_target_invalid_returns_none() {
        assert!(parse_amz_target("NoDotHere").is_none());
        assert!(parse_amz_target("").is_none());
    }

    #[test]
    fn parse_amz_target_cloudwatch_json() {
        // CloudWatch's awsJson1_0 target service shape.
        let result = parse_amz_target("GraniteServiceVersion20100801.PutMetricData").unwrap();
        assert_eq!(result.service, "monitoring");
        assert_eq!(result.action, "PutMetricData");
        assert_eq!(result.protocol, AwsProtocol::Json);
    }

    #[test]
    fn flatten_json_to_query_nested() {
        let body = Bytes::from(
            serde_json::json!({
                "Namespace": "MyApp",
                "MetricData": [{
                    "MetricName": "Latency",
                    "Value": 12.5,
                    "StatisticValues": {"SampleCount": 3, "Sum": 10},
                    "Dimensions": [{"Name": "Endpoint", "Value": "/api"}]
                }]
            })
            .to_string(),
        );
        let flat = flatten_json_to_query(&body);
        assert_eq!(flat.get("Namespace").unwrap(), "MyApp");
        assert_eq!(
            flat.get("MetricData.member.1.MetricName").unwrap(),
            "Latency"
        );
        assert_eq!(flat.get("MetricData.member.1.Value").unwrap(), "12.5");
        assert_eq!(
            flat.get("MetricData.member.1.StatisticValues.SampleCount")
                .unwrap(),
            "3"
        );
        assert_eq!(
            flat.get("MetricData.member.1.Dimensions.member.1.Name")
                .unwrap(),
            "Endpoint"
        );
        assert_eq!(
            flat.get("MetricData.member.1.Dimensions.member.1.Value")
                .unwrap(),
            "/api"
        );
    }

    #[test]
    fn flatten_json_to_query_non_object_is_empty() {
        assert!(flatten_json_to_query(&Bytes::from_static(b"[]")).is_empty());
        assert!(flatten_json_to_query(&Bytes::from_static(b"not json")).is_empty());
    }

    #[test]
    fn parse_amz_target_various_prefixes() {
        assert_eq!(
            parse_amz_target("AmazonSQS.SendMessage").unwrap().service,
            "sqs"
        );
        assert_eq!(
            parse_amz_target("AmazonSNS.Publish").unwrap().service,
            "sns"
        );
        assert_eq!(
            parse_amz_target("DynamoDB_20120810.GetItem")
                .unwrap()
                .service,
            "dynamodb"
        );
        assert_eq!(
            parse_amz_target("Logs_20140328.PutLogEvents")
                .unwrap()
                .service,
            "logs"
        );
        assert_eq!(
            parse_amz_target("secretsmanager.GetSecretValue")
                .unwrap()
                .service,
            "secretsmanager"
        );
        assert_eq!(
            parse_amz_target("TrentService.Encrypt").unwrap().service,
            "kms"
        );
        assert_eq!(
            parse_amz_target("AWSCognitoIdentityProviderService.InitiateAuth")
                .unwrap()
                .service,
            "cognito-idp"
        );
        assert_eq!(
            parse_amz_target("AWSStepFunctions.StartExecution")
                .unwrap()
                .service,
            "states"
        );
        assert_eq!(
            parse_amz_target("AWSOrganizationsV20161128.CreateOrganization")
                .unwrap()
                .service,
            "organizations"
        );
        assert!(parse_amz_target("UnknownServicePrefix.Action").is_none());
    }

    #[test]
    fn infer_service_from_action_maps_sts() {
        assert_eq!(
            infer_service_from_action("AssumeRole").as_deref(),
            Some("sts")
        );
        assert_eq!(
            infer_service_from_action("GetCallerIdentity").as_deref(),
            Some("sts")
        );
        // Any Query operation name one modeled service has routes too.
        for (action, service) in [
            ("ListTopics", "sns"),
            ("ListQueues", "sqs"),
            ("DescribeRegions", "ec2"),
            ("DescribeStacks", "cloudformation"),
            ("DescribeLoadBalancers", "elasticloadbalancing"),
        ] {
            assert_eq!(
                infer_service_from_action(action).as_deref(),
                Some(service),
                "{action}"
            );
        }
        // A name several services share does not.
        assert_eq!(infer_service_from_action("ListTagsForResource"), None);
        // Every STS operation routes an unsigned form POST on the plain host.
        for action in [
            "AssumeRoot",
            "GetWebIdentityToken",
            "GetDelegatedAccessToken",
        ] {
            assert_eq!(infer_service_from_action(action).as_deref(), Some("sts"));
        }
    }

    #[test]
    fn infer_service_from_action_maps_iam() {
        assert_eq!(
            infer_service_from_action("CreateUser").as_deref(),
            Some("iam")
        );
        assert_eq!(
            infer_service_from_action("ListRoles").as_deref(),
            Some("iam")
        );
    }

    #[test]
    fn infer_service_from_action_maps_ses() {
        assert_eq!(
            infer_service_from_action("SendEmail").as_deref(),
            Some("ses")
        );
        assert_eq!(
            infer_service_from_action("ListIdentities").as_deref(),
            Some("ses")
        );
    }

    #[test]
    fn infer_service_from_action_maps_sns_confirmation_flow() {
        // SNS hands subscribers unsigned SubscribeURL / UnsubscribeUrl GETs,
        // so the service must be inferred from the action alone.
        assert_eq!(
            infer_service_from_action("ConfirmSubscription").as_deref(),
            Some("sns")
        );
        assert_eq!(
            infer_service_from_action("Unsubscribe").as_deref(),
            Some("sns")
        );
    }

    #[test]
    fn detect_service_routes_unsigned_confirm_subscription_to_sns() {
        // Mirror the bare GET an HTTP/S subscriber issues at the SubscribeURL:
        // no Authorization header, bare-localhost Host, Action in the query.
        let mut headers = HeaderMap::new();
        headers.insert("host", "localhost:4566".parse().unwrap());
        let mut query_params = HashMap::new();
        query_params.insert("Action".to_string(), "ConfirmSubscription".to_string());
        query_params.insert(
            "TopicArn".to_string(),
            "arn:aws:sns:us-east-1:000000000000:t".to_string(),
        );
        query_params.insert("Token".to_string(), "abc123".to_string());

        let detected = detect_service(&headers, &query_params, &Bytes::new())
            .expect("ConfirmSubscription must route to a service");
        assert_eq!(detected.service, "sns");
        assert_eq!(detected.action, "ConfirmSubscription");
        assert_eq!(detected.protocol, AwsProtocol::Query);
    }

    #[test]
    fn infer_service_from_action_unknown_returns_none() {
        assert!(infer_service_from_action("NotARealAction").is_none());
    }

    #[test]
    fn rest_protocol_for_returns_none_for_non_rest_service() {
        assert!(rest_protocol_for("sqs").is_none());
    }

    #[test]
    fn url_decode_handles_percent_and_plus() {
        assert_eq!(url_decode("hello+world"), "hello world");
        assert_eq!(url_decode("hello%20world"), "hello world");
        assert_eq!(url_decode("100%25"), "100%");
    }

    #[test]
    fn url_decode_ignores_malformed_percent() {
        assert_eq!(url_decode("%ZZ"), "");
    }

    #[test]
    fn from_hex_valid_digits() {
        assert_eq!(from_hex(b'0'), Some(0));
        assert_eq!(from_hex(b'9'), Some(9));
        assert_eq!(from_hex(b'a'), Some(10));
        assert_eq!(from_hex(b'F'), Some(15));
    }

    #[test]
    fn from_hex_invalid_returns_none() {
        assert!(from_hex(b'g').is_none());
        assert!(from_hex(b' ').is_none());
    }

    #[test]
    fn detect_service_via_amz_target() {
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-target", "AmazonSSM.GetParameter".parse().unwrap());
        let query = HashMap::new();
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "ssm");
        assert_eq!(detected.action, "GetParameter");
    }

    #[test]
    fn detect_service_via_query_action_with_inferred_service() {
        let headers = HeaderMap::new();
        let mut query = HashMap::new();
        query.insert("Action".to_string(), "AssumeRole".to_string());
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "sts");
        assert_eq!(detected.action, "AssumeRole");
        assert_eq!(detected.protocol, AwsProtocol::Query);
    }

    #[test]
    fn detect_service_via_form_body() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "content-type",
            "Application/X-WWW-Form-Urlencoded ; charset=UTF-8"
                .parse()
                .unwrap(),
        );
        let query = HashMap::new();
        let body = Bytes::from("Action=SendEmail&Source=x%40y.com");
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "ses");
        assert_eq!(detected.action, "SendEmail");
        // The same body declared as anything else, or untyped, is no Query
        // call.
        for content_type in [None, Some("text/plain"), Some("application/json")] {
            let mut headers = HeaderMap::new();
            if let Some(ct) = content_type {
                headers.insert("content-type", ct.parse().unwrap());
            }
            assert!(
                detect_service(&headers, &query, &body).is_none(),
                "{content_type:?}"
            );
        }
    }

    #[test]
    fn detect_service_via_sigv2_presigned() {
        let headers = HeaderMap::new();
        let mut query = HashMap::new();
        query.insert("AWSAccessKeyId".to_string(), "AKID".to_string());
        query.insert("Signature".to_string(), "sig".to_string());
        query.insert("Expires".to_string(), "1234567890".to_string());
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "s3");
        assert_eq!(detected.protocol, AwsProtocol::Rest);
    }

    #[test]
    fn detect_service_via_sigv4_presigned_credential() {
        let headers = HeaderMap::new();
        let mut query = HashMap::new();
        query.insert(
            "X-Amz-Credential".to_string(),
            "AKID/20240101/us-east-1/s3/aws4_request".to_string(),
        );
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "s3");
        assert_eq!(detected.protocol, AwsProtocol::Rest);
    }

    #[test]
    fn detect_service_unknown_returns_none() {
        let headers = HeaderMap::new();
        let query = HashMap::new();
        let body = Bytes::new();
        assert!(detect_service(&headers, &query, &body).is_none());
    }

    #[test]
    fn normalize_service_name_aliases_apigatewayv2_to_apigateway() {
        // Real AWS API Gateway V2 SDK signs with `apigateway` per the
        // model's `endpointPrefix`, but Smithy-driven tooling (including
        // our conformance probe) sends `apigatewayv2`. Both routes resolve
        // to the same fakecloud service registry entry.
        assert_eq!(normalize_service_name("apigatewayv2"), "apigateway");
    }

    #[test]
    fn normalize_service_name_aliases_bedrock_runtime_to_bedrock() {
        // The bedrock-runtime credential scope shares path namespace with
        // the bedrock control plane (`POST /guardrail/{id}/version/{ver}/apply`
        // is implemented under BedrockService). Routing must resolve to
        // the bedrock service so the existing handlers run. See #1232.
        assert_eq!(normalize_service_name("bedrock-runtime"), "bedrock");
    }

    #[test]
    fn normalize_service_name_passes_through_unaliased_services() {
        // Every service that isn't on the alias list must round-trip
        // unchanged — including the canonical bedrock name itself, so a
        // plain bedrock request takes the same code path it always has.
        assert_eq!(normalize_service_name("bedrock"), "bedrock");
        assert_eq!(normalize_service_name("s3"), "s3");
        assert_eq!(normalize_service_name("lambda"), "lambda");
        assert_eq!(normalize_service_name(""), "");
        assert_eq!(
            normalize_service_name("unknown-future-service"),
            "unknown-future-service"
        );
    }

    #[test]
    fn detect_service_via_authorization_header_normalizes_bedrock_runtime() {
        // SigV4 auth header carries `bedrock-runtime` in the credential
        // scope; dispatcher must route to the bedrock service handler so
        // `/guardrail/...` lands on `BedrockService` instead of falling
        // through to API Gateway.
        let mut headers = HeaderMap::new();
        headers.insert(
            "authorization",
            "AWS4-HMAC-SHA256 \
             Credential=AKID/20240101/us-east-1/bedrock-runtime/aws4_request, \
             SignedHeaders=host, Signature=abc"
                .parse()
                .unwrap(),
        );
        let query = HashMap::new();
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "bedrock");
        assert_eq!(detected.protocol, AwsProtocol::RestJson);
    }

    #[test]
    fn detect_service_via_sigv4_presigned_credential_normalizes_bedrock_runtime() {
        // Same alias normalization on the presigned-URL path: a request
        // signed with bedrock-runtime in the X-Amz-Credential query param
        // must still resolve to the bedrock service handler.
        let headers = HeaderMap::new();
        let mut query = HashMap::new();
        query.insert(
            "X-Amz-Credential".to_string(),
            "AKID/20240101/us-east-1/bedrock-runtime/aws4_request".to_string(),
        );
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "bedrock");
        assert_eq!(detected.protocol, AwsProtocol::RestJson);
    }

    #[test]
    fn normalize_host_strips_port_and_case() {
        assert_eq!(normalize_host("Rest.Example.com:4566"), "rest.example.com");
        assert_eq!(normalize_host("rest.example.com."), "rest.example.com");
        assert_eq!(normalize_host("rest.example.com"), "rest.example.com");
        assert_eq!(normalize_host("[::1]:4566"), "[::1]");
        assert_eq!(normalize_host("localhost"), "localhost");
    }

    #[test]
    fn parse_routing_host_execute_api_hosts() {
        // LocalStack's execute-api host has no region label: it must not be
        // read as region `execute-api`.
        assert_eq!(
            parse_routing_host("abc123.execute-api.localhost.localstack.cloud:4566"),
            None
        );
        let h =
            parse_routing_host("abc123.execute-api.eu-west-1.localhost.localstack.cloud").unwrap();
        assert_eq!(h.service, "execute-api");
        assert_eq!(h.region, "eu-west-1");
        let h = parse_routing_host("abc123.execute-api.us-west-2.amazonaws.com").unwrap();
        assert_eq!(h.service, "execute-api");
        assert_eq!(h.region, "us-west-2");
        assert_eq!(h.bucket, None);
    }

    #[test]
    fn detect_rpc_v2_cbor_requests() {
        let mut headers = HeaderMap::new();
        let path = "/service/GraniteServiceVersion20100801/operation/GetMetricData";
        // Without the protocol header the path alone is not enough.
        assert!(detect_rpc_v2_cbor(&headers, path).is_none());
        headers.insert("smithy-protocol", "rpc-v2-cbor".parse().unwrap());
        let d = detect_rpc_v2_cbor(&headers, path).unwrap();
        assert_eq!(d.service, "monitoring");
        assert_eq!(d.action, "GetMetricData");
        assert_eq!(d.protocol, AwsProtocol::RpcV2Cbor);
        assert!(detect_rpc_v2_cbor(&headers, "/service/Unknown/operation/Op").is_none());
        assert!(detect_rpc_v2_cbor(&headers, "/prod/items").is_none());
    }

    #[test]
    fn parse_routing_host_localstack_basic() {
        let h = parse_routing_host("sqs.us-east-1.localhost.localstack.cloud").unwrap();
        assert_eq!(h.service, "sqs");
        assert_eq!(h.region, "us-east-1");
        assert!(h.bucket.is_none());
    }

    #[test]
    fn parse_routing_host_localstack_with_port() {
        let h = parse_routing_host("lambda.eu-west-1.localhost.localstack.cloud:4566").unwrap();
        assert_eq!(h.service, "lambda");
        assert_eq!(h.region, "eu-west-1");
        assert!(h.bucket.is_none());
    }

    #[test]
    fn parse_routing_host_case_insensitive() {
        let h = parse_routing_host("SQS.US-EAST-1.LOCALHOST.LOCALSTACK.CLOUD:4566").unwrap();
        assert_eq!(h.service, "sqs");
        assert_eq!(h.region, "us-east-1");

        let h = parse_routing_host("LAMBDA.US-EAST-1.AMAZONAWS.COM").unwrap();
        assert_eq!(h.service, "lambda");
        assert_eq!(h.region, "us-east-1");
    }

    #[test]
    fn parse_routing_host_localstack_s3_virtual_hosted() {
        let h =
            parse_routing_host("my-bucket.s3.us-east-1.localhost.localstack.cloud:4566").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert_eq!(h.bucket.as_deref(), Some("my-bucket"));
    }

    #[test]
    fn parse_routing_host_localstack_s3_vhost_bucket_with_dots() {
        let h = parse_routing_host("a.b.c.s3.us-east-1.localhost.localstack.cloud").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert_eq!(h.bucket.as_deref(), Some("a.b.c"));
    }

    #[test]
    fn parse_routing_host_aws_service_region() {
        let h = parse_routing_host("sqs.us-east-1.amazonaws.com").unwrap();
        assert_eq!(h.service, "sqs");
        assert_eq!(h.region, "us-east-1");
        assert!(h.bucket.is_none());

        let h = parse_routing_host("dynamodb.eu-west-2.amazonaws.com:443").unwrap();
        assert_eq!(h.service, "dynamodb");
        assert_eq!(h.region, "eu-west-2");
    }

    #[test]
    fn parse_routing_host_aws_s3_path_style_modern() {
        let h = parse_routing_host("s3.us-east-1.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert!(h.bucket.is_none());
    }

    #[test]
    fn parse_routing_host_aws_s3_virtual_hosted_modern() {
        let h = parse_routing_host("my-bucket.s3.us-east-1.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert_eq!(h.bucket.as_deref(), Some("my-bucket"));
    }

    #[test]
    fn parse_routing_host_aws_s3_vhost_bucket_with_dots() {
        let h = parse_routing_host("a.b.c.s3.us-east-1.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert_eq!(h.bucket.as_deref(), Some("a.b.c"));
    }

    #[test]
    fn parse_routing_host_aws_s3_legacy_global() {
        // `s3.amazonaws.com` (no region) is the legacy S3 global endpoint —
        // AWS treats it as us-east-1 for both path-style and virtual-hosted.
        let h = parse_routing_host("s3.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert!(h.bucket.is_none());

        let h = parse_routing_host("my-bucket.s3.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert_eq!(h.bucket.as_deref(), Some("my-bucket"));
    }

    #[test]
    fn parse_routing_host_aws_s3_legacy_global_dotted_bucket() {
        // AWS allows buckets with dots (e.g. `a.b.c`) and still serves them
        // via the legacy `<bucket>.s3.amazonaws.com` global endpoint.
        let h = parse_routing_host("a.b.c.s3.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-east-1");
        assert_eq!(h.bucket.as_deref(), Some("a.b.c"));
    }

    #[test]
    fn query_string_action_is_a_parameter_only_of_s3_and_api_gateway() {
        let detect = |host: Option<&str>, auth: Option<&str>, query: &[(&str, &str)]| {
            let mut headers = HeaderMap::new();
            if let Some(host) = host {
                headers.insert(http::header::HOST, host.parse().unwrap());
            }
            if let Some(auth) = auth {
                headers.insert(http::header::AUTHORIZATION, auth.parse().unwrap());
            }
            let params: HashMap<String, String> = query
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect();
            let full =
                detect_service(&headers, &params, &Bytes::new()).map(|d| (d.service, d.action));
            let head =
                detect_service_headers_only(&headers, &params).map(|d| (d.service, d.action));
            assert_eq!(full, head, "{host:?} {query:?}");
            full
        };
        let s = |svc: &str, action: &str| Some((svc.to_string(), action.to_string()));
        // Nothing else names a service: the Action does (dispatch first
        // takes out a plain-host path naming an existing bucket).
        assert_eq!(
            detect(None, None, &[("Action", "ListQueues")]),
            s("sqs", "ListQueues")
        );
        // SES v1 is Query on SES's scope and host (v2 is restJson1 there).
        assert_eq!(
            detect(
                None,
                Some("AWS4-HMAC-SHA256 Credential=AKID/20261113/us-east-1/ses/aws4_request, SignedHeaders=host, Signature=x"),
                &[("Action", "ListIdentities")]
            ),
            s("ses", "ListIdentities")
        );
        assert_eq!(
            detect(
                Some("email.us-east-1.amazonaws.com"),
                None,
                &[("Action", "ListIdentities")]
            ),
            s("ses", "ListIdentities")
        );
        assert_eq!(
            detect(
                None,
                None,
                &[
                    ("Action", "ListIdentities"),
                    (
                        "X-Amz-Credential",
                        "AKID/20261113/us-east-1/ses/aws4_request"
                    ),
                ]
            ),
            s("ses", "ListIdentities")
        );
        // SigV2 Query authentication (SignatureVersion + Timestamp) is the
        // Action's service's; an S3 SigV2 presign (Expires) is S3's.
        for (action, service) in [("ListQueues", "sqs"), ("DescribeRegions", "ec2")] {
            assert_eq!(
                detect(
                    None,
                    None,
                    &[
                        ("Action", action),
                        ("AWSAccessKeyId", "AKID"),
                        ("Signature", "x"),
                        ("SignatureVersion", "2"),
                        ("SignatureMethod", "HmacSHA256"),
                        ("Timestamp", "2026-11-13T00:00:00Z"),
                    ]
                ),
                s(service, action)
            );
        }
        assert_eq!(
            detect(
                None,
                None,
                &[
                    ("Action", "ListQueues"),
                    ("AWSAccessKeyId", "AKID"),
                    ("Signature", "x"),
                    ("Expires", "1900000000"),
                ]
            ),
            s("s3", "")
        );
        // An S3 or API Gateway request keeps it: `Action` is a parameter.
        assert_eq!(
            detect(
                Some("pubb.s3.us-east-1.amazonaws.com"),
                None,
                &[("Action", "DescribeRegions")]
            ),
            s("s3", "")
        );
        assert_eq!(
            detect(
                Some("abc123.execute-api.us-east-1.amazonaws.com"),
                None,
                &[("Action", "Publish")]
            ),
            None,
            "an invocation: dispatch hands it to API Gateway"
        );
        assert_eq!(
            detect(
                None,
                None,
                &[
                    ("Action", "ListQueues"),
                    (
                        "X-Amz-Credential",
                        "AKID/20261113/us-east-1/s3/aws4_request"
                    ),
                ]
            ),
            s("s3", "")
        );
        assert_eq!(
            detect(
                None,
                Some("AWS4-HMAC-SHA256 Credential=AKID/20261113/us-east-1/s3/aws4_request, SignedHeaders=host, Signature=x"),
                &[("Action", "ListTopics")]
            ),
            s("s3", "")
        );
        // A Query service's request is its Query call.
        assert_eq!(
            detect(
                Some("sqs.us-east-1.amazonaws.com"),
                None,
                &[("Action", "ListQueues")]
            ),
            s("sqs", "ListQueues")
        );
        assert_eq!(
            detect(
                None,
                None,
                &[
                    ("Action", "GetCallerIdentity"),
                    (
                        "X-Amz-Credential",
                        "AKID/20261113/us-east-1/sts/aws4_request"
                    ),
                ]
            ),
            s("sts", "GetCallerIdentity")
        );
    }

    #[test]
    fn query_action_table_matches_the_models() {
        // Re-derives the table: rerun scripts/generate-query-action-services.py
        // after a model refresh.
        let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../aws-models");
        let mut owners: std::collections::BTreeMap<String, std::collections::BTreeSet<String>> =
            Default::default();
        for entry in std::fs::read_dir(&dir).unwrap() {
            let path = entry.unwrap().path();
            let text = std::fs::read_to_string(&path).unwrap();
            if !["awsQuery\"", "ec2Query\"", "awsQueryCompatible\""]
                .iter()
                .any(|p| text.contains(&format!("aws.protocols#{p}")))
            {
                continue;
            }
            let model: serde_json::Value = serde_json::from_str(&text).unwrap();
            let shapes = model["shapes"].as_object().unwrap();
            let service = shapes.values().find(|s| s["type"] == "service").unwrap();
            let traits = service["traits"].as_object().unwrap();
            if ![
                "aws.protocols#awsQuery",
                "aws.protocols#ec2Query",
                "aws.protocols#awsQueryCompatible",
            ]
            .iter()
            .any(|p| traits.contains_key(*p))
            {
                continue;
            }
            let signing = traits["aws.auth#sigv4"]["name"].as_str().unwrap();
            for (id, shape) in shapes {
                if shape["type"] == "operation" {
                    let op = id.split_once('#').unwrap().1;
                    owners
                        .entry(op.to_string())
                        .or_default()
                        .insert(signing.to_string());
                }
            }
        }
        let derived: Vec<(String, String)> = owners
            .into_iter()
            .filter(|(_, s)| s.len() == 1)
            .map(|(op, s)| (op, s.into_iter().next().unwrap()))
            .collect();
        let table: Vec<(String, String)> = query_actions::QUERY_ACTION_SERVICES
            .iter()
            .map(|(a, s)| (a.to_string(), s.to_string()))
            .collect();
        assert_eq!(
            table, derived,
            "rerun scripts/generate-query-action-services.py"
        );
    }

    #[test]
    fn parse_routing_host_email_is_ses() {
        for host in [
            "email.us-east-1.amazonaws.com",
            "email.eu-west-1.localhost.localstack.cloud:4566",
        ] {
            assert_eq!(parse_routing_host(host).unwrap().service, "ses", "{host}");
        }
    }

    #[test]
    fn parse_routing_host_global_endpoints() {
        for service in ["iam", "sts", "route53", "cloudfront"] {
            let h = parse_routing_host(&format!("{service}.amazonaws.com:443")).unwrap();
            assert_eq!(h.service, service);
            assert_eq!(h.region, "us-east-1");
            assert_eq!(h.bucket, None);
        }
        // Only those services have a region-less endpoint.
        assert!(parse_routing_host("lambda.amazonaws.com").is_none());
        assert!(parse_routing_host("route53.amazonaws.com.cn").is_none());
    }

    #[test]
    fn parse_routing_host_aws_s3_dash_separated() {
        // Older dash-separated form still served by AWS.
        let h = parse_routing_host("s3-us-west-2.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-west-2");
        assert!(h.bucket.is_none());

        let h = parse_routing_host("my-bucket.s3-us-west-2.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-west-2");
        assert_eq!(h.bucket.as_deref(), Some("my-bucket"));
    }

    #[test]
    fn parse_routing_host_aws_s3_dualstack_and_fips() {
        for (host, bucket) in [
            (
                "my-bucket.s3.dualstack.us-east-1.amazonaws.com",
                Some("my-bucket"),
            ),
            ("a.b.s3.dualstack.eu-west-2.amazonaws.com", Some("a.b")),
            ("s3.dualstack.eu-west-2.amazonaws.com", None),
            (
                "my-bucket.s3-fips.us-gov-west-1.amazonaws.com",
                Some("my-bucket"),
            ),
            (
                "my-bucket.s3-fips.dualstack.us-east-1.amazonaws.com",
                Some("my-bucket"),
            ),
        ] {
            let h = parse_routing_host(host).unwrap();
            assert_eq!(h.service, "s3", "{host}");
            assert_eq!(h.bucket.as_deref(), bucket, "{host}");
        }
        let h = parse_routing_host("b.s3.dualstack.eu-west-2.amazonaws.com").unwrap();
        assert_eq!(h.region, "eu-west-2");
    }

    #[test]
    fn parse_routing_host_aws_s3_website_endpoints() {
        // Dash form: the region is what follows `s3-website-`, not
        // `website-<region>` as the generic `s3-<region>` rule would read it.
        let h = parse_routing_host("site.s3-website-us-west-2.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "us-west-2");
        assert_eq!(h.bucket.as_deref(), Some("site"));

        // Dot form (every region launched since 2014).
        let h = parse_routing_host("my.site.s3-website.eu-central-1.amazonaws.com").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "eu-central-1");
        assert_eq!(h.bucket.as_deref(), Some("my.site"));
    }

    #[test]
    fn parse_routing_host_other_partition_suffixes() {
        let h = parse_routing_host("my-bucket.s3.cn-north-1.amazonaws.com.cn").unwrap();
        assert_eq!(h.service, "s3");
        assert_eq!(h.region, "cn-north-1");
        assert_eq!(h.bucket.as_deref(), Some("my-bucket"));

        let h = parse_routing_host("sqs.cn-northwest-1.amazonaws.com.cn").unwrap();
        assert_eq!(h.service, "sqs");
        assert_eq!(h.region, "cn-northwest-1");

        let h = parse_routing_host("b.s3.us-iso-east-1.c2s.ic.gov").unwrap();
        assert_eq!(h.region, "us-iso-east-1");
        assert_eq!(h.bucket.as_deref(), Some("b"));

        assert!(parse_routing_host("amazonaws.com.cn").is_none());
        assert!(parse_routing_host(".amazonaws.com.cn").is_none());
    }

    #[test]
    fn parse_routing_host_rejects_plain_localhost() {
        assert!(parse_routing_host("localhost:4566").is_none());
        assert!(parse_routing_host("127.0.0.1:4566").is_none());
    }

    #[test]
    fn parse_routing_host_rejects_unknown_suffix() {
        assert!(parse_routing_host("sqs.us-east-1.example.com").is_none());
        assert!(parse_routing_host("s3.us-east-1.aws").is_none());
    }

    #[test]
    fn parse_routing_host_empty_and_malformed_rejected() {
        assert!(parse_routing_host("").is_none());
        assert!(parse_routing_host(".localhost.localstack.cloud").is_none());
        assert!(parse_routing_host("..localhost.localstack.cloud").is_none());
        assert!(parse_routing_host("sqs.localhost.localstack.cloud").is_none());
        assert!(parse_routing_host("foo.bar.baz.localhost.localstack.cloud").is_none());
        assert!(parse_routing_host(".amazonaws.com").is_none());
        assert!(parse_routing_host("amazonaws.com").is_none());
    }

    #[test]
    fn parse_routing_host_bare_s3_accesspoint_does_not_panic() {
        // A single-label "s3-accesspoint" host has < 2 labels, so the
        // virtual-hosted `len() - 2` slice would underflow and panic without
        // the length guard. It must be rejected, not crash the router.
        assert!(parse_routing_host("s3-accesspoint").is_none());
    }

    #[test]
    fn detect_service_via_host_for_rest_service() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "host",
            "s3.us-east-1.localhost.localstack.cloud:4566"
                .parse()
                .unwrap(),
        );
        let query = HashMap::new();
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "s3");
        assert_eq!(detected.protocol, AwsProtocol::Rest);
    }

    #[test]
    fn detect_service_via_host_for_rest_json_service() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "host",
            "lambda.us-east-1.localhost.localstack.cloud:4566"
                .parse()
                .unwrap(),
        );
        let query = HashMap::new();
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "lambda");
        assert_eq!(detected.protocol, AwsProtocol::RestJson);
    }

    #[test]
    fn detect_service_via_host_plus_query_action() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "host",
            "sqs.us-east-1.localhost.localstack.cloud:4566"
                .parse()
                .unwrap(),
        );
        let mut query = HashMap::new();
        query.insert("Action".to_string(), "ListQueues".to_string());
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "sqs");
        assert_eq!(detected.action, "ListQueues");
        assert_eq!(detected.protocol, AwsProtocol::Query);
    }

    #[test]
    fn detect_service_sigv4_wins_over_host() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "authorization",
            "AWS4-HMAC-SHA256 Credential=AKID/20240101/us-east-1/s3/aws4_request, \
             SignedHeaders=host, Signature=abc"
                .parse()
                .unwrap(),
        );
        headers.insert(
            "host",
            "lambda.us-east-1.localhost.localstack.cloud:4566"
                .parse()
                .unwrap(),
        );
        let query = HashMap::new();
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        // SigV4 credential scope says s3; Host header says lambda. SigV4 wins.
        assert_eq!(detected.service, "s3");
        assert_eq!(detected.protocol, AwsProtocol::Rest);
    }

    #[test]
    fn detect_service_host_for_virtual_hosted_s3() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "host",
            "my-bucket.s3.us-east-1.localhost.localstack.cloud:4566"
                .parse()
                .unwrap(),
        );
        let query = HashMap::new();
        let body = Bytes::new();
        let detected = detect_service(&headers, &query, &body).unwrap();
        assert_eq!(detected.service, "s3");
        assert_eq!(detected.protocol, AwsProtocol::Rest);
    }
}