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