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    if config.verify_sigv4 && !is_root_bypass(caller_akid) && config.credential_resolver.is_some() {
328        let amz_date = parts
329            .headers
330            .get("x-amz-date")
331            .and_then(|v| v.to_str().ok());
332        let parsed = fakecloud_aws::sigv4::parse_sigv4_header(auth_header, amz_date)
333            .or_else(|| fakecloud_aws::sigv4::parse_sigv4_presigned(&query_params));
334        let parsed = match parsed {
335            Some(p) => p,
336            None => {
337                return build_error_response(
338                    StatusCode::FORBIDDEN,
339                    "IncompleteSignature",
340                    "Request is missing or has a malformed AWS Signature",
341                    &request_id,
342                    detected.protocol,
343                );
344            }
345        };
346        let resolved_for_verify = match resolved.as_ref() {
347            Some(r) => r,
348            None => {
349                return build_error_response(
350                    StatusCode::FORBIDDEN,
351                    "InvalidClientTokenId",
352                    "The security token included in the request is invalid",
353                    &request_id,
354                    detected.protocol,
355                );
356            }
357        };
358        let headers_vec = fakecloud_aws::sigv4::headers_from_http(&parts.headers);
359        let raw_query_for_verify = parts.uri.query().unwrap_or("").to_string();
360        let verify_req = fakecloud_aws::sigv4::VerifyRequest {
361            method: parts.method.as_str(),
362            path: parts.uri.path(),
363            query: &raw_query_for_verify,
364            headers: &headers_vec,
365            body: &body_bytes,
366        };
367        match fakecloud_aws::sigv4::verify(
368            &parsed,
369            &verify_req,
370            &resolved_for_verify.secret_access_key,
371            chrono::Utc::now(),
372        ) {
373            Ok(()) => {}
374            Err(fakecloud_aws::sigv4::SigV4Error::RequestTimeTooSkewed { .. }) => {
375                return build_error_response(
376                    StatusCode::FORBIDDEN,
377                    "RequestTimeTooSkewed",
378                    "The difference between the request time and the current time is too large",
379                    &request_id,
380                    detected.protocol,
381                );
382            }
383            Err(fakecloud_aws::sigv4::SigV4Error::InvalidDate(msg)) => {
384                return build_error_response(
385                    StatusCode::FORBIDDEN,
386                    "IncompleteSignature",
387                    &format!("Invalid x-amz-date: {msg}"),
388                    &request_id,
389                    detected.protocol,
390                );
391            }
392            Err(fakecloud_aws::sigv4::SigV4Error::Malformed(msg)) => {
393                return build_error_response(
394                    StatusCode::FORBIDDEN,
395                    "IncompleteSignature",
396                    &format!("Malformed SigV4 signature: {msg}"),
397                    &request_id,
398                    detected.protocol,
399                );
400            }
401            Err(fakecloud_aws::sigv4::SigV4Error::SignatureMismatch) => {
402                return build_error_response(
403                    StatusCode::FORBIDDEN,
404                    "SignatureDoesNotMatch",
405                    "The request signature we calculated does not match the signature you provided",
406                    &request_id,
407                    detected.protocol,
408                );
409            }
410            Err(fakecloud_aws::sigv4::SigV4Error::PresignedUrlExpired { .. }) => {
411                return build_error_response(
412                    StatusCode::FORBIDDEN,
413                    "AccessDenied",
414                    "Request has expired",
415                    &request_id,
416                    detected.protocol,
417                );
418            }
419            Err(fakecloud_aws::sigv4::SigV4Error::InvalidPresignExpires(_)) => {
420                return build_error_response(
421                    StatusCode::BAD_REQUEST,
422                    "AuthorizationQueryParametersError",
423                    "X-Amz-Expires must be a number between 1 and 604800 seconds",
424                    &request_id,
425                    detected.protocol,
426                );
427            }
428        }
429    }
430
431    // Build path segments. For S3 virtual-hosted-style requests the bucket
432    // lives in the Host header, not the path — prepend it so the S3 handler
433    // sees a uniform path-style request. SigV4 verification above already
434    // ran against the wire path, so this rewrite is signature-safe.
435    let wire_path = parts.uri.path();
436    let path = if detected.service == "s3" {
437        if let Some(bucket) = host_info.as_ref().and_then(|h| h.bucket.as_deref()) {
438            let prefix_with_slash = format!("/{bucket}/");
439            let is_bucket_root = wire_path.trim_end_matches('/') == format!("/{bucket}");
440            if wire_path.starts_with(&prefix_with_slash) || is_bucket_root {
441                wire_path.to_string()
442            } else if wire_path == "/" || wire_path.is_empty() {
443                format!("/{bucket}")
444            } else {
445                format!("/{bucket}{wire_path}")
446            }
447        } else {
448            wire_path.to_string()
449        }
450    } else {
451        wire_path.to_string()
452    };
453    let raw_query = parts.uri.query().unwrap_or("").to_string();
454    let path_segments: Vec<String> = path
455        .split('/')
456        .filter(|s| !s.is_empty())
457        .map(|s| s.to_string())
458        .collect();
459
460    // For JSON protocol, validate that non-empty bodies are valid JSON
461    if detected.protocol == AwsProtocol::Json
462        && !body_bytes.is_empty()
463        && serde_json::from_slice::<serde_json::Value>(&body_bytes).is_err()
464    {
465        return build_error_response(
466            StatusCode::BAD_REQUEST,
467            "SerializationException",
468            "Start of structure or map found where not expected",
469            &request_id,
470            AwsProtocol::Json,
471        );
472    }
473
474    // Merge query params with form body params for Query protocol
475    let mut all_params = query_params;
476    if detected.protocol == AwsProtocol::Query {
477        let body_params = protocol::parse_query_body(&body_bytes);
478        for (k, v) in body_params {
479            all_params.entry(k).or_insert(v);
480        }
481    }
482
483    // CloudWatch (`monitoring`) advertises awsJson1_0 alongside awsQuery. Its
484    // handlers all read the flat awsQuery param map, so when a client uses the
485    // JSON protocol we flatten the JSON body into that same map, leaving the
486    // handlers unchanged. The handler emits a JSON response for JSON callers.
487    if detected.protocol == AwsProtocol::Json && detected.service == "monitoring" {
488        let body_params = protocol::flatten_json_to_query(&body_bytes);
489        for (k, v) in body_params {
490            all_params.entry(k).or_insert(v);
491        }
492    }
493
494    let aws_request = AwsRequest {
495        service: detected.service.clone(),
496        action: detected.action.clone(),
497        region,
498        account_id: caller_principal
499            .as_ref()
500            .map(|p| p.account_id.clone())
501            .unwrap_or_else(|| config.account_id.clone()),
502        request_id: request_id.clone(),
503        headers: parts.headers,
504        query_params: all_params,
505        body: body_bytes,
506        body_stream: parking_lot::Mutex::new(body_stream),
507        path_segments,
508        raw_path: path,
509        raw_query,
510        method: parts.method,
511        is_query_protocol: detected.protocol == AwsProtocol::Query,
512        access_key_id,
513        principal: caller_principal,
514    };
515
516    tracing::info!(
517        service = %aws_request.service,
518        action = %aws_request.action,
519        request_id = %aws_request.request_id,
520        "handling request"
521    );
522
523    // Opt-in IAM identity-policy enforcement. Runs before the service
524    // handler so a deny never reaches business logic. Root principals
525    // (both `test*` bypass AKIDs and the account's IAM root) are exempt,
526    // matching AWS behavior. Services that haven't opted in via
527    // `iam_enforceable()` are transparently skipped — the startup log
528    // lists which services are under enforcement so users always know.
529    if config.iam_mode.is_enabled()
530        && service.iam_enforceable()
531        && !is_root_bypass(aws_request.access_key_id.as_deref().unwrap_or(""))
532    {
533        if let Some(evaluator) = config.policy_evaluator.as_ref() {
534            if let Some(principal) = aws_request.principal.as_ref() {
535                if !principal.is_root() {
536                    if let Some(iam_action) = service.iam_action_for(&aws_request) {
537                        let mut condition_context = build_condition_context(
538                            principal,
539                            remote_addr,
540                            &aws_request.region,
541                            is_secure_transport(&aws_request.headers),
542                        );
543                        // F3 keys riding on the resolved credential. STS
544                        // populates these at mint time so subsequent
545                        // requests under the credential can be evaluated
546                        // against `aws:MultiFactorAuthPresent`,
547                        // `aws:MultiFactorAuthAge`, `aws:TokenIssueTime`,
548                        // and `aws:FederatedProvider`. IAM user access
549                        // keys carry none of these, matching AWS.
550                        if let Some(rc) = resolved.as_ref() {
551                            condition_context.aws_mfa_present = Some(rc.mfa_present);
552                            condition_context.aws_token_issue_time = rc.token_issued_at;
553                            condition_context.aws_federated_provider =
554                                rc.federated_provider.clone();
555                            // `aws:MultiFactorAuthAge` is "seconds since
556                            // MFA was asserted" — computed at evaluation
557                            // time from the token issue moment so the
558                            // value increases monotonically as the session
559                            // ages. Only set when the session was actually
560                            // minted with MFA; otherwise the key is
561                            // absent, matching AWS.
562                            if rc.mfa_present {
563                                if let Some(issued) = rc.token_issued_at {
564                                    let age = chrono::Utc::now()
565                                        .signed_duration_since(issued)
566                                        .num_seconds()
567                                        .max(0);
568                                    condition_context.aws_mfa_age_seconds = Some(age);
569                                }
570                            }
571                        }
572                        condition_context.service_keys =
573                            service.iam_condition_keys_for(&aws_request, &iam_action);
574
575                        // ABAC: populate tag-based condition keys.
576                        // aws:ResourceTag/*
577                        match service.resource_tags_for(&iam_action.resource) {
578                            Some(tags) => condition_context.resource_tags = Some(tags),
579                            None => tracing::debug!(
580                                target: "fakecloud::iam::audit",
581                                service = %detected.service,
582                                resource = %iam_action.resource,
583                                "service does not expose resource tags for ABAC; skipping aws:ResourceTag/* evaluation"
584                            ),
585                        }
586                        // aws:RequestTag/* + aws:TagKeys
587                        match service.request_tags_from(&aws_request, iam_action.action) {
588                            Some(tags) => condition_context.request_tags = Some(tags),
589                            None => tracing::debug!(
590                                target: "fakecloud::iam::audit",
591                                service = %detected.service,
592                                action = %iam_action.action_string(),
593                                "service does not expose request tags for ABAC; skipping aws:RequestTag/* / aws:TagKeys evaluation"
594                            ),
595                        }
596                        // aws:PrincipalTag/*
597                        condition_context.principal_tags = principal.tags.clone();
598
599                        // Phase 2: fetch the resource-based policy (if
600                        // any) attached to the target resource and
601                        // pass it to the evaluator alongside the
602                        // principal's identity policies. The resource's
603                        // owning account is parsed from the ARN (#381
604                        // multi-account alignment); S3 ARNs have an
605                        // empty account field, so we fall back to the
606                        // server's configured account ID in that case.
607                        let resource_policy_json =
608                            config.resource_policy_provider.as_ref().and_then(|p| {
609                                p.resource_policy(&detected.service, &iam_action.resource)
610                            });
611                        // Derive the resource-owning account. Prefer a provider
612                        // lookup (S3 ARNs carry no account, so the bucket's
613                        // owner is resolved from state — without this, account
614                        // A reaching account B's bucket would be mis-read as
615                        // same-account and skip B's bucket-policy requirement,
616                        // bug-audit 2026-05-28, 5.3), then fall back to the
617                        // account embedded in the ARN (SQS/SNS/Lambda/…), then
618                        // to the caller's account for wildcard / unscoped
619                        // actions (ListQueues, GetCallerIdentity).
620                        let resource_account_id = config
621                            .resource_policy_provider
622                            .as_ref()
623                            .and_then(|p| {
624                                p.resource_owner_account(&detected.service, &iam_action.resource)
625                            })
626                            .or_else(|| parse_account_from_arn(&iam_action.resource))
627                            .unwrap_or_else(|| principal.account_id.clone());
628                        // SCP ceiling: resolve the inherited SCP chain
629                        // for this principal (management accounts and
630                        // service-linked roles come back as `None`, in
631                        // which case the evaluator treats the layer as
632                        // absent). Audit breadcrumbs emitted by the
633                        // resolver itself, not here.
634                        let scps = config
635                            .scp_resolver
636                            .as_ref()
637                            .and_then(|r| r.scps_for(principal));
638                        let decision = evaluator.evaluate_with_resource_policy(
639                            principal,
640                            &iam_action,
641                            &condition_context,
642                            resource_policy_json.as_deref(),
643                            &resource_account_id,
644                            &caller_session_policies,
645                            scps.as_deref(),
646                        );
647                        if !decision.is_allow() {
648                            tracing::warn!(
649                                target: "fakecloud::iam::audit",
650                                service = %detected.service,
651                                action = %iam_action.action_string(),
652                                resource = %iam_action.resource,
653                                principal = %principal.arn,
654                                resource_policy_present = resource_policy_json.is_some(),
655                                decision = ?decision,
656                                mode = %config.iam_mode,
657                                request_id = %request_id,
658                                "IAM policy evaluation denied request"
659                            );
660                            if config.iam_mode.is_strict() {
661                                // Real AWS includes an "Encoded
662                                // authorization failure message" suffix
663                                // on AccessDeniedException — an opaque
664                                // base64+zlib JSON blob that the caller
665                                // can pass to STS
666                                // `DecodeAuthorizationMessage` to
667                                // recover the structured deny reason
668                                // (action, principal, matched
669                                // statements, condition context). We
670                                // produce the same blob inline so
671                                // existing tooling that decodes deny
672                                // reasons works against fakecloud.
673                                let context_summary = serde_json::json!({
674                                    "aws:PrincipalArn": principal.arn,
675                                    "aws:PrincipalAccount": principal.account_id,
676                                    "aws:RequestedRegion": condition_context
677                                        .aws_requested_region
678                                        .clone()
679                                        .unwrap_or_default(),
680                                    "aws:SecureTransport": condition_context
681                                        .aws_secure_transport
682                                        .unwrap_or(false),
683                                    "aws:Action": iam_action.action_string(),
684                                    "aws:Resource": iam_action.resource,
685                                    "decision": format!("{:?}", decision),
686                                });
687                                let action_string = iam_action.action_string();
688                                let encoded = crate::auth_message::encode_deny(
689                                    matches!(decision, crate::auth::IamDecision::ExplicitDeny),
690                                    Some(&action_string),
691                                    Some(&principal.arn),
692                                    Vec::new(),
693                                    Some(context_summary),
694                                );
695                                return build_error_response(
696                                    StatusCode::FORBIDDEN,
697                                    "AccessDeniedException",
698                                    &format!(
699                                        "User: {} is not authorized to perform: {} on resource: {} Encoded authorization failure message: {}",
700                                        principal.arn,
701                                        iam_action.action_string(),
702                                        iam_action.resource,
703                                        encoded,
704                                    ),
705                                    &request_id,
706                                    detected.protocol,
707                                );
708                            }
709                            // Soft mode: audit log already emitted; fall
710                            // through to the handler.
711                        }
712                    } else {
713                        // Service opted in but didn't return an IamAction
714                        // for this specific operation — programming bug,
715                        // surface it loudly in soft/strict mode so it's
716                        // visible during rollout.
717                        tracing::warn!(
718                            target: "fakecloud::iam::audit",
719                            service = %detected.service,
720                            action = %aws_request.action,
721                            "service is iam_enforceable but has no IamAction mapping for this action; skipping evaluation"
722                        );
723                    }
724                }
725            } else if aws_request.access_key_id.is_none() {
726                // Truly anonymous (unsigned) caller — no Authorization header at
727                // all. No identity policies exist, so authorization rests
728                // entirely on the resource policy (a bucket policy granting
729                // `Principal:"*"`) and public-read ACLs — mirroring AWS, which
730                // denies anonymous requests unless the resource is explicitly
731                // made public. Without this an anonymous request that reached an
732                // iam_enforceable service in enforcement mode would bypass
733                // authorization entirely.
734                //
735                // A request that carried an Authorization header but whose
736                // credential did not resolve (principal `None` with
737                // `access_key_id` `Some`) is intentionally left alone here: with
738                // SigV4 verification off, fakecloud does not reject unverified
739                // signed requests, and turning them into anonymous denials would
740                // change long-standing behavior.
741                if let Some(iam_action) = service.iam_action_for(&aws_request) {
742                    let now = chrono::Utc::now();
743                    let mut condition_context = ConditionContext {
744                        aws_source_ip: remote_addr.map(|sa| sa.ip()),
745                        aws_current_time: Some(now),
746                        aws_epoch_time: Some(now.timestamp()),
747                        aws_secure_transport: Some(is_secure_transport(&aws_request.headers)),
748                        aws_requested_region: Some(aws_request.region.clone()),
749                        ..Default::default()
750                    };
751                    condition_context.service_keys =
752                        service.iam_condition_keys_for(&aws_request, &iam_action);
753                    let resource_policy_json = config
754                        .resource_policy_provider
755                        .as_ref()
756                        .and_then(|p| p.resource_policy(&detected.service, &iam_action.resource));
757                    let policy_decision = evaluator.evaluate_anonymous(
758                        &iam_action,
759                        &condition_context,
760                        resource_policy_json.as_deref(),
761                    );
762                    let policy_allows = policy_decision.is_allow();
763                    // An explicit Deny in the resource policy always wins, even
764                    // over a public-read ACL — matching AWS's Deny-overrides
765                    // precedence. Collapsing the decision to a bool and ORing the
766                    // ACL let a public ACL override an explicit anonymous Deny.
767                    let policy_explicit_deny =
768                        matches!(policy_decision, crate::auth::IamDecision::ExplicitDeny);
769                    let acl_allows = !policy_explicit_deny
770                        && config.resource_policy_provider.as_ref().is_some_and(|p| {
771                            p.public_acl_allows(
772                                &detected.service,
773                                &iam_action.resource,
774                                iam_action.action,
775                            )
776                        });
777                    if !policy_allows && !acl_allows {
778                        tracing::warn!(
779                            target: "fakecloud::iam::audit",
780                            service = %detected.service,
781                            action = %iam_action.action_string(),
782                            resource = %iam_action.resource,
783                            resource_policy_present = resource_policy_json.is_some(),
784                            mode = %config.iam_mode,
785                            request_id = %request_id,
786                            "anonymous request denied: no public bucket policy or ACL grants the action"
787                        );
788                        if config.iam_mode.is_strict() {
789                            return build_error_response(
790                                StatusCode::FORBIDDEN,
791                                "AccessDenied",
792                                "Access Denied",
793                                &request_id,
794                                detected.protocol,
795                            );
796                        }
797                        // Soft mode: audit log emitted; fall through to the handler.
798                    }
799                }
800            }
801        }
802    }
803
804    match service.handle(aws_request).await {
805        Ok(resp) => {
806            let mut builder = Response::builder()
807                .status(resp.status)
808                .header("x-amzn-requestid", &request_id)
809                .header("x-amz-request-id", &request_id);
810
811            if !resp.content_type.is_empty() {
812                builder = builder.header("content-type", &resp.content_type);
813            }
814
815            let has_content_length = resp
816                .headers
817                .iter()
818                .any(|(k, _)| k.as_str().eq_ignore_ascii_case("content-length"));
819
820            for (k, v) in &resp.headers {
821                builder = builder.header(k, v);
822            }
823
824            match resp.body {
825                ResponseBody::Bytes(b) => builder.body(Body::from(b)).unwrap(),
826                ResponseBody::File { file, size } => {
827                    let stream = tokio_util::io::ReaderStream::new(file);
828                    let body = Body::from_stream(stream);
829                    if !has_content_length {
830                        builder = builder.header("content-length", size.to_string());
831                    }
832                    builder.body(body).unwrap()
833                }
834            }
835        }
836        Err(err) => {
837            tracing::warn!(
838                service = %detected.service,
839                action = %detected.action,
840                error = %err,
841                "request failed"
842            );
843            let error_headers = err.response_headers().to_vec();
844            let mut resp = build_error_response_with_fields(
845                err.status(),
846                err.code(),
847                &err.message(),
848                &request_id,
849                detected.protocol,
850                err.extra_fields(),
851            );
852            for (k, v) in &error_headers {
853                if let (Ok(name), Ok(val)) = (
854                    k.parse::<http::header::HeaderName>(),
855                    v.parse::<http::header::HeaderValue>(),
856                ) {
857                    resp.headers_mut().insert(name, val);
858                }
859            }
860            resp
861        }
862    }
863}
864
865/// Configuration passed to the dispatch handler.
866#[derive(Clone)]
867pub struct DispatchConfig {
868    pub region: String,
869    pub account_id: String,
870    /// Whether to cryptographically verify SigV4 signatures on incoming
871    /// requests. Wired through from `--verify-sigv4` /
872    /// `FAKECLOUD_VERIFY_SIGV4`. Off by default.
873    pub verify_sigv4: bool,
874    /// IAM policy evaluation mode. Wired through from `--iam` /
875    /// `FAKECLOUD_IAM`. Defaults to [`IamMode::Off`]. Actual evaluation is
876    /// added in a later batch; today this field is plumbed but never
877    /// consulted.
878    pub iam_mode: IamMode,
879    /// Resolves access key IDs to their secrets and owning principals.
880    /// Required when `verify_sigv4` or `iam_mode != Off`. When `None`, both
881    /// features gracefully degrade to off-by-default behavior.
882    pub credential_resolver: Option<Arc<dyn CredentialResolver>>,
883    /// Evaluates IAM identity policies for a resolved principal + action.
884    /// Required when `iam_mode != Off`. When `None`, enforcement silently
885    /// degrades to off even if `iam_mode` is set.
886    pub policy_evaluator: Option<Arc<dyn IamPolicyEvaluator>>,
887    /// Resolves resource-based policies (S3 bucket policies in the
888    /// initial rollout) to hand to the evaluator alongside the
889    /// principal's identity policies. `None` means the server was
890    /// started without any resource-policy-owning service registered;
891    /// dispatch then behaves as if no resource policy is attached to
892    /// any resource, identical to the Phase 1 behavior.
893    pub resource_policy_provider: Option<Arc<dyn ResourcePolicyProvider>>,
894    /// Resolves the ordered SCP chain that applies to a principal's
895    /// account (root-OU first, account-direct last). `None` means no
896    /// organizations resolver has been registered — SCPs never gate
897    /// any request in that case. Off-by-default matches the Batch 4
898    /// contract: zero behavior change until a user calls
899    /// `CreateOrganization` and the resolver is wired.
900    pub scp_resolver: Option<Arc<dyn crate::auth::ScpResolver>>,
901}
902
903impl std::fmt::Debug for DispatchConfig {
904    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
905        f.debug_struct("DispatchConfig")
906            .field("region", &self.region)
907            .field("account_id", &self.account_id)
908            .field("verify_sigv4", &self.verify_sigv4)
909            .field("iam_mode", &self.iam_mode)
910            .field(
911                "credential_resolver",
912                &self
913                    .credential_resolver
914                    .as_ref()
915                    .map(|_| "<CredentialResolver>"),
916            )
917            .field(
918                "policy_evaluator",
919                &self
920                    .policy_evaluator
921                    .as_ref()
922                    .map(|_| "<IamPolicyEvaluator>"),
923            )
924            .field(
925                "resource_policy_provider",
926                &self
927                    .resource_policy_provider
928                    .as_ref()
929                    .map(|_| "<ResourcePolicyProvider>"),
930            )
931            .field(
932                "scp_resolver",
933                &self.scp_resolver.as_ref().map(|_| "<ScpResolver>"),
934            )
935            .finish()
936    }
937}
938
939impl DispatchConfig {
940    /// Minimal constructor for tests and call sites that don't care about the
941    /// opt-in security features.
942    pub fn new(region: impl Into<String>, account_id: impl Into<String>) -> Self {
943        Self {
944            region: region.into(),
945            account_id: account_id.into(),
946            verify_sigv4: false,
947            iam_mode: IamMode::Off,
948            credential_resolver: None,
949            policy_evaluator: None,
950            resource_policy_provider: None,
951            scp_resolver: None,
952        }
953    }
954}
955
956/// Extract the 12-digit account ID segment from an AWS ARN.
957///
958/// ARNs follow `arn:<partition>:<service>:<region>:<account>:<resource>`.
959/// Identifies routes that opt into streaming request bodies. Returns
960/// `Some((service, action_hint))` when the dispatch path should hand
961/// the raw body to the service handler unbuffered, otherwise `None`
962/// for the default buffered path. The handler reads the stream via
963/// [`crate::service::AwsRequest::take_body_stream`].
964///
965/// Streaming-eligible routes today:
966///
967/// * `s3` PUT object — `PUT /<bucket>/<key>` with a SigV4 (or
968///   presigned) auth header. Covers PutObject, UploadPart, and
969///   UploadPartCopy. The S3 service spills to disk via
970///   [`fakecloud_persistence::BodySource::File`] when the stream is
971///   present.
972/// * `ecr` OCI Distribution v2 blob upload — `PATCH` and `PUT` on
973///   `/v2/{name}/blobs/uploads/{uuid}`. The ECR service spools the
974///   stream into a per-upload temp file before computing the digest.
975fn streaming_route(
976    method: &http::Method,
977    path: &str,
978    headers: &http::HeaderMap,
979    query_params: &HashMap<String, String>,
980) -> Option<(&'static str, &'static str)> {
981    // ECR OCI v2 blob upload (PATCH chunk + final PUT).
982    if (method == http::Method::PATCH || method == http::Method::PUT)
983        && path.starts_with("/v2/")
984        && path.contains("/blobs/uploads/")
985    {
986        return Some(("ecr", ""));
987    }
988
989    // S3 PutObject / UploadPart / UploadPartCopy. Detect either via
990    // SigV4 service field in the Authorization header OR via a SigV4
991    // presigned URL (X-Amz-Credential .../s3/...) OR a SigV2 presigned
992    // URL (AWSAccessKeyId + Signature + Expires query parameters).
993    if method == http::Method::PUT {
994        let after = path.trim_start_matches('/');
995        // Path-style PutObject is `PUT /<bucket>/<key>` (path contains a
996        // slash); virtual-hosted-style is `PUT /<key>` with the bucket
997        // in the Host header. For virtual-hosted, accept any non-empty
998        // path so the key flows through the streaming dispatch — the
999        // Host parser already routed this request to S3.
1000        let virtual_hosted_s3 = protocol::parse_routing_host_from_headers(headers)
1001            .filter(|h| h.service == "s3" && h.bucket.is_some())
1002            .is_some();
1003        if after.is_empty() || (!virtual_hosted_s3 && !after.contains('/')) {
1004            return None;
1005        }
1006        let header_s3 = headers
1007            .get("authorization")
1008            .and_then(|v| v.to_str().ok())
1009            .and_then(fakecloud_aws::sigv4::parse_sigv4)
1010            .map(|info| info.service == "s3")
1011            .unwrap_or(false);
1012        let presigned_v4_s3 = query_params
1013            .get("X-Amz-Credential")
1014            .and_then(|c| c.split('/').nth(3).map(|s| s.to_string()))
1015            .map(|service| service == "s3")
1016            .unwrap_or(false);
1017        let presigned_v2 = query_params.contains_key("AWSAccessKeyId")
1018            && query_params.contains_key("Signature")
1019            && query_params.contains_key("Expires");
1020        if header_s3 || presigned_v4_s3 || presigned_v2 {
1021            return Some(("s3", ""));
1022        }
1023    }
1024
1025    None
1026}
1027
1028/// Default request-body buffering cap. fakecloud reads the entire
1029/// request body into memory before handing it to a service handler,
1030/// so this ceiling caps RAM usage per in-flight request.
1031///
1032/// Default 1 GiB — comfortably above legitimate single S3 PutObject
1033/// payloads (AWS recommends multipart above ~100 MiB) and each
1034/// multipart part dispatches through here separately. Override with
1035/// `FAKECLOUD_MAX_REQUEST_BODY_BYTES` (decimal bytes) when running
1036/// stress tests that push past the default.
1037const DEFAULT_MAX_REQUEST_BODY_BYTES: usize = 1024 * 1024 * 1024;
1038
1039/// The server-wide buffered request body cap in bytes, from
1040/// `FAKECLOUD_MAX_REQUEST_BODY_BYTES` (default 1 GiB). Public so buffered
1041/// sub-proxies (e.g. the CloudFront viewer data plane) apply the SAME cap as
1042/// direct traffic rather than an inconsistent limit of their own.
1043pub fn max_request_body_bytes() -> usize {
1044    static CACHED: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
1045    *CACHED.get_or_init(|| {
1046        std::env::var("FAKECLOUD_MAX_REQUEST_BODY_BYTES")
1047            .ok()
1048            .and_then(|s| s.parse::<usize>().ok())
1049            .filter(|&n| n > 0)
1050            .unwrap_or(DEFAULT_MAX_REQUEST_BODY_BYTES)
1051    })
1052}
1053
1054/// For the cross-account decision in IAM enforcement, the "resource
1055/// account" is the ARN's account segment. Some services (notably S3)
1056/// produce ARNs with an empty account field — for those we return
1057/// `None` and let the caller fall back to the server's configured
1058/// account ID. Malformed or non-ARN strings also return `None`.
1059fn parse_account_from_arn(arn: &str) -> Option<String> {
1060    let mut parts = arn.splitn(6, ':');
1061    if parts.next()? != "arn" {
1062        return None;
1063    }
1064    let _partition = parts.next()?;
1065    let _service = parts.next()?;
1066    let _region = parts.next()?;
1067    let account = parts.next()?;
1068    // Resource segment must exist (parts.next().is_some()) for the ARN
1069    // to be well-formed, but we don't consume its value here.
1070    parts.next()?;
1071    if account.is_empty() {
1072        None
1073    } else {
1074        Some(account.to_string())
1075    }
1076}
1077
1078/// Whether the request's `user-agent` (or `x-amz-user-agent`) carries the
1079/// `api/neptune` SDK-metadata token the `aws-sdk-neptune` client stamps in.
1080/// Used to disambiguate Neptune from RDS, which share the `rds` SigV4 scope
1081/// and endpoint. Matches the `api/neptune` token that AWS SDKs emit (a
1082/// leading `api/neptune` optionally followed by `#`/`/` and a version).
1083fn user_agent_indicates_neptune(headers: &http::HeaderMap) -> bool {
1084    for name in ["user-agent", "x-amz-user-agent"] {
1085        if let Some(ua) = headers.get(name).and_then(|v| v.to_str().ok()) {
1086            for part in ua.split_whitespace() {
1087                if let Some(rest) = part.strip_prefix("api/neptune") {
1088                    if rest.is_empty() || rest.starts_with('#') || rest.starts_with('/') {
1089                        return true;
1090                    }
1091                }
1092            }
1093        }
1094    }
1095    false
1096}
1097
1098/// Extract region from User-Agent header suffix `region/<region>`.
1099/// Whether the request's `user-agent` (or `x-amz-user-agent`) carries the
1100/// `api/docdb` SDK-metadata token the `aws-sdk-docdb` client stamps in. Used
1101/// to disambiguate DocumentDB from RDS, which share the `rds` SigV4 scope
1102/// and endpoint. Matches the `api/docdb` token that AWS SDKs emit (a leading
1103/// `api/docdb` optionally followed by `#`/`/` and a version).
1104fn user_agent_indicates_docdb(headers: &http::HeaderMap) -> bool {
1105    for name in ["user-agent", "x-amz-user-agent"] {
1106        if let Some(ua) = headers.get(name).and_then(|v| v.to_str().ok()) {
1107            for part in ua.split_whitespace() {
1108                if let Some(rest) = part.strip_prefix("api/docdb") {
1109                    if rest.is_empty() || rest.starts_with('#') || rest.starts_with('/') {
1110                        return true;
1111                    }
1112                }
1113            }
1114        }
1115    }
1116    false
1117}
1118
1119fn extract_region_from_user_agent(headers: &http::HeaderMap) -> Option<String> {
1120    let ua = headers.get("user-agent")?.to_str().ok()?;
1121    for part in ua.split_whitespace() {
1122        if let Some(region) = part.strip_prefix("region/") {
1123            if !region.is_empty() {
1124                return Some(region.to_string());
1125            }
1126        }
1127    }
1128    None
1129}
1130
1131fn build_error_response(
1132    status: StatusCode,
1133    code: &str,
1134    message: &str,
1135    request_id: &str,
1136    protocol: AwsProtocol,
1137) -> Response<Body> {
1138    build_error_response_with_fields(status, code, message, request_id, protocol, &[])
1139}
1140
1141fn build_error_response_with_fields(
1142    status: StatusCode,
1143    code: &str,
1144    message: &str,
1145    request_id: &str,
1146    protocol: AwsProtocol,
1147    extra_fields: &[(String, String)],
1148) -> Response<Body> {
1149    let (status, content_type, body) = match protocol {
1150        AwsProtocol::Query => {
1151            fakecloud_aws::error::xml_error_response(status, code, message, request_id)
1152        }
1153        AwsProtocol::Rest => fakecloud_aws::error::s3_xml_error_response_with_fields(
1154            status,
1155            code,
1156            message,
1157            request_id,
1158            extra_fields,
1159        ),
1160        AwsProtocol::Json | AwsProtocol::RestJson => {
1161            fakecloud_aws::error::json_error_response_with_fields(
1162                status,
1163                code,
1164                message,
1165                extra_fields,
1166            )
1167        }
1168    };
1169
1170    // S3 (and other REST-XML services) place the error code in
1171    // `x-amz-error-code` so HEAD responses — which HTTP forbids from
1172    // carrying a body — still surface the code. AWS SDKs read this header
1173    // when the body is empty. Emit it on every error response so HEAD,
1174    // OPTIONS, and any client that strips the body still see the code.
1175    // Backend errors regularly include newlines (multi-line stderr from
1176    // docker/podman/etc.); HTTP header values reject control characters,
1177    // so sanitize before insertion or the builder rejects the response
1178    // and the connection drops.
1179    let safe_code = sanitize_header_value(code);
1180    let safe_message = sanitize_header_value(message);
1181    let mut builder = Response::builder()
1182        .status(status)
1183        .header("content-type", content_type)
1184        .header("x-amzn-requestid", request_id)
1185        .header("x-amz-request-id", request_id);
1186    if let Ok(v) = http::HeaderValue::from_str(&safe_code) {
1187        builder = builder.header("x-amz-error-code", v);
1188    }
1189    if let Ok(v) = http::HeaderValue::from_str(&safe_message) {
1190        builder = builder.header("x-amz-error-message", v);
1191    }
1192    builder.body(Body::from(body)).unwrap_or_else(|_| {
1193        // Builder only fails if a header is invalid; we sanitized the two
1194        // we control, so the remaining ones (content-type, request id) are
1195        // ASCII and safe. This fallback exists purely so we never panic.
1196        Response::new(Body::empty())
1197    })
1198}
1199
1200/// Strip characters that HTTP header values reject (control bytes, CR/LF/TAB)
1201/// and truncate to a length that AWS SDKs handle cleanly. Backend tools
1202/// (docker, podman, kubectl, …) emit multi-line stderr, and forwarding that
1203/// raw into `x-amz-error-message` previously panicked the dispatcher.
1204fn sanitize_header_value(s: &str) -> String {
1205    const MAX_LEN: usize = 1024;
1206    let mut out = String::with_capacity(s.len().min(MAX_LEN));
1207    for ch in s.chars() {
1208        if out.len() >= MAX_LEN {
1209            break;
1210        }
1211        // Header values forbid CR, LF, and other control bytes (RFC 9110).
1212        // Replace with a single space so multi-line messages stay readable.
1213        if ch.is_control() {
1214            if !out.ends_with(' ') {
1215                out.push(' ');
1216            }
1217        } else {
1218            out.push(ch);
1219        }
1220    }
1221    out.trim().to_string()
1222}
1223
1224/// Build the [`ConditionContext`] passed to the IAM evaluator for one
1225/// request. Populates the 10 global condition keys from the resolved
1226/// principal + the HTTP request. Service-specific keys are deferred to
1227/// a follow-up batch and left empty.
1228/// For an unsigned request that no other detection rule claimed, return the
1229/// bucket name when the first path segment names an existing S3 bucket.
1230///
1231/// fakecloud serves every service from one endpoint, so an anonymous
1232/// path-style S3 request (`GET /bucket/key`, no SigV4) is indistinguishable
1233/// from an API Gateway execute-api call by headers alone. Bucket existence is
1234/// the disambiguator: if the segment is a real bucket, route to S3; otherwise
1235/// fall through to the apigateway catch-all. Uses the already-wired
1236/// `resource_policy_provider`, which resolves S3 bucket ownership from state
1237/// (`Some` => the bucket exists). Returns `None` when no provider is wired.
1238/// Recover the access key from a SigV2 presigned URL. AWS SigV2 presigning
1239/// puts the key in the `AWSAccessKeyId` query parameter alongside `Signature`
1240/// and `Expires`; all three must be present for the URL to be a SigV2 presign.
1241/// Returns None for SigV4 presigns (which use `X-Amz-Credential`) or unsigned
1242/// requests.
1243fn sigv2_presigned_access_key(query_params: &HashMap<String, String>) -> Option<String> {
1244    if query_params.contains_key("Signature") && query_params.contains_key("Expires") {
1245        query_params.get("AWSAccessKeyId").cloned()
1246    } else {
1247        None
1248    }
1249}
1250
1251fn anonymous_s3_bucket(uri: &http::Uri, config: &DispatchConfig) -> Option<String> {
1252    let provider = config.resource_policy_provider.as_ref()?;
1253    let segment = uri.path().split('/').find(|s| !s.is_empty())?.to_string();
1254    let arn = format!("arn:aws:s3:::{segment}");
1255    provider.resource_owner_account("s3", &arn).map(|_| segment)
1256}
1257
1258fn build_condition_context(
1259    principal: &Principal,
1260    remote_addr: Option<SocketAddr>,
1261    region: &str,
1262    secure_transport: bool,
1263) -> ConditionContext {
1264    let now = chrono::Utc::now();
1265    ConditionContext {
1266        aws_username: aws_username_from_principal(principal),
1267        aws_userid: Some(principal.user_id.clone()),
1268        aws_principal_arn: Some(principal.arn.clone()),
1269        aws_principal_account: Some(principal.account_id.clone()),
1270        aws_principal_type: Some(principal_type_label(principal.principal_type).to_string()),
1271        aws_source_ip: remote_addr.map(|sa| sa.ip()),
1272        aws_current_time: Some(now),
1273        aws_epoch_time: Some(now.timestamp()),
1274        aws_secure_transport: Some(secure_transport),
1275        aws_requested_region: Some(region.to_string()),
1276        // F3 keys: populated from the caller's session context when STS
1277        // mints credentials with MFA / SAML / OIDC / VPC-endpoint hints.
1278        // Default-None here so tests/dispatch sites that don't set them
1279        // safe-fail any policy referencing them — matching AWS for keys
1280        // that aren't asserted.
1281        aws_mfa_present: None,
1282        aws_mfa_age_seconds: None,
1283        aws_called_via: Vec::new(),
1284        aws_source_vpce: None,
1285        aws_source_vpc: None,
1286        aws_vpc_source_ip: None,
1287        aws_federated_provider: None,
1288        aws_token_issue_time: None,
1289        service_keys: Default::default(),
1290        resource_tags: None,
1291        request_tags: None,
1292        principal_tags: None,
1293    }
1294}
1295
1296/// `aws:username` is only set for IAM users, matching AWS. For assumed
1297/// roles, federated users, root, and unknown principals the key is
1298/// absent — operators that reference it without `IfExists` safe-fail.
1299fn aws_username_from_principal(principal: &Principal) -> Option<String> {
1300    if principal.principal_type != PrincipalType::User {
1301        return None;
1302    }
1303    let after = principal.arn.rsplit_once(":user/").map(|(_, s)| s)?;
1304    // Strip any IAM path prefix; bare username is the last segment.
1305    Some(after.rsplit('/').next().unwrap_or(after).to_string())
1306}
1307
1308/// AWS's `aws:PrincipalType` uses PascalCase identifiers, distinct from
1309/// the lowercase ones [`PrincipalType::as_str`] returns for ARNs.
1310fn principal_type_label(t: PrincipalType) -> &'static str {
1311    match t {
1312        PrincipalType::User => "User",
1313        PrincipalType::AssumedRole => "AssumedRole",
1314        PrincipalType::FederatedUser => "FederatedUser",
1315        PrincipalType::Root => "Account",
1316        PrincipalType::Unknown => "Unknown",
1317    }
1318}
1319
1320/// Best-effort detection of TLS-terminated requests. Direct HTTPS
1321/// connections are not yet supported by the fakecloud server (it speaks
1322/// plain HTTP), so the only signal is an `x-forwarded-proto: https`
1323/// header set by an upstream proxy. Anything else evaluates to `false`,
1324/// which matches the typical local-dev setup.
1325fn is_secure_transport(headers: &http::HeaderMap) -> bool {
1326    headers
1327        .get("x-forwarded-proto")
1328        .and_then(|v| v.to_str().ok())
1329        .map(|s| s.eq_ignore_ascii_case("https"))
1330        .unwrap_or(false)
1331}
1332
1333trait ProtocolExt {
1334    fn error_status(&self) -> StatusCode;
1335}
1336
1337impl ProtocolExt for AwsProtocol {
1338    fn error_status(&self) -> StatusCode {
1339        StatusCode::BAD_REQUEST
1340    }
1341}
1342
1343#[cfg(test)]
1344mod tests {
1345    use super::*;
1346
1347    #[test]
1348    fn default_max_request_body_bytes_is_one_gib() {
1349        // Without the env override, the cap defaults to 1 GiB. The
1350        // public function caches via OnceLock so only the first call
1351        // in the process matters; we assert the constant directly.
1352        assert_eq!(DEFAULT_MAX_REQUEST_BODY_BYTES, 1024 * 1024 * 1024);
1353    }
1354
1355    #[test]
1356    fn sigv2_presigned_access_key_extracted_with_signature_and_expires() {
1357        let mut q = HashMap::new();
1358        q.insert("AWSAccessKeyId".to_string(), "AKIAEXAMPLE".to_string());
1359        q.insert("Signature".to_string(), "abc%2Bdef".to_string());
1360        q.insert("Expires".to_string(), "1700000000".to_string());
1361        assert_eq!(
1362            sigv2_presigned_access_key(&q).as_deref(),
1363            Some("AKIAEXAMPLE")
1364        );
1365    }
1366
1367    #[test]
1368    fn sigv2_presigned_access_key_none_without_signature_or_expires() {
1369        // AWSAccessKeyId alone (e.g. a stray query param) is not a SigV2
1370        // presign and must not be treated as a credential.
1371        let mut q = HashMap::new();
1372        q.insert("AWSAccessKeyId".to_string(), "AKIAEXAMPLE".to_string());
1373        assert_eq!(sigv2_presigned_access_key(&q), None);
1374
1375        q.insert("Expires".to_string(), "1700000000".to_string());
1376        assert_eq!(
1377            sigv2_presigned_access_key(&q),
1378            None,
1379            "missing Signature must not qualify"
1380        );
1381    }
1382
1383    #[test]
1384    fn sigv2_presigned_access_key_none_for_unsigned_request() {
1385        assert_eq!(sigv2_presigned_access_key(&HashMap::new()), None);
1386    }
1387
1388    #[test]
1389    fn dispatch_config_new_defaults_to_off() {
1390        let cfg = DispatchConfig::new("us-east-1", "123456789012");
1391        assert_eq!(cfg.region, "us-east-1");
1392        assert_eq!(cfg.account_id, "123456789012");
1393        assert!(!cfg.verify_sigv4);
1394        assert_eq!(cfg.iam_mode, IamMode::Off);
1395    }
1396
1397    #[test]
1398    fn aws_username_strips_iam_path_for_users() {
1399        let p = Principal {
1400            arn: "arn:aws:iam::123456789012:user/engineering/alice".into(),
1401            user_id: "AIDAALICE".into(),
1402            account_id: "123456789012".into(),
1403            principal_type: PrincipalType::User,
1404            source_identity: None,
1405            tags: None,
1406        };
1407        assert_eq!(aws_username_from_principal(&p), Some("alice".into()));
1408    }
1409
1410    #[test]
1411    fn aws_username_unset_for_assumed_role() {
1412        let p = Principal {
1413            arn: "arn:aws:sts::123456789012:assumed-role/ops/session".into(),
1414            user_id: "AROAOPS:session".into(),
1415            account_id: "123456789012".into(),
1416            principal_type: PrincipalType::AssumedRole,
1417            source_identity: None,
1418            tags: None,
1419        };
1420        assert_eq!(aws_username_from_principal(&p), None);
1421    }
1422
1423    #[test]
1424    fn principal_type_label_matches_aws_casing() {
1425        assert_eq!(principal_type_label(PrincipalType::User), "User");
1426        assert_eq!(
1427            principal_type_label(PrincipalType::AssumedRole),
1428            "AssumedRole"
1429        );
1430        assert_eq!(principal_type_label(PrincipalType::Root), "Account");
1431    }
1432
1433    #[test]
1434    fn build_condition_context_populates_global_keys() {
1435        let p = Principal {
1436            arn: "arn:aws:iam::123456789012:user/alice".into(),
1437            user_id: "AIDAALICE".into(),
1438            account_id: "123456789012".into(),
1439            principal_type: PrincipalType::User,
1440            source_identity: None,
1441            tags: None,
1442        };
1443        let addr: SocketAddr = "10.0.0.1:54321".parse().unwrap();
1444        let ctx = build_condition_context(&p, Some(addr), "us-east-1", false);
1445        assert_eq!(ctx.aws_username.as_deref(), Some("alice"));
1446        assert_eq!(ctx.aws_userid.as_deref(), Some("AIDAALICE"));
1447        assert_eq!(
1448            ctx.aws_principal_arn.as_deref(),
1449            Some("arn:aws:iam::123456789012:user/alice")
1450        );
1451        assert_eq!(ctx.aws_principal_account.as_deref(), Some("123456789012"));
1452        assert_eq!(ctx.aws_principal_type.as_deref(), Some("User"));
1453        assert_eq!(
1454            ctx.aws_source_ip.map(|i| i.to_string()).as_deref(),
1455            Some("10.0.0.1")
1456        );
1457        assert_eq!(ctx.aws_requested_region.as_deref(), Some("us-east-1"));
1458        assert_eq!(ctx.aws_secure_transport, Some(false));
1459        assert!(ctx.aws_current_time.is_some());
1460        assert!(ctx.aws_epoch_time.is_some());
1461    }
1462
1463    #[test]
1464    fn is_secure_transport_reads_x_forwarded_proto() {
1465        let mut headers = http::HeaderMap::new();
1466        headers.insert("x-forwarded-proto", "https".parse().unwrap());
1467        assert!(is_secure_transport(&headers));
1468        headers.insert("x-forwarded-proto", "http".parse().unwrap());
1469        assert!(!is_secure_transport(&headers));
1470        let empty = http::HeaderMap::new();
1471        assert!(!is_secure_transport(&empty));
1472    }
1473
1474    #[test]
1475    fn parse_account_from_arn_extracts_standard_shapes() {
1476        assert_eq!(
1477            parse_account_from_arn("arn:aws:sqs:us-east-1:123456789012:queue"),
1478            Some("123456789012".to_string())
1479        );
1480        assert_eq!(
1481            parse_account_from_arn("arn:aws:iam::123456789012:user/alice"),
1482            Some("123456789012".to_string())
1483        );
1484    }
1485
1486    #[test]
1487    fn parse_account_from_arn_returns_none_for_s3_empty_account() {
1488        // S3 ARNs have both region and account empty.
1489        assert_eq!(parse_account_from_arn("arn:aws:s3:::my-bucket"), None);
1490        assert_eq!(
1491            parse_account_from_arn("arn:aws:s3:::my-bucket/path/to/key"),
1492            None
1493        );
1494    }
1495
1496    #[test]
1497    fn parse_account_from_arn_returns_none_for_malformed() {
1498        assert_eq!(parse_account_from_arn(""), None);
1499        assert_eq!(parse_account_from_arn("not-an-arn"), None);
1500        assert_eq!(parse_account_from_arn("arn:aws:sqs:us-east-1"), None);
1501        assert_eq!(parse_account_from_arn("arn:aws:sqs"), None);
1502    }
1503
1504    #[test]
1505    fn extract_region_from_user_agent_finds_region_segment() {
1506        let mut headers = http::HeaderMap::new();
1507        headers.insert(
1508            "user-agent",
1509            "aws-sdk-rust/1.0 os/linux region/eu-central-1"
1510                .parse()
1511                .unwrap(),
1512        );
1513        assert_eq!(
1514            extract_region_from_user_agent(&headers),
1515            Some("eu-central-1".to_string())
1516        );
1517    }
1518
1519    #[test]
1520    fn extract_region_from_user_agent_none_without_header() {
1521        let headers = http::HeaderMap::new();
1522        assert_eq!(extract_region_from_user_agent(&headers), None);
1523    }
1524
1525    #[test]
1526    fn extract_region_from_user_agent_ignores_empty_region() {
1527        let mut headers = http::HeaderMap::new();
1528        headers.insert("user-agent", "aws-sdk-java region/".parse().unwrap());
1529        assert_eq!(extract_region_from_user_agent(&headers), None);
1530    }
1531
1532    #[test]
1533    fn extract_region_from_user_agent_none_when_no_region_marker() {
1534        let mut headers = http::HeaderMap::new();
1535        headers.insert("user-agent", "curl/7.79.1".parse().unwrap());
1536        assert_eq!(extract_region_from_user_agent(&headers), None);
1537    }
1538
1539    #[test]
1540    fn aws_username_none_for_root() {
1541        let p = Principal {
1542            arn: "arn:aws:iam::123456789012:root".into(),
1543            user_id: "123456789012".into(),
1544            account_id: "123456789012".into(),
1545            principal_type: PrincipalType::Root,
1546            source_identity: None,
1547            tags: None,
1548        };
1549        assert_eq!(aws_username_from_principal(&p), None);
1550    }
1551
1552    #[test]
1553    fn aws_username_bare_no_path() {
1554        let p = Principal {
1555            arn: "arn:aws:iam::123456789012:user/bob".into(),
1556            user_id: "AIDABOB".into(),
1557            account_id: "123456789012".into(),
1558            principal_type: PrincipalType::User,
1559            source_identity: None,
1560            tags: None,
1561        };
1562        assert_eq!(aws_username_from_principal(&p), Some("bob".into()));
1563    }
1564
1565    #[test]
1566    fn principal_type_label_covers_federated_and_unknown() {
1567        assert_eq!(
1568            principal_type_label(PrincipalType::FederatedUser),
1569            "FederatedUser"
1570        );
1571        assert_eq!(principal_type_label(PrincipalType::Unknown), "Unknown");
1572    }
1573
1574    #[test]
1575    fn build_condition_context_marks_secure_when_flag_set() {
1576        let p = Principal {
1577            arn: "arn:aws:iam::123456789012:user/alice".into(),
1578            user_id: "AIDAALICE".into(),
1579            account_id: "123456789012".into(),
1580            principal_type: PrincipalType::User,
1581            source_identity: None,
1582            tags: None,
1583        };
1584        let ctx = build_condition_context(&p, None, "us-west-2", true);
1585        assert_eq!(ctx.aws_secure_transport, Some(true));
1586        assert!(ctx.aws_source_ip.is_none());
1587        assert_eq!(ctx.aws_requested_region.as_deref(), Some("us-west-2"));
1588    }
1589
1590    #[test]
1591    fn is_secure_transport_case_insensitive() {
1592        let mut headers = http::HeaderMap::new();
1593        headers.insert("x-forwarded-proto", "HTTPS".parse().unwrap());
1594        assert!(is_secure_transport(&headers));
1595    }
1596
1597    #[test]
1598    fn is_secure_transport_non_ascii_bytes_false() {
1599        let mut headers = http::HeaderMap::new();
1600        headers.insert(
1601            "x-forwarded-proto",
1602            http::HeaderValue::from_bytes(&[0xFF, 0xFE]).unwrap(),
1603        );
1604        assert!(!is_secure_transport(&headers));
1605    }
1606
1607    #[test]
1608    fn protocol_ext_error_status_is_bad_request() {
1609        assert_eq!(AwsProtocol::Query.error_status(), StatusCode::BAD_REQUEST);
1610        assert_eq!(AwsProtocol::Json.error_status(), StatusCode::BAD_REQUEST);
1611        assert_eq!(AwsProtocol::Rest.error_status(), StatusCode::BAD_REQUEST);
1612        assert_eq!(
1613            AwsProtocol::RestJson.error_status(),
1614            StatusCode::BAD_REQUEST
1615        );
1616    }
1617
1618    #[test]
1619    fn build_error_response_json_has_json_content_type() {
1620        let resp = build_error_response(
1621            StatusCode::BAD_REQUEST,
1622            "TestCode",
1623            "test msg",
1624            "req-1",
1625            AwsProtocol::Json,
1626        );
1627        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1628        let ct = resp
1629            .headers()
1630            .get("content-type")
1631            .unwrap()
1632            .to_str()
1633            .unwrap();
1634        assert!(ct.contains("json"));
1635        let rid = resp
1636            .headers()
1637            .get("x-amzn-requestid")
1638            .unwrap()
1639            .to_str()
1640            .unwrap();
1641        assert_eq!(rid, "req-1");
1642    }
1643
1644    #[test]
1645    fn build_error_response_rest_returns_xml_content_type() {
1646        let resp = build_error_response(
1647            StatusCode::NOT_FOUND,
1648            "NoSuchBucket",
1649            "bucket missing",
1650            "req-2",
1651            AwsProtocol::Rest,
1652        );
1653        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1654        let ct = resp
1655            .headers()
1656            .get("content-type")
1657            .unwrap()
1658            .to_str()
1659            .unwrap();
1660        assert!(ct.contains("xml"));
1661    }
1662
1663    #[test]
1664    fn build_error_response_query_returns_xml() {
1665        let resp = build_error_response(
1666            StatusCode::BAD_REQUEST,
1667            "InvalidParameter",
1668            "bad param",
1669            "req-3",
1670            AwsProtocol::Query,
1671        );
1672        let ct = resp
1673            .headers()
1674            .get("content-type")
1675            .unwrap()
1676            .to_str()
1677            .unwrap();
1678        assert!(ct.contains("xml"));
1679    }
1680
1681    /// Regression for issue #1539: multi-line backend errors (e.g. podman
1682    /// stderr) used to panic the dispatcher when stuffed into the
1683    /// `x-amz-error-message` HTTP header. The response must build cleanly
1684    /// and the header value must not contain control characters.
1685    #[test]
1686    fn build_error_response_with_multiline_message_does_not_panic() {
1687        let resp = build_error_response(
1688            StatusCode::INTERNAL_SERVER_ERROR,
1689            "ServiceException",
1690            "Lambda execution failed: container failed to start: docker start failed: \
1691             Error: unable to start container \"abc\": \
1692             failed to create new hosts file:\nhost-gateway is empty\n",
1693            "req-multi",
1694            AwsProtocol::Json,
1695        );
1696        assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
1697        let msg = resp
1698            .headers()
1699            .get("x-amz-error-message")
1700            .expect("x-amz-error-message must be set even when input contains newlines")
1701            .to_str()
1702            .unwrap();
1703        assert!(!msg.contains('\n'));
1704        assert!(!msg.contains('\r'));
1705        assert!(msg.contains("Lambda execution failed"));
1706        assert!(msg.contains("host-gateway is empty"));
1707    }
1708
1709    #[test]
1710    fn build_error_response_with_control_chars_strips_them() {
1711        let resp = build_error_response(
1712            StatusCode::BAD_REQUEST,
1713            "Code\twith\ttabs",
1714            "msg\x00with\x01nulls",
1715            "req-ctrl",
1716            AwsProtocol::Json,
1717        );
1718        let code = resp
1719            .headers()
1720            .get("x-amz-error-code")
1721            .unwrap()
1722            .to_str()
1723            .unwrap();
1724        let msg = resp
1725            .headers()
1726            .get("x-amz-error-message")
1727            .unwrap()
1728            .to_str()
1729            .unwrap();
1730        assert!(!code.contains('\t'));
1731        assert!(!msg.contains('\x00'));
1732        assert!(!msg.contains('\x01'));
1733    }
1734
1735    #[test]
1736    fn sanitize_header_value_truncates_long_input() {
1737        let huge = "x".repeat(5_000);
1738        let out = sanitize_header_value(&huge);
1739        assert!(out.len() <= 1024);
1740    }
1741
1742    #[test]
1743    fn sanitize_header_value_collapses_consecutive_control_runs() {
1744        let out = sanitize_header_value("a\n\n\n\rb");
1745        assert_eq!(out, "a b");
1746    }
1747
1748    #[test]
1749    fn dispatch_config_carries_opt_in_flags() {
1750        let cfg = DispatchConfig {
1751            region: "eu-west-1".to_string(),
1752            account_id: "000000000000".to_string(),
1753            verify_sigv4: true,
1754            iam_mode: IamMode::Strict,
1755            credential_resolver: None,
1756            policy_evaluator: None,
1757            resource_policy_provider: None,
1758            scp_resolver: None,
1759        };
1760        assert!(cfg.verify_sigv4);
1761        assert!(cfg.iam_mode.is_strict());
1762        assert!(cfg.resource_policy_provider.is_none());
1763        assert!(cfg.scp_resolver.is_none());
1764    }
1765
1766    fn s3_sigv4_headers() -> http::HeaderMap {
1767        let mut headers = http::HeaderMap::new();
1768        headers.insert(
1769            "authorization",
1770            "AWS4-HMAC-SHA256 Credential=test/20240101/us-east-1/s3/aws4_request, \
1771             SignedHeaders=host, Signature=fake"
1772                .parse()
1773                .unwrap(),
1774        );
1775        headers
1776    }
1777
1778    #[test]
1779    fn streaming_route_path_style_s3_put_object() {
1780        let headers = s3_sigv4_headers();
1781        assert_eq!(
1782            streaming_route(
1783                &http::Method::PUT,
1784                "/my-bucket/key.txt",
1785                &headers,
1786                &HashMap::new(),
1787            ),
1788            Some(("s3", "")),
1789        );
1790    }
1791
1792    #[test]
1793    fn streaming_route_path_style_create_bucket_skipped() {
1794        // `PUT /bucket` (no trailing key) is CreateBucket — must NOT
1795        // hit the streaming path.
1796        let headers = s3_sigv4_headers();
1797        assert_eq!(
1798            streaming_route(&http::Method::PUT, "/my-bucket", &headers, &HashMap::new(),),
1799            None,
1800        );
1801    }
1802
1803    #[test]
1804    fn streaming_route_virtual_hosted_s3_put_object() {
1805        let mut headers = s3_sigv4_headers();
1806        headers.insert(
1807            "host",
1808            "vhost-bucket.s3.us-east-1.localhost.localstack.cloud:4566"
1809                .parse()
1810                .unwrap(),
1811        );
1812        // Virtual-hosted PUT has no bucket in the URL path (`/<key>`),
1813        // so the slash check used for path-style would reject it. The
1814        // Host parser confirms this is virtual-hosted S3 and the key
1815        // flows through the streaming dispatch.
1816        assert_eq!(
1817            streaming_route(&http::Method::PUT, "/hello.txt", &headers, &HashMap::new(),),
1818            Some(("s3", "")),
1819        );
1820    }
1821
1822    #[test]
1823    fn streaming_route_virtual_hosted_s3_root_skipped() {
1824        // `PUT /` against a virtual-hosted Host = CreateBucket, which
1825        // is handled buffered. Empty path-after-slash must short-circuit.
1826        let mut headers = s3_sigv4_headers();
1827        headers.insert(
1828            "host",
1829            "vhost-bucket.s3.us-east-1.localhost.localstack.cloud:4566"
1830                .parse()
1831                .unwrap(),
1832        );
1833        assert_eq!(
1834            streaming_route(&http::Method::PUT, "/", &headers, &HashMap::new()),
1835            None,
1836        );
1837    }
1838
1839    #[test]
1840    fn streaming_route_ecr_blob_upload() {
1841        let headers = http::HeaderMap::new();
1842        assert_eq!(
1843            streaming_route(
1844                &http::Method::PATCH,
1845                "/v2/my-repo/blobs/uploads/abcd1234",
1846                &headers,
1847                &HashMap::new(),
1848            ),
1849            Some(("ecr", "")),
1850        );
1851        assert_eq!(
1852            streaming_route(
1853                &http::Method::PUT,
1854                "/v2/my-repo/blobs/uploads/abcd1234",
1855                &headers,
1856                &HashMap::new(),
1857            ),
1858            Some(("ecr", "")),
1859        );
1860    }
1861
1862    #[test]
1863    fn streaming_route_presigned_v4_s3_put() {
1864        let headers = http::HeaderMap::new();
1865        let mut query_params = HashMap::new();
1866        query_params.insert(
1867            "X-Amz-Credential".to_string(),
1868            "test/20240101/us-east-1/s3/aws4_request".to_string(),
1869        );
1870        assert_eq!(
1871            streaming_route(
1872                &http::Method::PUT,
1873                "/my-bucket/key.txt",
1874                &headers,
1875                &query_params,
1876            ),
1877            Some(("s3", "")),
1878        );
1879    }
1880
1881    #[test]
1882    fn streaming_route_non_s3_auth_header_skipped() {
1883        // Same path shape but the SigV4 service is lambda — must not
1884        // wire the streaming dispatch (Lambda has its own buffered path).
1885        let mut headers = http::HeaderMap::new();
1886        headers.insert(
1887            "authorization",
1888            "AWS4-HMAC-SHA256 Credential=test/20240101/us-east-1/lambda/aws4_request, \
1889             SignedHeaders=host, Signature=fake"
1890                .parse()
1891                .unwrap(),
1892        );
1893        assert_eq!(
1894            streaming_route(
1895                &http::Method::PUT,
1896                "/my-bucket/key.txt",
1897                &headers,
1898                &HashMap::new(),
1899            ),
1900            None,
1901        );
1902    }
1903
1904    #[test]
1905    fn streaming_route_get_skipped() {
1906        let headers = s3_sigv4_headers();
1907        assert_eq!(
1908            streaming_route(
1909                &http::Method::GET,
1910                "/my-bucket/key.txt",
1911                &headers,
1912                &HashMap::new(),
1913            ),
1914            None,
1915        );
1916    }
1917}