Skip to main content

fakecloud_core/
dispatch.rs

1use axum::body::Body;
2use axum::extract::{ConnectInfo, Extension, Query};
3use axum::http::{Request, StatusCode};
4use axum::response::Response;
5use bytes::Bytes;
6use std::collections::HashMap;
7use std::net::SocketAddr;
8use std::sync::Arc;
9
10use crate::auth::{
11    is_root_bypass, ConditionContext, CredentialResolver, IamMode, IamPolicyEvaluator, Principal,
12    PrincipalType, ResourcePolicyProvider,
13};
14use crate::protocol::{self, AwsProtocol};
15use crate::registry::ServiceRegistry;
16use crate::service::{AwsRequest, ResponseBody};
17
18/// The main dispatch handler. All HTTP requests come through here.
19pub async fn dispatch(
20    ConnectInfo(remote_addr): ConnectInfo<SocketAddr>,
21    Extension(registry): Extension<Arc<ServiceRegistry>>,
22    Extension(config): Extension<Arc<DispatchConfig>>,
23    Query(query_params): Query<HashMap<String, String>>,
24    request: Request<Body>,
25) -> Response<Body> {
26    let remote_addr = Some(remote_addr);
27    let request_id = uuid::Uuid::new_v4().to_string();
28
29    let (parts, body) = request.into_parts();
30
31    // Streaming opt-in: if the route is a known large-body S3 / ECR
32    // upload, we skip the buffered `to_bytes` step entirely and hand
33    // the raw body to the service handler. The handler spills it to
34    // disk on the fly. Header-only detection covers every streaming
35    // candidate (none of them rely on form-body sniffing).
36    let stream_route = streaming_route(
37        &parts.method,
38        parts.uri.path(),
39        &parts.headers,
40        &query_params,
41    );
42    let header_only = protocol::detect_service_headers_only(&parts.headers, &query_params);
43    let stream_dispatch = match (&stream_route, &header_only) {
44        // Header-only detection agrees with the URL match — covers S3
45        // PUT object (SigV4 service=s3 in Authorization).
46        (Some(sr), Some(detected)) if sr.0 == detected.service => Some(detected.clone()),
47        // ECR OCI v2 blob upload has no AWS auth header; the path
48        // alone (`/v2/.../blobs/uploads/...`) tells us the route is
49        // ECR. Synthesize a DetectedRequest so dispatch picks the
50        // streaming path. Same special-case the buffered branch
51        // applies on detect_service None (see below).
52        (Some((service, _)), None) if *service == "ecr" => Some(protocol::DetectedRequest {
53            service: "ecr".to_string(),
54            action: String::new(),
55            protocol: AwsProtocol::Rest,
56        }),
57        _ => None,
58    };
59
60    let (body_bytes, body_stream) = if stream_dispatch.is_some() {
61        (Bytes::new(), Some(body))
62    } else {
63        // Buffered path: materialize the body into memory under the
64        // configured cap. `FAKECLOUD_MAX_REQUEST_BODY_BYTES` (default
65        // 1 GiB) caps non-streaming requests; streaming routes have no
66        // cap because nothing materializes the entire body in RAM.
67        let max_body_bytes = max_request_body_bytes();
68        match axum::body::to_bytes(body, max_body_bytes).await {
69            Ok(b) => (b, None),
70            Err(_) => {
71                return build_error_response(
72                    StatusCode::PAYLOAD_TOO_LARGE,
73                    "RequestEntityTooLarge",
74                    "Request body too large",
75                    &request_id,
76                    AwsProtocol::Query,
77                );
78            }
79        }
80    };
81
82    // Detect service and action
83    let detected = if let Some(d) = stream_dispatch {
84        d
85    } else {
86        match protocol::detect_service(&parts.headers, &query_params, &body_bytes) {
87            Some(d) => d,
88            None => {
89                // A request carrying X-Amz-Target is unambiguously an awsJson
90                // call whose operation we couldn't map to a known service. AWS
91                // answers these with UnknownOperationException; routing them to
92                // the apigateway catch-all below would 404 with a misleading
93                // "Stage not found" instead.
94                if let Some(target) = parts
95                    .headers
96                    .get("x-amz-target")
97                    .and_then(|v| v.to_str().ok())
98                {
99                    return build_error_response(
100                        StatusCode::BAD_REQUEST,
101                        "UnknownOperationException",
102                        &format!("The operation {target} is not recognized."),
103                        &request_id,
104                        AwsProtocol::Json,
105                    );
106                }
107                // OPTIONS requests (CORS preflight) don't carry Authorization headers.
108                // Route them to S3 since S3 is the only REST service that handles CORS.
109                // Note: API Gateway CORS preflight is not fully supported in this emulator
110                // because we can't distinguish between S3 and API Gateway OPTIONS requests
111                // without additional context (in real AWS, they have different domains).
112                if parts.method == http::Method::OPTIONS {
113                    protocol::DetectedRequest {
114                        service: "s3".to_string(),
115                        action: String::new(),
116                        protocol: AwsProtocol::Rest,
117                    }
118                } else if parts.uri.path() == "/v2" || parts.uri.path().starts_with("/v2/") {
119                    // OCI Distribution v2 protocol. Docker CLI / OCI clients
120                    // use Basic auth (not SigV4) and GET /v2/ with no body,
121                    // so this must be matched before the apigateway fallback.
122                    protocol::DetectedRequest {
123                        service: "ecr".to_string(),
124                        action: String::new(),
125                        protocol: AwsProtocol::Rest,
126                    }
127                } else if let Some(bucket) = anonymous_s3_bucket(&parts.uri, &config) {
128                    // Unsigned request whose first path segment names an
129                    // existing S3 bucket: an anonymous path-style S3 access
130                    // (e.g. serving a public-read object to a browser). Without
131                    // this it would fall through to the apigateway catch-all
132                    // below and 404 with "Stage not found" (#1707). Authorization
133                    // for the anonymous caller still runs in the IAM block.
134                    tracing::debug!(bucket = %bucket, "routing unsigned request to S3 (existing bucket)");
135                    protocol::DetectedRequest {
136                        service: "s3".to_string(),
137                        action: String::new(),
138                        protocol: AwsProtocol::Rest,
139                    }
140                } else if !parts.uri.path().starts_with("/_") {
141                    // Requests without AWS auth that don't match any service might be
142                    // API Gateway execute API calls (plain HTTP without signatures).
143                    // Route them to apigateway service which will validate if a matching
144                    // API/stage exists. Skip special FakeCloud endpoints (/_*).
145                    protocol::DetectedRequest {
146                        service: "apigateway".to_string(),
147                        action: String::new(),
148                        protocol: AwsProtocol::RestJson,
149                    }
150                } else {
151                    return build_error_response(
152                        StatusCode::BAD_REQUEST,
153                        "MissingAction",
154                        "Could not determine target service or action from request",
155                        &request_id,
156                        AwsProtocol::Query,
157                    );
158                }
159            }
160        }
161    };
162
163    // Bedrock-agent and bedrock-runtime both send `bedrock` in the SigV4
164    // credential scope, but bedrock-agent has its own service handler.
165    // Disambiguate based on the request path.
166    let detected = if detected.service == "bedrock" {
167        let first_seg = parts.uri.path().split('/').nth(1);
168        if matches!(
169            first_seg,
170            Some(
171                "agents"
172                    | "knowledgebases"
173                    | "flows"
174                    | "prompts"
175                    | "tags"
176                    | "retrieveAndGenerate"
177                    | "retrieveAndGenerateStream"
178                    | "optimize-prompt"
179                    | "sessions"
180                    | "invocations"
181                    | "generate-query"
182                    | "rerank"
183            )
184        ) {
185            // Further disambiguate runtime vs control plane for agents/flows paths
186            let segs: Vec<&str> = parts.uri.path().split('/').collect();
187            let is_runtime = matches!(
188                segs.as_slice(),
189                ["", "agents", _, "agentAliases", _, ..]  // InvokeAgent
190                    | ["", "flows", _, "aliases", _]   // InvokeFlow
191                    | ["", "knowledgebases", _, "retrieve"] // Retrieve
192                    | ["", "retrieveAndGenerate"]
193                    | ["", "retrieveAndGenerateStream"]
194                    | ["", "optimize-prompt"]
195                    | ["", "sessions", ..]
196                    | ["", "invocations", ..]
197                    | ["", "generate-query"]
198                    | ["", "rerank"]
199            );
200            if is_runtime {
201                protocol::DetectedRequest {
202                    service: "bedrock-agent-runtime".to_string(),
203                    ..detected
204                }
205            } else {
206                protocol::DetectedRequest {
207                    service: "bedrock-agent".to_string(),
208                    ..detected
209                }
210            }
211        } else {
212            detected
213        }
214    } else {
215        detected
216    };
217
218    // Amazon DocumentDB shares RDS's Query wire protocol, `rds` SigV4
219    // signing scope, and `rds.<region>.amazonaws.com` endpoint, so a real
220    // `aws-sdk-docdb` request is indistinguishable from `aws-sdk-rds` by
221    // signing name or host alone. The DocumentDB SDK does stamp an
222    // `api/docdb` token into its `user-agent`; use it to route the request
223    // to the dedicated `docdb` handler. (The conformance probe signs the
224    // `docdb` scope directly, so it never reaches this branch.) Requests
225    // without the token stay on `rds`.
226    let detected = if detected.service == "rds" && user_agent_indicates_docdb(&parts.headers) {
227        protocol::DetectedRequest {
228            service: "docdb".to_string(),
229            ..detected
230        }
231    } else {
232        detected
233    };
234
235    // Amazon Neptune shares RDS's Query wire protocol, `rds` SigV4 signing
236    // scope, and `rds.<region>.amazonaws.com` endpoint, so a real
237    // `aws-sdk-neptune` request is indistinguishable from `aws-sdk-rds` by
238    // signing name or host alone. The Neptune SDK does stamp an
239    // `api/neptune` token into its `user-agent`; use it to route the request
240    // to the dedicated `neptune` handler. (The conformance probe signs the
241    // `neptune` scope directly, so it never reaches this branch.) Requests
242    // without the token stay on `rds`.
243    let detected = if detected.service == "rds" && user_agent_indicates_neptune(&parts.headers) {
244        protocol::DetectedRequest {
245            service: "neptune".to_string(),
246            ..detected
247        }
248    } else {
249        detected
250    };
251
252    // Look up service
253    let service = match registry.get(&detected.service) {
254        Some(s) => s,
255        None => {
256            return build_error_response(
257                detected.protocol.error_status(),
258                "UnknownService",
259                &format!("Service '{}' is not available", detected.service),
260                &request_id,
261                detected.protocol,
262            );
263        }
264    };
265
266    // Extract region and access key from auth header (or presigned query).
267    let auth_header = parts
268        .headers
269        .get("authorization")
270        .and_then(|v| v.to_str().ok())
271        .unwrap_or("");
272    let header_info = fakecloud_aws::sigv4::parse_sigv4(auth_header);
273    let presigned_info = if header_info.is_none() {
274        // Presigned URL: credentials live in the query string.
275        fakecloud_aws::sigv4::parse_sigv4_presigned(&query_params).map(|p| p.as_info())
276    } else {
277        None
278    };
279    let sigv4_info = header_info.or(presigned_info);
280    // SigV2 presigned URLs (`AWSAccessKeyId` + `Signature` + `Expires` query
281    // parameters) carry the access key outside the SigV4 grammar, so the SigV4
282    // parsers above return None. Recover the key here so a SigV2-presigned
283    // request is attributed to its caller instead of being treated as
284    // anonymous (which would deny it under the object read auth gate).
285    let access_key_id = sigv4_info
286        .as_ref()
287        .map(|info| info.access_key.clone())
288        .or_else(|| sigv2_presigned_access_key(&query_params));
289
290    // Host-header routing hint: LocalStack-shaped
291    // `<svc>.<region>.localhost.localstack.cloud[:port]`, real-AWS
292    // `<svc>.<region>.amazonaws.com`, and every S3 virtual-hosted variant
293    // of both. Secondary region source and carries the bucket for
294    // virtual-hosted S3 path rewrite.
295    let host_info = protocol::parse_routing_host_from_headers(&parts.headers);
296
297    let region = sigv4_info
298        .map(|info| info.region)
299        .or_else(|| host_info.as_ref().map(|h| h.region.clone()))
300        .or_else(|| extract_region_from_user_agent(&parts.headers))
301        .unwrap_or_else(|| config.region.clone());
302
303    // Resolve the caller's principal up front so both SigV4 verification
304    // (which needs the secret) and the service handler (which needs the
305    // identity for GetCallerIdentity and IAM enforcement) share a single
306    // lookup. The root-bypass AKID skips resolution entirely — `test`
307    // credentials have no backing identity and must always pass.
308    let caller_akid = access_key_id.as_deref().unwrap_or("");
309    let resolved = if !caller_akid.is_empty() && !is_root_bypass(caller_akid) {
310        config
311            .credential_resolver
312            .as_ref()
313            .and_then(|r| r.resolve(caller_akid))
314    } else {
315        None
316    };
317    let caller_principal = resolved.as_ref().map(|r| r.principal.clone());
318    let caller_session_policies = resolved
319        .as_ref()
320        .map(|r| r.session_policies.clone())
321        .unwrap_or_default();
322
323    // Opt-in SigV4 cryptographic verification. Runs before the service
324    // handler so a failing signature never reaches business logic. The
325    // reserved `test*` root identity short-circuits verification to keep
326    // local-dev workflows frictionless.
327    //
328    // A fully anonymous request — no `Authorization` header AND no presigned
329    // credential (SigV4 `X-Amz-Credential` or SigV2 `AWSAccessKeyId`) — carries
330    // no signature to verify, so it must NOT be hard-403'd here even under
331    // `--verify-sigv4`. AWS treats an unsigned request as the anonymous
332    // principal and lets a resource policy / public-read ACL authorize it
333    // (e.g. a public S3 object GET). Let it fall through to the
334    // anonymous-authorization path below, which allows a public resource and
335    // otherwise denies. A request that DID present a credential but failed to
336    // parse is NOT anonymous (`Authorization` non-empty or a presign query
337    // present) and still returns the malformed-signature error below.
338    let is_fully_anonymous = auth_header.is_empty()
339        && !query_params.contains_key("X-Amz-Credential")
340        && sigv2_presigned_access_key(&query_params).is_none();
341    if config.verify_sigv4
342        && !is_fully_anonymous
343        && !is_root_bypass(caller_akid)
344        && config.credential_resolver.is_some()
345    {
346        let amz_date = parts
347            .headers
348            .get("x-amz-date")
349            .and_then(|v| v.to_str().ok());
350        let parsed = fakecloud_aws::sigv4::parse_sigv4_header(auth_header, amz_date)
351            .or_else(|| fakecloud_aws::sigv4::parse_sigv4_presigned(&query_params));
352        let parsed = match parsed {
353            Some(p) => p,
354            None => {
355                return build_error_response(
356                    StatusCode::FORBIDDEN,
357                    "IncompleteSignature",
358                    "Request is missing or has a malformed AWS Signature",
359                    &request_id,
360                    detected.protocol,
361                );
362            }
363        };
364        let resolved_for_verify = match resolved.as_ref() {
365            Some(r) => r,
366            None => {
367                return build_error_response(
368                    StatusCode::FORBIDDEN,
369                    "InvalidClientTokenId",
370                    "The security token included in the request is invalid",
371                    &request_id,
372                    detected.protocol,
373                );
374            }
375        };
376        let headers_vec = fakecloud_aws::sigv4::headers_from_http(&parts.headers);
377        let raw_query_for_verify = parts.uri.query().unwrap_or("").to_string();
378        let verify_req = fakecloud_aws::sigv4::VerifyRequest {
379            method: parts.method.as_str(),
380            path: parts.uri.path(),
381            query: &raw_query_for_verify,
382            headers: &headers_vec,
383            body: &body_bytes,
384        };
385        match fakecloud_aws::sigv4::verify(
386            &parsed,
387            &verify_req,
388            &resolved_for_verify.secret_access_key,
389            chrono::Utc::now(),
390        ) {
391            Ok(()) => {
392                // Bind the buffered request body to the signed
393                // `x-amz-content-sha256`. The sigv4 canonical builder uses that
394                // header value verbatim (it is a signed header) and deliberately
395                // does NOT re-hash the body: for streaming / aws-chunked routes
396                // `body_bytes` is empty at that layer, so re-hashing there would
397                // reject legitimate signed requests (see the
398                // `feedback_sigv4_body_hash_wrong_layer` caveat). S3 rebinds the
399                // body hash in its own write path (`XAmzContentSHA256Mismatch`);
400                // every other service is bound HERE, where the full buffered body
401                // is available. Only a genuine 64-char lowercase-hex digest is
402                // checked — `UNSIGNED-PAYLOAD`, `STREAMING-*`, and presigned
403                // (UNSIGNED-PAYLOAD) requests are skipped, so correct clients are
404                // unaffected. A mismatch means the body was altered after signing;
405                // AWS would then compute a different canonical request, so return
406                // `SignatureDoesNotMatch`.
407                if !parsed.is_presigned && detected.service != "s3" {
408                    if let Some(signed_hash) = parts
409                        .headers
410                        .get("x-amz-content-sha256")
411                        .and_then(|v| v.to_str().ok())
412                        .filter(|h| is_hex_sha256(h))
413                    {
414                        if sha256_hex_lower(&body_bytes) != signed_hash {
415                            return build_error_response(
416                                StatusCode::FORBIDDEN,
417                                "SignatureDoesNotMatch",
418                                "The request signature we calculated does not match the signature you provided",
419                                &request_id,
420                                detected.protocol,
421                            );
422                        }
423                    }
424                }
425            }
426            Err(fakecloud_aws::sigv4::SigV4Error::RequestTimeTooSkewed { .. }) => {
427                return build_error_response(
428                    StatusCode::FORBIDDEN,
429                    "RequestTimeTooSkewed",
430                    "The difference between the request time and the current time is too large",
431                    &request_id,
432                    detected.protocol,
433                );
434            }
435            Err(fakecloud_aws::sigv4::SigV4Error::InvalidDate(msg)) => {
436                return build_error_response(
437                    StatusCode::FORBIDDEN,
438                    "IncompleteSignature",
439                    &format!("Invalid x-amz-date: {msg}"),
440                    &request_id,
441                    detected.protocol,
442                );
443            }
444            Err(fakecloud_aws::sigv4::SigV4Error::Malformed(msg)) => {
445                return build_error_response(
446                    StatusCode::FORBIDDEN,
447                    "IncompleteSignature",
448                    &format!("Malformed SigV4 signature: {msg}"),
449                    &request_id,
450                    detected.protocol,
451                );
452            }
453            Err(fakecloud_aws::sigv4::SigV4Error::SignatureMismatch) => {
454                return build_error_response(
455                    StatusCode::FORBIDDEN,
456                    "SignatureDoesNotMatch",
457                    "The request signature we calculated does not match the signature you provided",
458                    &request_id,
459                    detected.protocol,
460                );
461            }
462            Err(fakecloud_aws::sigv4::SigV4Error::PresignedUrlExpired { .. }) => {
463                return build_error_response(
464                    StatusCode::FORBIDDEN,
465                    "AccessDenied",
466                    "Request has expired",
467                    &request_id,
468                    detected.protocol,
469                );
470            }
471            Err(fakecloud_aws::sigv4::SigV4Error::InvalidPresignExpires(_)) => {
472                return build_error_response(
473                    StatusCode::BAD_REQUEST,
474                    "AuthorizationQueryParametersError",
475                    "X-Amz-Expires must be a number between 1 and 604800 seconds",
476                    &request_id,
477                    detected.protocol,
478                );
479            }
480        }
481    }
482
483    // Build path segments. For S3 virtual-hosted-style requests the bucket
484    // lives in the Host header, not the path — prepend it so the S3 handler
485    // sees a uniform path-style request. SigV4 verification above already
486    // ran against the wire path, so this rewrite is signature-safe.
487    let wire_path = parts.uri.path();
488    let path = if detected.service == "s3" {
489        if let Some(bucket) = host_info.as_ref().and_then(|h| h.bucket.as_deref()) {
490            let prefix_with_slash = format!("/{bucket}/");
491            let is_bucket_root = wire_path.trim_end_matches('/') == format!("/{bucket}");
492            if wire_path.starts_with(&prefix_with_slash) || is_bucket_root {
493                wire_path.to_string()
494            } else if wire_path == "/" || wire_path.is_empty() {
495                format!("/{bucket}")
496            } else {
497                format!("/{bucket}{wire_path}")
498            }
499        } else {
500            wire_path.to_string()
501        }
502    } else {
503        wire_path.to_string()
504    };
505    let raw_query = parts.uri.query().unwrap_or("").to_string();
506    let path_segments: Vec<String> = path
507        .split('/')
508        .filter(|s| !s.is_empty())
509        .map(|s| s.to_string())
510        .collect();
511
512    // For JSON protocol, validate that non-empty bodies are valid JSON
513    if detected.protocol == AwsProtocol::Json
514        && !body_bytes.is_empty()
515        && serde_json::from_slice::<serde_json::Value>(&body_bytes).is_err()
516    {
517        return build_error_response(
518            StatusCode::BAD_REQUEST,
519            "SerializationException",
520            "Start of structure or map found where not expected",
521            &request_id,
522            AwsProtocol::Json,
523        );
524    }
525
526    // Merge query params with form body params for the Query family (awsQuery
527    // and ec2Query share identical form-encoded request bodies).
528    let mut all_params = query_params;
529    if matches!(
530        detected.protocol,
531        AwsProtocol::Query | AwsProtocol::Ec2Query
532    ) {
533        let body_params = protocol::parse_query_body(&body_bytes);
534        for (k, v) in body_params {
535            all_params.entry(k).or_insert(v);
536        }
537    }
538
539    // CloudWatch (`monitoring`) advertises awsJson1_0 alongside awsQuery. Its
540    // handlers all read the flat awsQuery param map, so when a client uses the
541    // JSON protocol we flatten the JSON body into that same map, leaving the
542    // handlers unchanged. The handler emits a JSON response for JSON callers.
543    if detected.protocol == AwsProtocol::Json && detected.service == "monitoring" {
544        let body_params = protocol::flatten_json_to_query(&body_bytes);
545        for (k, v) in body_params {
546            all_params.entry(k).or_insert(v);
547        }
548    }
549
550    let aws_request = AwsRequest {
551        service: detected.service.clone(),
552        action: detected.action.clone(),
553        region,
554        account_id: caller_principal
555            .as_ref()
556            .map(|p| p.account_id.clone())
557            .unwrap_or_else(|| config.account_id.clone()),
558        request_id: request_id.clone(),
559        headers: parts.headers,
560        query_params: all_params,
561        body: body_bytes,
562        body_stream: parking_lot::Mutex::new(body_stream),
563        path_segments,
564        raw_path: path,
565        raw_query,
566        method: parts.method,
567        is_query_protocol: matches!(
568            detected.protocol,
569            AwsProtocol::Query | AwsProtocol::Ec2Query
570        ),
571        access_key_id,
572        principal: caller_principal,
573    };
574
575    tracing::info!(
576        service = %aws_request.service,
577        action = %aws_request.action,
578        request_id = %aws_request.request_id,
579        "handling request"
580    );
581
582    // Opt-in IAM identity-policy enforcement. Runs before the service
583    // handler so a deny never reaches business logic. Root principals
584    // (both `test*` bypass AKIDs and the account's IAM root) are exempt,
585    // matching AWS behavior. Services that haven't opted in via
586    // `iam_enforceable()` are transparently skipped — the startup log
587    // lists which services are under enforcement so users always know.
588    if config.iam_mode.is_enabled()
589        && service.iam_enforceable()
590        && !is_root_bypass(aws_request.access_key_id.as_deref().unwrap_or(""))
591    {
592        if let Some(evaluator) = config.policy_evaluator.as_ref() {
593            if let Some(principal) = aws_request.principal.as_ref() {
594                if !principal.is_root() {
595                    // A request can need several authorizations -- one per
596                    // table in a batch, say -- and every one must allow it.
597                    let iam_actions = service.iam_actions_for(&aws_request);
598                    if !iam_actions.is_empty() {
599                        for iam_action in &iam_actions {
600                            let mut condition_context = build_condition_context(
601                                principal,
602                                remote_addr,
603                                &aws_request.region,
604                                is_secure_transport(&aws_request.headers),
605                            );
606                            // F3 keys riding on the resolved credential. STS
607                            // populates these at mint time so subsequent
608                            // requests under the credential can be evaluated
609                            // against `aws:MultiFactorAuthPresent`,
610                            // `aws:MultiFactorAuthAge`, `aws:TokenIssueTime`,
611                            // and `aws:FederatedProvider`. IAM user access
612                            // keys carry none of these, matching AWS.
613                            if let Some(rc) = resolved.as_ref() {
614                                condition_context.aws_mfa_present = Some(rc.mfa_present);
615                                condition_context.aws_token_issue_time = rc.token_issued_at;
616                                condition_context.aws_federated_provider =
617                                    rc.federated_provider.clone();
618                                // `aws:MultiFactorAuthAge` is "seconds since
619                                // MFA was asserted" — computed at evaluation
620                                // time from the token issue moment so the
621                                // value increases monotonically as the session
622                                // ages. Only set when the session was actually
623                                // minted with MFA; otherwise the key is
624                                // absent, matching AWS.
625                                if rc.mfa_present {
626                                    if let Some(issued) = rc.token_issued_at {
627                                        let age = chrono::Utc::now()
628                                            .signed_duration_since(issued)
629                                            .num_seconds()
630                                            .max(0);
631                                        condition_context.aws_mfa_age_seconds = Some(age);
632                                    }
633                                }
634                            }
635                            condition_context.service_keys =
636                                service.iam_condition_keys_for(&aws_request, iam_action);
637
638                            // ABAC: populate tag-based condition keys.
639                            // aws:ResourceTag/*
640                            match service.resource_tags_for(&iam_action.resource) {
641                                Some(tags) => condition_context.resource_tags = Some(tags),
642                                None => tracing::debug!(
643                                    target: "fakecloud::iam::audit",
644                                    service = %detected.service,
645                                    resource = %iam_action.resource,
646                                    "service does not expose resource tags for ABAC; skipping aws:ResourceTag/* evaluation"
647                                ),
648                            }
649                            // aws:RequestTag/* + aws:TagKeys
650                            match service.request_tags_from(&aws_request, iam_action.action) {
651                                Some(tags) => condition_context.request_tags = Some(tags),
652                                None => tracing::debug!(
653                                    target: "fakecloud::iam::audit",
654                                    service = %detected.service,
655                                    action = %iam_action.action_string(),
656                                    "service does not expose request tags for ABAC; skipping aws:RequestTag/* / aws:TagKeys evaluation"
657                                ),
658                            }
659                            // aws:PrincipalTag/*
660                            condition_context.principal_tags = principal.tags.clone();
661
662                            // Phase 2: fetch the resource-based policy (if
663                            // any) attached to the target resource and
664                            // pass it to the evaluator alongside the
665                            // principal's identity policies. The resource's
666                            // owning account is parsed from the ARN (#381
667                            // multi-account alignment); S3 ARNs have an
668                            // empty account field, so we fall back to the
669                            // server's configured account ID in that case.
670                            let resource_policy_json =
671                                config.resource_policy_provider.as_ref().and_then(|p| {
672                                    p.resource_policy(&detected.service, &iam_action.resource)
673                                });
674                            // Derive the resource-owning account. Prefer a provider
675                            // lookup (S3 ARNs carry no account, so the bucket's
676                            // owner is resolved from state — without this, account
677                            // A reaching account B's bucket would be mis-read as
678                            // same-account and skip B's bucket-policy requirement,
679                            // bug-audit 2026-05-28, 5.3), then fall back to the
680                            // account embedded in the ARN (SQS/SNS/Lambda/…), then
681                            // to the caller's account for wildcard / unscoped
682                            // actions (ListQueues, GetCallerIdentity).
683                            let resource_account_id = config
684                                .resource_policy_provider
685                                .as_ref()
686                                .and_then(|p| {
687                                    p.resource_owner_account(
688                                        &detected.service,
689                                        &iam_action.resource,
690                                    )
691                                })
692                                .or_else(|| parse_account_from_arn(&iam_action.resource))
693                                .unwrap_or_else(|| principal.account_id.clone());
694                            // SCP ceiling: resolve the inherited SCP chain
695                            // for this principal (management accounts and
696                            // service-linked roles come back as `None`, in
697                            // which case the evaluator treats the layer as
698                            // absent). Audit breadcrumbs emitted by the
699                            // resolver itself, not here.
700                            let scps = config
701                                .scp_resolver
702                                .as_ref()
703                                .and_then(|r| r.scps_for(principal));
704                            let decision = evaluator.evaluate_with_resource_policy(
705                                principal,
706                                iam_action,
707                                &condition_context,
708                                resource_policy_json.as_deref(),
709                                &resource_account_id,
710                                &caller_session_policies,
711                                scps.as_deref(),
712                            );
713                            if !decision.is_allow() {
714                                tracing::warn!(
715                                    target: "fakecloud::iam::audit",
716                                    service = %detected.service,
717                                    action = %iam_action.action_string(),
718                                    resource = %iam_action.resource,
719                                    principal = %principal.arn,
720                                    resource_policy_present = resource_policy_json.is_some(),
721                                    decision = ?decision,
722                                    mode = %config.iam_mode,
723                                    request_id = %request_id,
724                                    "IAM policy evaluation denied request"
725                                );
726                                if config.iam_mode.is_strict() {
727                                    // Real AWS includes an "Encoded
728                                    // authorization failure message" suffix
729                                    // on AccessDeniedException — an opaque
730                                    // base64+zlib JSON blob that the caller
731                                    // can pass to STS
732                                    // `DecodeAuthorizationMessage` to
733                                    // recover the structured deny reason
734                                    // (action, principal, matched
735                                    // statements, condition context). We
736                                    // produce the same blob inline so
737                                    // existing tooling that decodes deny
738                                    // reasons works against fakecloud.
739                                    let context_summary = serde_json::json!({
740                                        "aws:PrincipalArn": principal.arn,
741                                        "aws:PrincipalAccount": principal.account_id,
742                                        "aws:RequestedRegion": condition_context
743                                            .aws_requested_region
744                                            .clone()
745                                            .unwrap_or_default(),
746                                        "aws:SecureTransport": condition_context
747                                            .aws_secure_transport
748                                            .unwrap_or(false),
749                                        "aws:Action": iam_action.action_string(),
750                                        "aws:Resource": iam_action.resource,
751                                        "decision": format!("{:?}", decision),
752                                    });
753                                    let action_string = iam_action.action_string();
754                                    let encoded = crate::auth_message::encode_deny(
755                                        matches!(decision, crate::auth::IamDecision::ExplicitDeny),
756                                        Some(&action_string),
757                                        Some(&principal.arn),
758                                        Vec::new(),
759                                        Some(context_summary),
760                                    );
761                                    return build_error_response(
762                                    StatusCode::FORBIDDEN,
763                                    "AccessDeniedException",
764                                    &format!(
765                                        "User: {} is not authorized to perform: {} on resource: {} Encoded authorization failure message: {}",
766                                        principal.arn,
767                                        iam_action.action_string(),
768                                        iam_action.resource,
769                                        encoded,
770                                    ),
771                                    &request_id,
772                                    detected.protocol,
773                                );
774                                }
775                                // Soft mode: audit log already emitted; fall
776                                // through to the handler.
777                            }
778                        }
779                    } else {
780                        // Service opted in via `iam_enforceable()` but its
781                        // `iam_action_for` returned no `IamAction` for this
782                        // specific operation (e.g. S3's `s3_detect_action`
783                        // has `_ => return None` arms for unrecognized
784                        // sub-resources). Under strict enforcement that must
785                        // fail closed: an operation we cannot map to an IAM
786                        // action cannot be authorized, so serving it would be
787                        // a fail-open bypass of `--iam strict`. Deny by
788                        // default. In soft mode we preserve the historical
789                        // warn-and-allow so an incomplete mapping surfaces
790                        // during rollout without blocking traffic.
791                        tracing::warn!(
792                            target: "fakecloud::iam::audit",
793                            service = %detected.service,
794                            action = %aws_request.action,
795                            mode = %config.iam_mode,
796                            request_id = %request_id,
797                            "service is iam_enforceable but has no IamAction mapping for this action; denying under strict, allowing under soft"
798                        );
799                        if config.iam_mode.is_strict() {
800                            return build_error_response(
801                                StatusCode::FORBIDDEN,
802                                "AccessDeniedException",
803                                &format!(
804                                    "User: {} is not authorized to perform: {}: no IAM action mapping exists for this operation, so it cannot be authorized under strict IAM enforcement",
805                                    principal.arn, aws_request.action,
806                                ),
807                                &request_id,
808                                detected.protocol,
809                            );
810                        }
811                        // Soft mode: audit log emitted; fall through to the
812                        // handler.
813                    }
814                }
815            } else if aws_request.access_key_id.is_none() {
816                // Truly anonymous (unsigned) caller — no Authorization header at
817                // all. No identity policies exist, so authorization rests
818                // entirely on the resource policy (a bucket policy granting
819                // `Principal:"*"`) and public-read ACLs — mirroring AWS, which
820                // denies anonymous requests unless the resource is explicitly
821                // made public. Without this an anonymous request that reached an
822                // iam_enforceable service in enforcement mode would bypass
823                // authorization entirely.
824                //
825                // A request that carried an Authorization header but whose
826                // credential did not resolve (principal `None` with
827                // `access_key_id` `Some`) is intentionally left alone here: with
828                // SigV4 verification off, fakecloud does not reject unverified
829                // signed requests, and turning them into anonymous denials would
830                // change long-standing behavior.
831                let iam_actions = service.iam_actions_for(&aws_request);
832                if !iam_actions.is_empty() {
833                    for iam_action in &iam_actions {
834                        let now = chrono::Utc::now();
835                        let mut condition_context = ConditionContext {
836                            aws_source_ip: remote_addr.map(|sa| sa.ip()),
837                            aws_current_time: Some(now),
838                            aws_epoch_time: Some(now.timestamp()),
839                            aws_secure_transport: Some(is_secure_transport(&aws_request.headers)),
840                            aws_requested_region: Some(aws_request.region.clone()),
841                            ..Default::default()
842                        };
843                        condition_context.service_keys =
844                            service.iam_condition_keys_for(&aws_request, iam_action);
845                        let resource_policy_json =
846                            config.resource_policy_provider.as_ref().and_then(|p| {
847                                p.resource_policy(&detected.service, &iam_action.resource)
848                            });
849                        let policy_decision = evaluator.evaluate_anonymous(
850                            iam_action,
851                            &condition_context,
852                            resource_policy_json.as_deref(),
853                        );
854                        let policy_allows = policy_decision.is_allow();
855                        // An explicit Deny in the resource policy always wins, even
856                        // over a public-read ACL — matching AWS's Deny-overrides
857                        // precedence. Collapsing the decision to a bool and ORing the
858                        // ACL let a public ACL override an explicit anonymous Deny.
859                        let policy_explicit_deny =
860                            matches!(policy_decision, crate::auth::IamDecision::ExplicitDeny);
861                        let acl_allows = !policy_explicit_deny
862                            && config.resource_policy_provider.as_ref().is_some_and(|p| {
863                                p.public_acl_allows(
864                                    &detected.service,
865                                    &iam_action.resource,
866                                    iam_action.action,
867                                )
868                            });
869                        if !policy_allows && !acl_allows {
870                            tracing::warn!(
871                                target: "fakecloud::iam::audit",
872                                service = %detected.service,
873                                action = %iam_action.action_string(),
874                                resource = %iam_action.resource,
875                                resource_policy_present = resource_policy_json.is_some(),
876                                mode = %config.iam_mode,
877                                request_id = %request_id,
878                                "anonymous request denied: no public bucket policy or ACL grants the action"
879                            );
880                            if config.iam_mode.is_strict() {
881                                return build_error_response(
882                                    StatusCode::FORBIDDEN,
883                                    "AccessDenied",
884                                    "Access Denied",
885                                    &request_id,
886                                    detected.protocol,
887                                );
888                            }
889                            // Soft mode: audit log emitted; fall through to the handler.
890                        }
891                    }
892                } else {
893                    // Anonymous request to an iam_enforceable service whose
894                    // operation has no IamAction mapping. Mirror the signed
895                    // branch above: an operation we cannot map to an IAM action
896                    // cannot be authorized, so serving it would be a fail-open
897                    // bypass of `--iam strict`. Deny by default under strict;
898                    // soft mode warns and falls through.
899                    tracing::warn!(
900                        target: "fakecloud::iam::audit",
901                        service = %detected.service,
902                        action = %aws_request.action,
903                        mode = %config.iam_mode,
904                        request_id = %request_id,
905                        "anonymous request to iam_enforceable service has no IamAction mapping; denying under strict, allowing under soft"
906                    );
907                    if config.iam_mode.is_strict() {
908                        return build_error_response(
909                            StatusCode::FORBIDDEN,
910                            "AccessDenied",
911                            "Access Denied",
912                            &request_id,
913                            detected.protocol,
914                        );
915                    }
916                }
917            }
918        }
919    }
920
921    match service.handle(aws_request).await {
922        Ok(resp) => {
923            let mut builder = Response::builder()
924                .status(resp.status)
925                .header("x-amzn-requestid", &request_id)
926                .header("x-amz-request-id", &request_id);
927
928            if !resp.content_type.is_empty() {
929                builder = builder.header("content-type", &resp.content_type);
930            }
931
932            let has_content_length = resp
933                .headers
934                .iter()
935                .any(|(k, _)| k.as_str().eq_ignore_ascii_case("content-length"));
936
937            for (k, v) in &resp.headers {
938                builder = builder.header(k, v);
939            }
940
941            match resp.body {
942                ResponseBody::Bytes(b) => builder.body(Body::from(b)).unwrap(),
943                ResponseBody::File { file, size } => {
944                    let stream = tokio_util::io::ReaderStream::new(file);
945                    let body = Body::from_stream(stream);
946                    if !has_content_length {
947                        builder = builder.header("content-length", size.to_string());
948                    }
949                    builder.body(body).unwrap()
950                }
951            }
952        }
953        Err(err) => {
954            tracing::warn!(
955                service = %detected.service,
956                action = %detected.action,
957                error = %err,
958                "request failed"
959            );
960            let error_headers = err.response_headers().to_vec();
961            let mut resp = build_error_response_with_fields(
962                err.status(),
963                err.code(),
964                &err.message(),
965                &request_id,
966                detected.protocol,
967                err.extra_fields(),
968            );
969            for (k, v) in &error_headers {
970                if let (Ok(name), Ok(val)) = (
971                    k.parse::<http::header::HeaderName>(),
972                    v.parse::<http::header::HeaderValue>(),
973                ) {
974                    // `Vary` combines, so two entries for it must both survive
975                    // (S3 adds its CORS value to an error that may carry its
976                    // own) — but an identical value is not worth repeating.
977                    // Everything else replaces, so a service can still override
978                    // a header this builder already set.
979                    if name == http::header::VARY {
980                        let already = resp
981                            .headers()
982                            .get_all(&name)
983                            .iter()
984                            .any(|existing| existing == val);
985                        if !already {
986                            resp.headers_mut().append(name, val);
987                        }
988                    } else {
989                        resp.headers_mut().insert(name, val);
990                    }
991                }
992            }
993            resp
994        }
995    }
996}
997
998/// Configuration passed to the dispatch handler.
999#[derive(Clone)]
1000pub struct DispatchConfig {
1001    pub region: String,
1002    pub account_id: String,
1003    /// Whether to cryptographically verify SigV4 signatures on incoming
1004    /// requests. Wired through from `--verify-sigv4` /
1005    /// `FAKECLOUD_VERIFY_SIGV4`. Off by default.
1006    pub verify_sigv4: bool,
1007    /// IAM policy evaluation mode. Wired through from `--iam` /
1008    /// `FAKECLOUD_IAM`. Defaults to [`IamMode::Off`]. Actual evaluation is
1009    /// added in a later batch; today this field is plumbed but never
1010    /// consulted.
1011    pub iam_mode: IamMode,
1012    /// Resolves access key IDs to their secrets and owning principals.
1013    /// Required when `verify_sigv4` or `iam_mode != Off`. When `None`, both
1014    /// features gracefully degrade to off-by-default behavior.
1015    pub credential_resolver: Option<Arc<dyn CredentialResolver>>,
1016    /// Evaluates IAM identity policies for a resolved principal + action.
1017    /// Required when `iam_mode != Off`. When `None`, enforcement silently
1018    /// degrades to off even if `iam_mode` is set.
1019    pub policy_evaluator: Option<Arc<dyn IamPolicyEvaluator>>,
1020    /// Resolves resource-based policies (S3 bucket policies in the
1021    /// initial rollout) to hand to the evaluator alongside the
1022    /// principal's identity policies. `None` means the server was
1023    /// started without any resource-policy-owning service registered;
1024    /// dispatch then behaves as if no resource policy is attached to
1025    /// any resource, identical to the Phase 1 behavior.
1026    pub resource_policy_provider: Option<Arc<dyn ResourcePolicyProvider>>,
1027    /// Resolves the ordered SCP chain that applies to a principal's
1028    /// account (root-OU first, account-direct last). `None` means no
1029    /// organizations resolver has been registered — SCPs never gate
1030    /// any request in that case. Off-by-default matches the Batch 4
1031    /// contract: zero behavior change until a user calls
1032    /// `CreateOrganization` and the resolver is wired.
1033    pub scp_resolver: Option<Arc<dyn crate::auth::ScpResolver>>,
1034}
1035
1036impl std::fmt::Debug for DispatchConfig {
1037    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1038        f.debug_struct("DispatchConfig")
1039            .field("region", &self.region)
1040            .field("account_id", &self.account_id)
1041            .field("verify_sigv4", &self.verify_sigv4)
1042            .field("iam_mode", &self.iam_mode)
1043            .field(
1044                "credential_resolver",
1045                &self
1046                    .credential_resolver
1047                    .as_ref()
1048                    .map(|_| "<CredentialResolver>"),
1049            )
1050            .field(
1051                "policy_evaluator",
1052                &self
1053                    .policy_evaluator
1054                    .as_ref()
1055                    .map(|_| "<IamPolicyEvaluator>"),
1056            )
1057            .field(
1058                "resource_policy_provider",
1059                &self
1060                    .resource_policy_provider
1061                    .as_ref()
1062                    .map(|_| "<ResourcePolicyProvider>"),
1063            )
1064            .field(
1065                "scp_resolver",
1066                &self.scp_resolver.as_ref().map(|_| "<ScpResolver>"),
1067            )
1068            .finish()
1069    }
1070}
1071
1072impl DispatchConfig {
1073    /// Minimal constructor for tests and call sites that don't care about the
1074    /// opt-in security features.
1075    pub fn new(region: impl Into<String>, account_id: impl Into<String>) -> Self {
1076        Self {
1077            region: region.into(),
1078            account_id: account_id.into(),
1079            verify_sigv4: false,
1080            iam_mode: IamMode::Off,
1081            credential_resolver: None,
1082            policy_evaluator: None,
1083            resource_policy_provider: None,
1084            scp_resolver: None,
1085        }
1086    }
1087}
1088
1089/// Extract the 12-digit account ID segment from an AWS ARN.
1090///
1091/// ARNs follow `arn:<partition>:<service>:<region>:<account>:<resource>`.
1092/// Identifies routes that opt into streaming request bodies. Returns
1093/// `Some((service, action_hint))` when the dispatch path should hand
1094/// the raw body to the service handler unbuffered, otherwise `None`
1095/// for the default buffered path. The handler reads the stream via
1096/// [`crate::service::AwsRequest::take_body_stream`].
1097///
1098/// Streaming-eligible routes today:
1099///
1100/// * `s3` PUT object — `PUT /<bucket>/<key>` with a SigV4 (or
1101///   presigned) auth header. Covers PutObject, UploadPart, and
1102///   UploadPartCopy. The S3 service spills to disk via
1103///   [`fakecloud_persistence::BodySource::File`] when the stream is
1104///   present.
1105/// * `ecr` OCI Distribution v2 blob upload — `PATCH` and `PUT` on
1106///   `/v2/{name}/blobs/uploads/{uuid}`. The ECR service spools the
1107///   stream into a per-upload temp file before computing the digest.
1108fn streaming_route(
1109    method: &http::Method,
1110    path: &str,
1111    headers: &http::HeaderMap,
1112    query_params: &HashMap<String, String>,
1113) -> Option<(&'static str, &'static str)> {
1114    // ECR OCI v2 blob upload (PATCH chunk + final PUT).
1115    if (method == http::Method::PATCH || method == http::Method::PUT)
1116        && path.starts_with("/v2/")
1117        && path.contains("/blobs/uploads/")
1118    {
1119        return Some(("ecr", ""));
1120    }
1121
1122    // S3 PutObject / UploadPart / UploadPartCopy. Detect either via
1123    // SigV4 service field in the Authorization header OR via a SigV4
1124    // presigned URL (X-Amz-Credential .../s3/...) OR a SigV2 presigned
1125    // URL (AWSAccessKeyId + Signature + Expires query parameters).
1126    if method == http::Method::PUT {
1127        let after = path.trim_start_matches('/');
1128        // Path-style PutObject is `PUT /<bucket>/<key>` (path contains a
1129        // slash); virtual-hosted-style is `PUT /<key>` with the bucket
1130        // in the Host header. For virtual-hosted, accept any non-empty
1131        // path so the key flows through the streaming dispatch — the
1132        // Host parser already routed this request to S3.
1133        let virtual_hosted_s3 = protocol::parse_routing_host_from_headers(headers)
1134            .filter(|h| h.service == "s3" && h.bucket.is_some())
1135            .is_some();
1136        if after.is_empty() || (!virtual_hosted_s3 && !after.contains('/')) {
1137            return None;
1138        }
1139        let header_s3 = headers
1140            .get("authorization")
1141            .and_then(|v| v.to_str().ok())
1142            .and_then(fakecloud_aws::sigv4::parse_sigv4)
1143            .map(|info| info.service == "s3")
1144            .unwrap_or(false);
1145        let presigned_v4_s3 = query_params
1146            .get("X-Amz-Credential")
1147            .and_then(|c| c.split('/').nth(3).map(|s| s.to_string()))
1148            .map(|service| service == "s3")
1149            .unwrap_or(false);
1150        let presigned_v2 = query_params.contains_key("AWSAccessKeyId")
1151            && query_params.contains_key("Signature")
1152            && query_params.contains_key("Expires");
1153        if header_s3 || presigned_v4_s3 || presigned_v2 {
1154            return Some(("s3", ""));
1155        }
1156    }
1157
1158    None
1159}
1160
1161/// Default request-body buffering cap. fakecloud reads the entire
1162/// request body into memory before handing it to a service handler,
1163/// so this ceiling caps RAM usage per in-flight request.
1164///
1165/// Default 1 GiB — comfortably above legitimate single S3 PutObject
1166/// payloads (AWS recommends multipart above ~100 MiB) and each
1167/// multipart part dispatches through here separately. Override with
1168/// `FAKECLOUD_MAX_REQUEST_BODY_BYTES` (decimal bytes) when running
1169/// stress tests that push past the default.
1170const DEFAULT_MAX_REQUEST_BODY_BYTES: usize = 1024 * 1024 * 1024;
1171
1172/// The server-wide buffered request body cap in bytes, from
1173/// `FAKECLOUD_MAX_REQUEST_BODY_BYTES` (default 1 GiB). Public so buffered
1174/// sub-proxies (e.g. the CloudFront viewer data plane) apply the SAME cap as
1175/// direct traffic rather than an inconsistent limit of their own.
1176pub fn max_request_body_bytes() -> usize {
1177    static CACHED: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
1178    *CACHED.get_or_init(|| {
1179        std::env::var("FAKECLOUD_MAX_REQUEST_BODY_BYTES")
1180            .ok()
1181            .and_then(|s| s.parse::<usize>().ok())
1182            .filter(|&n| n > 0)
1183            .unwrap_or(DEFAULT_MAX_REQUEST_BODY_BYTES)
1184    })
1185}
1186
1187/// For the cross-account decision in IAM enforcement, the "resource
1188/// account" is the ARN's account segment. Some services (notably S3)
1189/// produce ARNs with an empty account field — for those we return
1190/// `None` and let the caller fall back to the server's configured
1191/// account ID. Malformed or non-ARN strings also return `None`.
1192fn parse_account_from_arn(arn: &str) -> Option<String> {
1193    let mut parts = arn.splitn(6, ':');
1194    if parts.next()? != "arn" {
1195        return None;
1196    }
1197    let _partition = parts.next()?;
1198    let _service = parts.next()?;
1199    let _region = parts.next()?;
1200    let account = parts.next()?;
1201    // Resource segment must exist (parts.next().is_some()) for the ARN
1202    // to be well-formed, but we don't consume its value here.
1203    parts.next()?;
1204    if account.is_empty() {
1205        None
1206    } else {
1207        Some(account.to_string())
1208    }
1209}
1210
1211/// Whether the request's `user-agent` (or `x-amz-user-agent`) carries the
1212/// `api/neptune` SDK-metadata token the `aws-sdk-neptune` client stamps in.
1213/// Used to disambiguate Neptune from RDS, which share the `rds` SigV4 scope
1214/// and endpoint. Matches the `api/neptune` token that AWS SDKs emit (a
1215/// leading `api/neptune` optionally followed by `#`/`/` and a version).
1216fn user_agent_indicates_neptune(headers: &http::HeaderMap) -> bool {
1217    for name in ["user-agent", "x-amz-user-agent"] {
1218        if let Some(ua) = headers.get(name).and_then(|v| v.to_str().ok()) {
1219            for part in ua.split_whitespace() {
1220                if let Some(rest) = part.strip_prefix("api/neptune") {
1221                    if rest.is_empty() || rest.starts_with('#') || rest.starts_with('/') {
1222                        return true;
1223                    }
1224                }
1225            }
1226        }
1227    }
1228    false
1229}
1230
1231/// Extract region from User-Agent header suffix `region/<region>`.
1232/// Whether the request's `user-agent` (or `x-amz-user-agent`) carries the
1233/// `api/docdb` SDK-metadata token the `aws-sdk-docdb` client stamps in. Used
1234/// to disambiguate DocumentDB from RDS, which share the `rds` SigV4 scope
1235/// and endpoint. Matches the `api/docdb` token that AWS SDKs emit (a leading
1236/// `api/docdb` optionally followed by `#`/`/` and a version).
1237fn user_agent_indicates_docdb(headers: &http::HeaderMap) -> bool {
1238    for name in ["user-agent", "x-amz-user-agent"] {
1239        if let Some(ua) = headers.get(name).and_then(|v| v.to_str().ok()) {
1240            for part in ua.split_whitespace() {
1241                if let Some(rest) = part.strip_prefix("api/docdb") {
1242                    if rest.is_empty() || rest.starts_with('#') || rest.starts_with('/') {
1243                        return true;
1244                    }
1245                }
1246            }
1247        }
1248    }
1249    false
1250}
1251
1252fn extract_region_from_user_agent(headers: &http::HeaderMap) -> Option<String> {
1253    let ua = headers.get("user-agent")?.to_str().ok()?;
1254    for part in ua.split_whitespace() {
1255        if let Some(region) = part.strip_prefix("region/") {
1256            if !region.is_empty() {
1257                return Some(region.to_string());
1258            }
1259        }
1260    }
1261    None
1262}
1263
1264fn build_error_response(
1265    status: StatusCode,
1266    code: &str,
1267    message: &str,
1268    request_id: &str,
1269    protocol: AwsProtocol,
1270) -> Response<Body> {
1271    build_error_response_with_fields(status, code, message, request_id, protocol, &[])
1272}
1273
1274fn build_error_response_with_fields(
1275    status: StatusCode,
1276    code: &str,
1277    message: &str,
1278    request_id: &str,
1279    protocol: AwsProtocol,
1280    extra_fields: &[(String, String)],
1281) -> Response<Body> {
1282    let (status, content_type, body) = match protocol {
1283        // awsQuery services (SQS, SNS, IAM, STS, RDS, ELBv2, CloudWatch,
1284        // AutoScaling, ...) share the `<ErrorResponse>` envelope.
1285        AwsProtocol::Query => {
1286            fakecloud_aws::error::xml_error_response(status, code, message, request_id)
1287        }
1288        // EC2 uses the distinct ec2Query error envelope
1289        // (`<Response><Errors><Error>...</Errors><RequestID>`). Only EC2 is
1290        // classified `Ec2Query`, so the other Query-protocol services above are
1291        // untouched.
1292        AwsProtocol::Ec2Query => {
1293            fakecloud_aws::ec2query::ec2_error_response(status, code, message, request_id)
1294        }
1295        AwsProtocol::Rest => fakecloud_aws::error::s3_xml_error_response_with_fields(
1296            status,
1297            code,
1298            message,
1299            request_id,
1300            extra_fields,
1301        ),
1302        AwsProtocol::Json | AwsProtocol::RestJson => {
1303            fakecloud_aws::error::json_error_response_with_fields(
1304                status,
1305                code,
1306                message,
1307                extra_fields,
1308            )
1309        }
1310    };
1311
1312    // S3 (and other REST-XML services) place the error code in
1313    // `x-amz-error-code` so HEAD responses — which HTTP forbids from
1314    // carrying a body — still surface the code. AWS SDKs read this header
1315    // when the body is empty. Emit it on every error response so HEAD,
1316    // OPTIONS, and any client that strips the body still see the code.
1317    // Backend errors regularly include newlines (multi-line stderr from
1318    // docker/podman/etc.); HTTP header values reject control characters,
1319    // so sanitize before insertion or the builder rejects the response
1320    // and the connection drops.
1321    let safe_code = sanitize_header_value(code);
1322    let safe_message = sanitize_header_value(message);
1323    let mut builder = Response::builder()
1324        .status(status)
1325        .header("content-type", content_type)
1326        .header("x-amzn-requestid", request_id)
1327        .header("x-amz-request-id", request_id);
1328    if let Ok(v) = http::HeaderValue::from_str(&safe_code) {
1329        builder = builder.header("x-amz-error-code", v);
1330    }
1331    if let Ok(v) = http::HeaderValue::from_str(&safe_message) {
1332        builder = builder.header("x-amz-error-message", v);
1333    }
1334    builder.body(Body::from(body)).unwrap_or_else(|_| {
1335        // Builder only fails if a header is invalid; we sanitized the two
1336        // we control, so the remaining ones (content-type, request id) are
1337        // ASCII and safe. This fallback exists purely so we never panic.
1338        Response::new(Body::empty())
1339    })
1340}
1341
1342/// Strip characters that HTTP header values reject (control bytes, CR/LF/TAB)
1343/// and truncate to a length that AWS SDKs handle cleanly. Backend tools
1344/// (docker, podman, kubectl, …) emit multi-line stderr, and forwarding that
1345/// raw into `x-amz-error-message` previously panicked the dispatcher.
1346fn sanitize_header_value(s: &str) -> String {
1347    const MAX_LEN: usize = 1024;
1348    let mut out = String::with_capacity(s.len().min(MAX_LEN));
1349    for ch in s.chars() {
1350        if out.len() >= MAX_LEN {
1351            break;
1352        }
1353        // Header values forbid CR, LF, and other control bytes (RFC 9110).
1354        // Replace with a single space so multi-line messages stay readable.
1355        if ch.is_control() {
1356            if !out.ends_with(' ') {
1357                out.push(' ');
1358            }
1359        } else {
1360            out.push(ch);
1361        }
1362    }
1363    out.trim().to_string()
1364}
1365
1366/// Build the [`ConditionContext`] passed to the IAM evaluator for one
1367/// request. Populates the 10 global condition keys from the resolved
1368/// principal + the HTTP request. Service-specific keys are deferred to
1369/// a follow-up batch and left empty.
1370/// For an unsigned request that no other detection rule claimed, return the
1371/// bucket name when the first path segment names an existing S3 bucket.
1372///
1373/// fakecloud serves every service from one endpoint, so an anonymous
1374/// path-style S3 request (`GET /bucket/key`, no SigV4) is indistinguishable
1375/// from an API Gateway execute-api call by headers alone. Bucket existence is
1376/// the disambiguator: if the segment is a real bucket, route to S3; otherwise
1377/// fall through to the apigateway catch-all. Uses the already-wired
1378/// `resource_policy_provider`, which resolves S3 bucket ownership from state
1379/// (`Some` => the bucket exists). Returns `None` when no provider is wired.
1380/// Recover the access key from a SigV2 presigned URL. AWS SigV2 presigning
1381/// puts the key in the `AWSAccessKeyId` query parameter alongside `Signature`
1382/// and `Expires`; all three must be present for the URL to be a SigV2 presign.
1383/// Returns None for SigV4 presigns (which use `X-Amz-Credential`) or unsigned
1384/// requests.
1385fn sigv2_presigned_access_key(query_params: &HashMap<String, String>) -> Option<String> {
1386    if query_params.contains_key("Signature") && query_params.contains_key("Expires") {
1387        query_params.get("AWSAccessKeyId").cloned()
1388    } else {
1389        None
1390    }
1391}
1392
1393/// True when `s` is a 64-character lowercase-hex SHA-256 digest — the form a
1394/// signed `x-amz-content-sha256` payload hash takes. Distinguishes a real body
1395/// hash from the `UNSIGNED-PAYLOAD` / `STREAMING-*` markers (shorter and
1396/// containing non-hex characters), so only genuine hashes are re-bound to the
1397/// buffered body.
1398fn is_hex_sha256(s: &str) -> bool {
1399    s.len() == 64 && s.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'))
1400}
1401
1402/// Lowercase-hex SHA-256 of `bytes`, used to re-bind a signed
1403/// `x-amz-content-sha256` to the buffered request body for non-S3 services.
1404fn sha256_hex_lower(bytes: &[u8]) -> String {
1405    use sha2::{Digest, Sha256};
1406    let digest = Sha256::digest(bytes);
1407    const HEX: &[u8] = b"0123456789abcdef";
1408    let mut out = String::with_capacity(64);
1409    for b in digest {
1410        out.push(HEX[(b >> 4) as usize] as char);
1411        out.push(HEX[(b & 0x0f) as usize] as char);
1412    }
1413    out
1414}
1415
1416fn anonymous_s3_bucket(uri: &http::Uri, config: &DispatchConfig) -> Option<String> {
1417    let provider = config.resource_policy_provider.as_ref()?;
1418    let segment = uri.path().split('/').find(|s| !s.is_empty())?.to_string();
1419    let arn = format!("arn:aws:s3:::{segment}");
1420    provider.resource_owner_account("s3", &arn).map(|_| segment)
1421}
1422
1423fn build_condition_context(
1424    principal: &Principal,
1425    remote_addr: Option<SocketAddr>,
1426    region: &str,
1427    secure_transport: bool,
1428) -> ConditionContext {
1429    let now = chrono::Utc::now();
1430    ConditionContext {
1431        aws_username: aws_username_from_principal(principal),
1432        aws_userid: Some(principal.user_id.clone()),
1433        aws_principal_arn: Some(principal.arn.clone()),
1434        aws_principal_account: Some(principal.account_id.clone()),
1435        aws_principal_type: Some(principal_type_label(principal.principal_type).to_string()),
1436        aws_source_ip: remote_addr.map(|sa| sa.ip()),
1437        aws_current_time: Some(now),
1438        aws_epoch_time: Some(now.timestamp()),
1439        aws_secure_transport: Some(secure_transport),
1440        aws_requested_region: Some(region.to_string()),
1441        // F3 keys: populated from the caller's session context when STS
1442        // mints credentials with MFA / SAML / OIDC / VPC-endpoint hints.
1443        // Default-None here so tests/dispatch sites that don't set them
1444        // safe-fail any policy referencing them — matching AWS for keys
1445        // that aren't asserted.
1446        aws_mfa_present: None,
1447        aws_mfa_age_seconds: None,
1448        aws_called_via: Vec::new(),
1449        aws_source_vpce: None,
1450        aws_source_vpc: None,
1451        aws_vpc_source_ip: None,
1452        aws_federated_provider: None,
1453        aws_token_issue_time: None,
1454        service_keys: Default::default(),
1455        resource_tags: None,
1456        request_tags: None,
1457        principal_tags: None,
1458    }
1459}
1460
1461/// `aws:username` is only set for IAM users, matching AWS. For assumed
1462/// roles, federated users, root, and unknown principals the key is
1463/// absent — operators that reference it without `IfExists` safe-fail.
1464fn aws_username_from_principal(principal: &Principal) -> Option<String> {
1465    if principal.principal_type != PrincipalType::User {
1466        return None;
1467    }
1468    let after = principal.arn.rsplit_once(":user/").map(|(_, s)| s)?;
1469    // Strip any IAM path prefix; bare username is the last segment.
1470    Some(after.rsplit('/').next().unwrap_or(after).to_string())
1471}
1472
1473/// AWS's `aws:PrincipalType` uses PascalCase identifiers, distinct from
1474/// the lowercase ones [`PrincipalType::as_str`] returns for ARNs.
1475fn principal_type_label(t: PrincipalType) -> &'static str {
1476    match t {
1477        PrincipalType::User => "User",
1478        PrincipalType::AssumedRole => "AssumedRole",
1479        PrincipalType::FederatedUser => "FederatedUser",
1480        PrincipalType::Root => "Account",
1481        PrincipalType::Unknown => "Unknown",
1482    }
1483}
1484
1485/// Best-effort detection of TLS-terminated requests. Direct HTTPS
1486/// connections are not yet supported by the fakecloud server (it speaks
1487/// plain HTTP), so the only signal is an `x-forwarded-proto: https`
1488/// header set by an upstream proxy. Anything else evaluates to `false`,
1489/// which matches the typical local-dev setup.
1490fn is_secure_transport(headers: &http::HeaderMap) -> bool {
1491    headers
1492        .get("x-forwarded-proto")
1493        .and_then(|v| v.to_str().ok())
1494        .map(|s| s.eq_ignore_ascii_case("https"))
1495        .unwrap_or(false)
1496}
1497
1498trait ProtocolExt {
1499    fn error_status(&self) -> StatusCode;
1500}
1501
1502impl ProtocolExt for AwsProtocol {
1503    fn error_status(&self) -> StatusCode {
1504        StatusCode::BAD_REQUEST
1505    }
1506}
1507
1508#[cfg(test)]
1509mod tests {
1510    use super::*;
1511
1512    #[test]
1513    fn default_max_request_body_bytes_is_one_gib() {
1514        // Without the env override, the cap defaults to 1 GiB. The
1515        // public function caches via OnceLock so only the first call
1516        // in the process matters; we assert the constant directly.
1517        assert_eq!(DEFAULT_MAX_REQUEST_BODY_BYTES, 1024 * 1024 * 1024);
1518    }
1519
1520    #[test]
1521    fn sigv2_presigned_access_key_extracted_with_signature_and_expires() {
1522        let mut q = HashMap::new();
1523        q.insert("AWSAccessKeyId".to_string(), "AKIAEXAMPLE".to_string());
1524        q.insert("Signature".to_string(), "abc%2Bdef".to_string());
1525        q.insert("Expires".to_string(), "1700000000".to_string());
1526        assert_eq!(
1527            sigv2_presigned_access_key(&q).as_deref(),
1528            Some("AKIAEXAMPLE")
1529        );
1530    }
1531
1532    #[test]
1533    fn sigv2_presigned_access_key_none_without_signature_or_expires() {
1534        // AWSAccessKeyId alone (e.g. a stray query param) is not a SigV2
1535        // presign and must not be treated as a credential.
1536        let mut q = HashMap::new();
1537        q.insert("AWSAccessKeyId".to_string(), "AKIAEXAMPLE".to_string());
1538        assert_eq!(sigv2_presigned_access_key(&q), None);
1539
1540        q.insert("Expires".to_string(), "1700000000".to_string());
1541        assert_eq!(
1542            sigv2_presigned_access_key(&q),
1543            None,
1544            "missing Signature must not qualify"
1545        );
1546    }
1547
1548    #[test]
1549    fn sigv2_presigned_access_key_none_for_unsigned_request() {
1550        assert_eq!(sigv2_presigned_access_key(&HashMap::new()), None);
1551    }
1552
1553    #[test]
1554    fn is_hex_sha256_accepts_real_digest_rejects_markers() {
1555        // A genuine 64-char lowercase-hex digest is a bindable body hash.
1556        assert!(is_hex_sha256(&sha256_hex_lower(b"hello")));
1557        assert!(is_hex_sha256(
1558            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
1559        ));
1560        // The SigV4 payload markers are NOT hashes and must be skipped.
1561        assert!(!is_hex_sha256("UNSIGNED-PAYLOAD"));
1562        assert!(!is_hex_sha256("STREAMING-AWS4-HMAC-SHA256-PAYLOAD"));
1563        assert!(!is_hex_sha256("STREAMING-UNSIGNED-PAYLOAD-TRAILER"));
1564        // Wrong length or non-lowercase-hex characters are rejected.
1565        assert!(!is_hex_sha256("abc123"));
1566        assert!(!is_hex_sha256(
1567            "E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855"
1568        ));
1569    }
1570
1571    #[test]
1572    fn sha256_hex_lower_matches_known_vectors() {
1573        // Empty input -> the well-known SHA-256 of "".
1574        assert_eq!(
1575            sha256_hex_lower(b""),
1576            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
1577        );
1578        assert_eq!(
1579            sha256_hex_lower(b"abc"),
1580            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
1581        );
1582        assert_eq!(sha256_hex_lower(b"abc").len(), 64);
1583    }
1584
1585    #[test]
1586    fn dispatch_config_new_defaults_to_off() {
1587        let cfg = DispatchConfig::new("us-east-1", "123456789012");
1588        assert_eq!(cfg.region, "us-east-1");
1589        assert_eq!(cfg.account_id, "123456789012");
1590        assert!(!cfg.verify_sigv4);
1591        assert_eq!(cfg.iam_mode, IamMode::Off);
1592    }
1593
1594    #[test]
1595    fn aws_username_strips_iam_path_for_users() {
1596        let p = Principal {
1597            arn: "arn:aws:iam::123456789012:user/engineering/alice".into(),
1598            user_id: "AIDAALICE".into(),
1599            account_id: "123456789012".into(),
1600            principal_type: PrincipalType::User,
1601            source_identity: None,
1602            tags: None,
1603        };
1604        assert_eq!(aws_username_from_principal(&p), Some("alice".into()));
1605    }
1606
1607    #[test]
1608    fn aws_username_unset_for_assumed_role() {
1609        let p = Principal {
1610            arn: "arn:aws:sts::123456789012:assumed-role/ops/session".into(),
1611            user_id: "AROAOPS:session".into(),
1612            account_id: "123456789012".into(),
1613            principal_type: PrincipalType::AssumedRole,
1614            source_identity: None,
1615            tags: None,
1616        };
1617        assert_eq!(aws_username_from_principal(&p), None);
1618    }
1619
1620    #[test]
1621    fn principal_type_label_matches_aws_casing() {
1622        assert_eq!(principal_type_label(PrincipalType::User), "User");
1623        assert_eq!(
1624            principal_type_label(PrincipalType::AssumedRole),
1625            "AssumedRole"
1626        );
1627        assert_eq!(principal_type_label(PrincipalType::Root), "Account");
1628    }
1629
1630    #[test]
1631    fn build_condition_context_populates_global_keys() {
1632        let p = Principal {
1633            arn: "arn:aws:iam::123456789012:user/alice".into(),
1634            user_id: "AIDAALICE".into(),
1635            account_id: "123456789012".into(),
1636            principal_type: PrincipalType::User,
1637            source_identity: None,
1638            tags: None,
1639        };
1640        let addr: SocketAddr = "10.0.0.1:54321".parse().unwrap();
1641        let ctx = build_condition_context(&p, Some(addr), "us-east-1", false);
1642        assert_eq!(ctx.aws_username.as_deref(), Some("alice"));
1643        assert_eq!(ctx.aws_userid.as_deref(), Some("AIDAALICE"));
1644        assert_eq!(
1645            ctx.aws_principal_arn.as_deref(),
1646            Some("arn:aws:iam::123456789012:user/alice")
1647        );
1648        assert_eq!(ctx.aws_principal_account.as_deref(), Some("123456789012"));
1649        assert_eq!(ctx.aws_principal_type.as_deref(), Some("User"));
1650        assert_eq!(
1651            ctx.aws_source_ip.map(|i| i.to_string()).as_deref(),
1652            Some("10.0.0.1")
1653        );
1654        assert_eq!(ctx.aws_requested_region.as_deref(), Some("us-east-1"));
1655        assert_eq!(ctx.aws_secure_transport, Some(false));
1656        assert!(ctx.aws_current_time.is_some());
1657        assert!(ctx.aws_epoch_time.is_some());
1658    }
1659
1660    #[test]
1661    fn is_secure_transport_reads_x_forwarded_proto() {
1662        let mut headers = http::HeaderMap::new();
1663        headers.insert("x-forwarded-proto", "https".parse().unwrap());
1664        assert!(is_secure_transport(&headers));
1665        headers.insert("x-forwarded-proto", "http".parse().unwrap());
1666        assert!(!is_secure_transport(&headers));
1667        let empty = http::HeaderMap::new();
1668        assert!(!is_secure_transport(&empty));
1669    }
1670
1671    #[test]
1672    fn parse_account_from_arn_extracts_standard_shapes() {
1673        assert_eq!(
1674            parse_account_from_arn("arn:aws:sqs:us-east-1:123456789012:queue"),
1675            Some("123456789012".to_string())
1676        );
1677        assert_eq!(
1678            parse_account_from_arn("arn:aws:iam::123456789012:user/alice"),
1679            Some("123456789012".to_string())
1680        );
1681    }
1682
1683    #[test]
1684    fn parse_account_from_arn_returns_none_for_s3_empty_account() {
1685        // S3 ARNs have both region and account empty.
1686        assert_eq!(parse_account_from_arn("arn:aws:s3:::my-bucket"), None);
1687        assert_eq!(
1688            parse_account_from_arn("arn:aws:s3:::my-bucket/path/to/key"),
1689            None
1690        );
1691    }
1692
1693    #[test]
1694    fn parse_account_from_arn_returns_none_for_malformed() {
1695        assert_eq!(parse_account_from_arn(""), None);
1696        assert_eq!(parse_account_from_arn("not-an-arn"), None);
1697        assert_eq!(parse_account_from_arn("arn:aws:sqs:us-east-1"), None);
1698        assert_eq!(parse_account_from_arn("arn:aws:sqs"), None);
1699    }
1700
1701    #[test]
1702    fn extract_region_from_user_agent_finds_region_segment() {
1703        let mut headers = http::HeaderMap::new();
1704        headers.insert(
1705            "user-agent",
1706            "aws-sdk-rust/1.0 os/linux region/eu-central-1"
1707                .parse()
1708                .unwrap(),
1709        );
1710        assert_eq!(
1711            extract_region_from_user_agent(&headers),
1712            Some("eu-central-1".to_string())
1713        );
1714    }
1715
1716    #[test]
1717    fn extract_region_from_user_agent_none_without_header() {
1718        let headers = http::HeaderMap::new();
1719        assert_eq!(extract_region_from_user_agent(&headers), None);
1720    }
1721
1722    #[test]
1723    fn extract_region_from_user_agent_ignores_empty_region() {
1724        let mut headers = http::HeaderMap::new();
1725        headers.insert("user-agent", "aws-sdk-java region/".parse().unwrap());
1726        assert_eq!(extract_region_from_user_agent(&headers), None);
1727    }
1728
1729    #[test]
1730    fn extract_region_from_user_agent_none_when_no_region_marker() {
1731        let mut headers = http::HeaderMap::new();
1732        headers.insert("user-agent", "curl/7.79.1".parse().unwrap());
1733        assert_eq!(extract_region_from_user_agent(&headers), None);
1734    }
1735
1736    #[test]
1737    fn aws_username_none_for_root() {
1738        let p = Principal {
1739            arn: "arn:aws:iam::123456789012:root".into(),
1740            user_id: "123456789012".into(),
1741            account_id: "123456789012".into(),
1742            principal_type: PrincipalType::Root,
1743            source_identity: None,
1744            tags: None,
1745        };
1746        assert_eq!(aws_username_from_principal(&p), None);
1747    }
1748
1749    #[test]
1750    fn aws_username_bare_no_path() {
1751        let p = Principal {
1752            arn: "arn:aws:iam::123456789012:user/bob".into(),
1753            user_id: "AIDABOB".into(),
1754            account_id: "123456789012".into(),
1755            principal_type: PrincipalType::User,
1756            source_identity: None,
1757            tags: None,
1758        };
1759        assert_eq!(aws_username_from_principal(&p), Some("bob".into()));
1760    }
1761
1762    #[test]
1763    fn principal_type_label_covers_federated_and_unknown() {
1764        assert_eq!(
1765            principal_type_label(PrincipalType::FederatedUser),
1766            "FederatedUser"
1767        );
1768        assert_eq!(principal_type_label(PrincipalType::Unknown), "Unknown");
1769    }
1770
1771    #[test]
1772    fn build_condition_context_marks_secure_when_flag_set() {
1773        let p = Principal {
1774            arn: "arn:aws:iam::123456789012:user/alice".into(),
1775            user_id: "AIDAALICE".into(),
1776            account_id: "123456789012".into(),
1777            principal_type: PrincipalType::User,
1778            source_identity: None,
1779            tags: None,
1780        };
1781        let ctx = build_condition_context(&p, None, "us-west-2", true);
1782        assert_eq!(ctx.aws_secure_transport, Some(true));
1783        assert!(ctx.aws_source_ip.is_none());
1784        assert_eq!(ctx.aws_requested_region.as_deref(), Some("us-west-2"));
1785    }
1786
1787    #[test]
1788    fn is_secure_transport_case_insensitive() {
1789        let mut headers = http::HeaderMap::new();
1790        headers.insert("x-forwarded-proto", "HTTPS".parse().unwrap());
1791        assert!(is_secure_transport(&headers));
1792    }
1793
1794    #[test]
1795    fn is_secure_transport_non_ascii_bytes_false() {
1796        let mut headers = http::HeaderMap::new();
1797        headers.insert(
1798            "x-forwarded-proto",
1799            http::HeaderValue::from_bytes(&[0xFF, 0xFE]).unwrap(),
1800        );
1801        assert!(!is_secure_transport(&headers));
1802    }
1803
1804    #[test]
1805    fn protocol_ext_error_status_is_bad_request() {
1806        assert_eq!(AwsProtocol::Query.error_status(), StatusCode::BAD_REQUEST);
1807        assert_eq!(AwsProtocol::Json.error_status(), StatusCode::BAD_REQUEST);
1808        assert_eq!(AwsProtocol::Rest.error_status(), StatusCode::BAD_REQUEST);
1809        assert_eq!(
1810            AwsProtocol::RestJson.error_status(),
1811            StatusCode::BAD_REQUEST
1812        );
1813    }
1814
1815    #[test]
1816    fn build_error_response_json_has_json_content_type() {
1817        let resp = build_error_response(
1818            StatusCode::BAD_REQUEST,
1819            "TestCode",
1820            "test msg",
1821            "req-1",
1822            AwsProtocol::Json,
1823        );
1824        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1825        let ct = resp
1826            .headers()
1827            .get("content-type")
1828            .unwrap()
1829            .to_str()
1830            .unwrap();
1831        assert!(ct.contains("json"));
1832        let rid = resp
1833            .headers()
1834            .get("x-amzn-requestid")
1835            .unwrap()
1836            .to_str()
1837            .unwrap();
1838        assert_eq!(rid, "req-1");
1839    }
1840
1841    #[test]
1842    fn build_error_response_rest_returns_xml_content_type() {
1843        let resp = build_error_response(
1844            StatusCode::NOT_FOUND,
1845            "NoSuchBucket",
1846            "bucket missing",
1847            "req-2",
1848            AwsProtocol::Rest,
1849        );
1850        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1851        let ct = resp
1852            .headers()
1853            .get("content-type")
1854            .unwrap()
1855            .to_str()
1856            .unwrap();
1857        assert!(ct.contains("xml"));
1858    }
1859
1860    #[test]
1861    fn build_error_response_query_returns_xml() {
1862        let resp = build_error_response(
1863            StatusCode::BAD_REQUEST,
1864            "InvalidParameter",
1865            "bad param",
1866            "req-3",
1867            AwsProtocol::Query,
1868        );
1869        let ct = resp
1870            .headers()
1871            .get("content-type")
1872            .unwrap()
1873            .to_str()
1874            .unwrap();
1875        assert!(ct.contains("xml"));
1876    }
1877
1878    /// Regression for issue #1539: multi-line backend errors (e.g. podman
1879    /// stderr) used to panic the dispatcher when stuffed into the
1880    /// `x-amz-error-message` HTTP header. The response must build cleanly
1881    /// and the header value must not contain control characters.
1882    #[test]
1883    fn build_error_response_with_multiline_message_does_not_panic() {
1884        let resp = build_error_response(
1885            StatusCode::INTERNAL_SERVER_ERROR,
1886            "ServiceException",
1887            "Lambda execution failed: container failed to start: docker start failed: \
1888             Error: unable to start container \"abc\": \
1889             failed to create new hosts file:\nhost-gateway is empty\n",
1890            "req-multi",
1891            AwsProtocol::Json,
1892        );
1893        assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
1894        let msg = resp
1895            .headers()
1896            .get("x-amz-error-message")
1897            .expect("x-amz-error-message must be set even when input contains newlines")
1898            .to_str()
1899            .unwrap();
1900        assert!(!msg.contains('\n'));
1901        assert!(!msg.contains('\r'));
1902        assert!(msg.contains("Lambda execution failed"));
1903        assert!(msg.contains("host-gateway is empty"));
1904    }
1905
1906    #[test]
1907    fn build_error_response_with_control_chars_strips_them() {
1908        let resp = build_error_response(
1909            StatusCode::BAD_REQUEST,
1910            "Code\twith\ttabs",
1911            "msg\x00with\x01nulls",
1912            "req-ctrl",
1913            AwsProtocol::Json,
1914        );
1915        let code = resp
1916            .headers()
1917            .get("x-amz-error-code")
1918            .unwrap()
1919            .to_str()
1920            .unwrap();
1921        let msg = resp
1922            .headers()
1923            .get("x-amz-error-message")
1924            .unwrap()
1925            .to_str()
1926            .unwrap();
1927        assert!(!code.contains('\t'));
1928        assert!(!msg.contains('\x00'));
1929        assert!(!msg.contains('\x01'));
1930    }
1931
1932    #[test]
1933    fn sanitize_header_value_truncates_long_input() {
1934        let huge = "x".repeat(5_000);
1935        let out = sanitize_header_value(&huge);
1936        assert!(out.len() <= 1024);
1937    }
1938
1939    #[test]
1940    fn sanitize_header_value_collapses_consecutive_control_runs() {
1941        let out = sanitize_header_value("a\n\n\n\rb");
1942        assert_eq!(out, "a b");
1943    }
1944
1945    #[test]
1946    fn dispatch_config_carries_opt_in_flags() {
1947        let cfg = DispatchConfig {
1948            region: "eu-west-1".to_string(),
1949            account_id: "000000000000".to_string(),
1950            verify_sigv4: true,
1951            iam_mode: IamMode::Strict,
1952            credential_resolver: None,
1953            policy_evaluator: None,
1954            resource_policy_provider: None,
1955            scp_resolver: None,
1956        };
1957        assert!(cfg.verify_sigv4);
1958        assert!(cfg.iam_mode.is_strict());
1959        assert!(cfg.resource_policy_provider.is_none());
1960        assert!(cfg.scp_resolver.is_none());
1961    }
1962
1963    fn s3_sigv4_headers() -> http::HeaderMap {
1964        let mut headers = http::HeaderMap::new();
1965        headers.insert(
1966            "authorization",
1967            "AWS4-HMAC-SHA256 Credential=test/20240101/us-east-1/s3/aws4_request, \
1968             SignedHeaders=host, Signature=fake"
1969                .parse()
1970                .unwrap(),
1971        );
1972        headers
1973    }
1974
1975    #[test]
1976    fn streaming_route_path_style_s3_put_object() {
1977        let headers = s3_sigv4_headers();
1978        assert_eq!(
1979            streaming_route(
1980                &http::Method::PUT,
1981                "/my-bucket/key.txt",
1982                &headers,
1983                &HashMap::new(),
1984            ),
1985            Some(("s3", "")),
1986        );
1987    }
1988
1989    #[test]
1990    fn streaming_route_path_style_create_bucket_skipped() {
1991        // `PUT /bucket` (no trailing key) is CreateBucket — must NOT
1992        // hit the streaming path.
1993        let headers = s3_sigv4_headers();
1994        assert_eq!(
1995            streaming_route(&http::Method::PUT, "/my-bucket", &headers, &HashMap::new(),),
1996            None,
1997        );
1998    }
1999
2000    #[test]
2001    fn streaming_route_virtual_hosted_s3_put_object() {
2002        let mut headers = s3_sigv4_headers();
2003        headers.insert(
2004            "host",
2005            "vhost-bucket.s3.us-east-1.localhost.localstack.cloud:4566"
2006                .parse()
2007                .unwrap(),
2008        );
2009        // Virtual-hosted PUT has no bucket in the URL path (`/<key>`),
2010        // so the slash check used for path-style would reject it. The
2011        // Host parser confirms this is virtual-hosted S3 and the key
2012        // flows through the streaming dispatch.
2013        assert_eq!(
2014            streaming_route(&http::Method::PUT, "/hello.txt", &headers, &HashMap::new(),),
2015            Some(("s3", "")),
2016        );
2017    }
2018
2019    #[test]
2020    fn streaming_route_virtual_hosted_s3_root_skipped() {
2021        // `PUT /` against a virtual-hosted Host = CreateBucket, which
2022        // is handled buffered. Empty path-after-slash must short-circuit.
2023        let mut headers = s3_sigv4_headers();
2024        headers.insert(
2025            "host",
2026            "vhost-bucket.s3.us-east-1.localhost.localstack.cloud:4566"
2027                .parse()
2028                .unwrap(),
2029        );
2030        assert_eq!(
2031            streaming_route(&http::Method::PUT, "/", &headers, &HashMap::new()),
2032            None,
2033        );
2034    }
2035
2036    #[test]
2037    fn streaming_route_ecr_blob_upload() {
2038        let headers = http::HeaderMap::new();
2039        assert_eq!(
2040            streaming_route(
2041                &http::Method::PATCH,
2042                "/v2/my-repo/blobs/uploads/abcd1234",
2043                &headers,
2044                &HashMap::new(),
2045            ),
2046            Some(("ecr", "")),
2047        );
2048        assert_eq!(
2049            streaming_route(
2050                &http::Method::PUT,
2051                "/v2/my-repo/blobs/uploads/abcd1234",
2052                &headers,
2053                &HashMap::new(),
2054            ),
2055            Some(("ecr", "")),
2056        );
2057    }
2058
2059    #[test]
2060    fn streaming_route_presigned_v4_s3_put() {
2061        let headers = http::HeaderMap::new();
2062        let mut query_params = HashMap::new();
2063        query_params.insert(
2064            "X-Amz-Credential".to_string(),
2065            "test/20240101/us-east-1/s3/aws4_request".to_string(),
2066        );
2067        assert_eq!(
2068            streaming_route(
2069                &http::Method::PUT,
2070                "/my-bucket/key.txt",
2071                &headers,
2072                &query_params,
2073            ),
2074            Some(("s3", "")),
2075        );
2076    }
2077
2078    #[test]
2079    fn streaming_route_non_s3_auth_header_skipped() {
2080        // Same path shape but the SigV4 service is lambda — must not
2081        // wire the streaming dispatch (Lambda has its own buffered path).
2082        let mut headers = http::HeaderMap::new();
2083        headers.insert(
2084            "authorization",
2085            "AWS4-HMAC-SHA256 Credential=test/20240101/us-east-1/lambda/aws4_request, \
2086             SignedHeaders=host, Signature=fake"
2087                .parse()
2088                .unwrap(),
2089        );
2090        assert_eq!(
2091            streaming_route(
2092                &http::Method::PUT,
2093                "/my-bucket/key.txt",
2094                &headers,
2095                &HashMap::new(),
2096            ),
2097            None,
2098        );
2099    }
2100
2101    #[test]
2102    fn streaming_route_get_skipped() {
2103        let headers = s3_sigv4_headers();
2104        assert_eq!(
2105            streaming_route(
2106                &http::Method::GET,
2107                "/my-bucket/key.txt",
2108                &headers,
2109                &HashMap::new(),
2110            ),
2111            None,
2112        );
2113    }
2114}