From dd9095584ac117d9ff58e3efa0ccde3ac7fd9335 Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Wed, 12 Aug 2026 16:20:24 +0200 Subject: [PATCH 1/9] [golang] Add apim weblog variant for azure/apim-callout Add a Go weblog variant `apim` exercising dd-trace-go/contrib/azure/apim-callout, at activation parity with `envoy`. Azure APIM has no runnable container, so the gateway role is filled by a system-tests-owned, stdlib-only Go shim that reimplements APIM `send-request` behavior. It never parses policy XML. test runner --> apim-gateway (shim, :80 -> host 127.0.0.1:7777) |-> apim-callout (POST :8080, health :8081) `-> http-app (:8080) The shim defaults to the 4-call flow (deferred bodies). A control header `X-Datadog-Apim-Body-Mode: inline` switches both sides to inline delivery, which suppresses `allowed-body-size` and collapses the exchange to 2 calls. The header is stripped before building the callout payload and before forwarding upstream. The shared suite never sends it. Unlike the APIM policy's ignore-error="true", the shim fails closed (502 + stderr) on detectable transport/protocol failure, so system-tests can assert on it. The callout's HTTP 200 `{}` for processor errors is indistinguishable from a legitimate phase-2/4 success, so the one available check is requiring a non-empty request-id in the phase-1 response. All 89 `envoy:` manifest rows are widened to `envoy, apim:`, transcribing enabled rows as enabled. Widening all 89 rather than only the non-redundant ones is required, not just harmless: 29 rows have a `"*"` sibling and in every one the `"*"` value differs from envoy's, so for 13 of them the fallback would have given apim an outright wrong declaration. `containers.py:998` also gains `apim`, otherwise get_image_list opens a nonexistent apim.Dockerfile and breaks get-image-list.py while every acceptance criterion still passes. --- docs/understand/scenarios/README.md | 20 +- manifests/golang.yml | 178 +++---- tests/external_processing/test_apm.py | 2 +- tests/test_semantic_conventions.py | 1 + tests/test_the_test/test_ci_orchestrator.py | 2 +- utils/_context/containers.py | 89 +++- utils/_context/weblog_infrastructure.py | 18 +- utils/build/docker/golang/apim/go.mod | 3 + utils/build/docker/golang/apim/main.go | 357 +++++++++++++ utils/build/docker/golang/apim/main_test.go | 503 ++++++++++++++++++ utils/build/docker/golang/go.work | 1 + utils/build/docker/golang/weblog_metadata.yml | 3 + utils/scripts/load-binary.sh | 4 + 13 files changed, 1077 insertions(+), 104 deletions(-) create mode 100644 utils/build/docker/golang/apim/go.mod create mode 100644 utils/build/docker/golang/apim/main.go create mode 100644 utils/build/docker/golang/apim/main_test.go diff --git a/docs/understand/scenarios/README.md b/docs/understand/scenarios/README.md index 68570c3325a..6852f7c8659 100644 --- a/docs/understand/scenarios/README.md +++ b/docs/understand/scenarios/README.md @@ -70,7 +70,7 @@ Please note that it requires the docker daemon to support IPv6. It should be ok A user has seen his network function altered after running it on a linux laptop (to be investigated). If it happen, `docker network prune` may solve the issue. -### Go proxies (Envoy and HAProxy) scenario +### Go proxies (Envoy, HAProxy, and APIM) scenario ```mermaid flowchart LR @@ -78,18 +78,26 @@ flowchart LR A("Test runner") B("Proxy (Envoy or HAProxy)") C("Go security processor") - D("HTTP app") - E("Proxy") - F("Agent") - G("Backend") + D("http-app
:8080") + E("apim-gateway
(shim, :80 -> host 127.0.0.1:7777)") + F("apim-callout
(POST :8080, health :8081)
library under test") + G("Proxy") + H("Agent") + I("Backend") %% Edge connections between nodes A --> B --> D B --> C --> B - C --> E --> F --> G + A --> E + E --> D + E --> F + C --> G + F --> G --> H --> I %% D -- Mermaid js --> I --> J ``` +For a proxy variant with `build_mode: none`, there is nothing to build: run `./utils/scripts/load-binary.sh golang` to write the image pointers, then `./run.sh --weblog `. + ## Scenario lifecycle System tests spawn several services before starting. Here is the lifecycle: diff --git a/manifests/golang.yml b/manifests/golang.yml index 2dfd3051a03..3af448dd57a 100644 --- a/manifests/golang.yml +++ b/manifests/golang.yml @@ -94,21 +94,21 @@ manifest: tests/appsec/api_security_testing/test_headers_collection.py::Test_SecurityTestingHeaders: - weblog_declaration: "*": v2.10.0-dev - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/appsec/api_security_testing/test_normalized_route.py::Test_NormalizedRoute: - weblog_declaration: - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) "*": missing_feature tests/appsec/api_security_testing/test_normalized_route.py::Test_NormalizedRouteMultiParamsInSegment: - weblog_declaration: - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) "*": missing_feature tests/appsec/api_security_testing/test_normalized_route.py::Test_NormalizedRouteOptionalParams: - weblog_declaration: - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) "*": missing_feature tests/appsec/iast/sink/test_code_injection.py::TestCodeInjection: missing_feature @@ -395,11 +395,11 @@ manifest: tests/appsec/test_agentic_onboarding.py::Test_AppsecAgenticOnboarding: - weblog_declaration: "*": v2.11.0-dev - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/appsec/test_alpha.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/test_alpha.py::Test_Basic: - weblog_declaration: @@ -498,7 +498,7 @@ manifest: tests/appsec/test_asm_standalone.py::Test_AppSecStandalone_APMDisabledMarker: missing_feature tests/appsec/test_asm_standalone.py::Test_AppSecStandalone_NotEnabled: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_asm_standalone.py::Test_AppSecStandalone_UpstreamPropagation_V2: v1.73.0-dev tests/appsec/test_asm_standalone.py::Test_IastStandalone_UpstreamPropagation_V2: missing_feature @@ -526,11 +526,11 @@ manifest: tests/appsec/test_automated_user_and_session_tracking.py::Test_Automated_User_Tracking: irrelevant (Go only supports SDK events) tests/appsec/test_blocking_addresses.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/test_blocking_addresses.py::Test_BlockingGraphqlResolvers: - weblog_declaration: - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) "*": missing_feature tests/appsec/test_blocking_addresses.py::Test_Blocking_client_ip: v1.51.0 @@ -548,7 +548,7 @@ manifest: net-http-span-pool: irrelevant tests/appsec/test_blocking_addresses.py::Test_Blocking_request_cookies::test_blocking_before: - weblog_declaration: - envoy: missing_feature (The endpoint /tag_value is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /tag_value is not implemented in the weblog) haproxy: missing_feature (The endpoint /tag_value is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_request_headers: - weblog_declaration: @@ -558,7 +558,7 @@ manifest: net-http-span-pool: irrelevant tests/appsec/test_blocking_addresses.py::Test_Blocking_request_headers::test_blocking_before: - weblog_declaration: - envoy: missing_feature (The endpoint /tag_value is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /tag_value is not implemented in the weblog) haproxy: missing_feature (The endpoint /tag_value is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_request_method: - weblog_declaration: @@ -568,7 +568,7 @@ manifest: net-http-span-pool: irrelevant tests/appsec/test_blocking_addresses.py::Test_Blocking_request_method::test_blocking_before: - weblog_declaration: - envoy: missing_feature (The endpoint /tag_value is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /tag_value is not implemented in the weblog) haproxy: missing_feature (The endpoint /tag_value is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_request_path_params: - weblog_declaration: @@ -578,15 +578,15 @@ manifest: net-http-span-pool: irrelevant tests/appsec/test_blocking_addresses.py::Test_Blocking_request_path_params::test_blocking: - weblog_declaration: - envoy: missing_feature (The endpoint /param is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /param is not implemented in the weblog) haproxy: missing_feature (The endpoint /param is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_request_path_params::test_blocking_before: - weblog_declaration: - envoy: missing_feature (The endpoint /param is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /param is not implemented in the weblog) haproxy: missing_feature (The endpoint /param is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_request_path_params::test_non_blocking: - weblog_declaration: - envoy: missing_feature (The endpoint /param is not implemented in the weblog (from the test_blocking test)) + envoy, apim: missing_feature (The endpoint /param is not implemented in the weblog (from the test_blocking test)) haproxy: missing_feature (The endpoint /param is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_request_query: - weblog_declaration: @@ -596,7 +596,7 @@ manifest: net-http-span-pool: irrelevant tests/appsec/test_blocking_addresses.py::Test_Blocking_request_query::test_blocking_before: - weblog_declaration: - envoy: missing_feature (The endpoint /tag_value is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /tag_value is not implemented in the weblog) haproxy: missing_feature (The endpoint /tag_value is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_request_uri: - weblog_declaration: @@ -606,7 +606,7 @@ manifest: net-http-span-pool: irrelevant tests/appsec/test_blocking_addresses.py::Test_Blocking_request_uri::test_blocking_before: - weblog_declaration: - envoy: missing_feature (The endpoint /tag_value is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /tag_value is not implemented in the weblog) haproxy: missing_feature (The endpoint /tag_value is not implemented in the weblog) tests/appsec/test_blocking_addresses.py::Test_Blocking_response_headers: missing_feature tests/appsec/test_blocking_addresses.py::Test_Blocking_response_status: # Modified by easy win activation script @@ -627,14 +627,14 @@ manifest: excluded_weblog: [gin] tests/appsec/test_blocking_addresses.py::Test_Blocking_user_id: - weblog_declaration: - envoy: irrelevant (not supported on proxies) + envoy, apim: irrelevant (not supported on proxies) haproxy: irrelevant (not supported on proxies) "*": v1.51.0 tests/appsec/test_blocking_addresses.py::Test_Suspicious_Request_Blocking: - weblog_declaration: "*": v2.3.0-dev net-http-orchestrion: v2.5.0 # Modified by easy win activation script - envoy: missing_feature (can't test with set tag) + envoy, apim: missing_feature (can't test with set tag) haproxy: missing_feature (can't test with set tag) tests/appsec/test_blocking_addresses.py::Test_Suspicious_Request_Blocking::test_blocking: # Created by easy win activation script - weblog_declaration: @@ -644,7 +644,7 @@ manifest: net-http-orchestrion: missing_feature (path params are not supported) tests/appsec/test_client_ip.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/test_client_ip.py::Test_StandardTagsClientIp: v1.44.1 tests/appsec/test_conf.py::Test_ConfigurationVariables: v2.0.0 @@ -654,12 +654,12 @@ manifest: tests/appsec/test_event_tracking.py::Test_CustomEvent: v1.47.0 tests/appsec/test_event_tracking.py::Test_CustomEvent::test_custom_event_event: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_event_tracking.py::Test_CustomEvent_Metrics: v2.1.0-dev tests/appsec/test_event_tracking.py::Test_CustomEvent_Metrics::test_user_login_success_event: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_event_tracking.py::Test_UserLoginFailureEvent: v1.47.0 tests/appsec/test_event_tracking.py::Test_UserLoginFailureEvent::test_user_login_failure_event: @@ -668,17 +668,17 @@ manifest: tests/appsec/test_event_tracking.py::Test_UserLoginFailureEvent_Metrics: v2.1.0-dev tests/appsec/test_event_tracking.py::Test_UserLoginFailureEvent_Metrics::test_user_login_success_event: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_event_tracking.py::Test_UserLoginSuccessEvent: - weblog_declaration: "*": v1.47.0 haproxy: missing_feature - envoy: missing_feature + envoy, apim: missing_feature tests/appsec/test_event_tracking.py::Test_UserLoginSuccessEvent_Metrics: v2.1.0-dev tests/appsec/test_event_tracking.py::Test_UserLoginSuccessEvent_Metrics::test_user_login_success_event: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_event_tracking_v2.py::Test_UserLoginFailureEventV2_HeaderCollection_AppsecDisabled: v2.2.0-dev # LANGPLAT-583 tests/appsec/test_event_tracking_v2.py::Test_UserLoginFailureEventV2_HeaderCollection_AppsecEnabled: v2.2.0-dev # LANGPLAT-583 @@ -747,31 +747,31 @@ manifest: tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Header_And_Network_Preprocessor: v1.69.0 tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Header_And_Network_Preprocessor::test_fingerprinting_header_blocking: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Header_Capability: v1.69.0 tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Network_Capability: v1.69.0 tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Session: v1.69.0 tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Session::test_session: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Session_Capability: v1.69.0 tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Session_Preprocessor: v1.69.0 tests/appsec/test_fingerprinting.py::Test_Fingerprinting_Session_Preprocessor::test_session_non_blocking: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_identify.py::Test_Basic: v1.37.0 tests/appsec/test_identify.py::Test_Basic::test_identify_tags_with_attack: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_inferred_spans.py::Test_Proxy_Inferred_Span_Tags: missing_feature tests/appsec/test_ip_blocking_full_denylist.py::Test_AppSecIPBlockingFullDenylist: - weblog_declaration: "*": v1.47.0 - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/test_logs.py::Test_Standardization: missing_feature tests/appsec/test_logs.py::Test_StandardizationBlockMode: missing_feature @@ -789,7 +789,7 @@ manifest: tests/appsec/test_remote_config_rule_changes.py::Test_UpdateRuleFileWithRemoteConfig: v2.0.0 tests/appsec/test_reports.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/test_reports.py::Test_ExtraTagsFromRule: - weblog_declaration: @@ -800,7 +800,7 @@ manifest: chi: v1.36.0 echo: v1.36.0 gin: v1.37.0 - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_reports.py::Test_RequestHeaders: - weblog_declaration: @@ -812,12 +812,12 @@ manifest: "*": v1.34.0 echo: v1.36.0 gin: v1.37.0 - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_request_blocking.py::Test_AppSecRequestBlocking: - weblog_declaration: "*": v1.50.0-rc.1 - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/test_runtime_activation.py::Test_RuntimeActivation: v1.69.0 tests/appsec/test_runtime_activation.py::Test_RuntimeActivationCapabilities: v1.69.0 @@ -831,12 +831,12 @@ manifest: tests/appsec/test_service_activation_metric.py::TestServiceActivationEnvVarConfigurationMetric: v2.4.0 tests/appsec/test_service_activation_metric.py::TestServiceActivationEnvVarConfigurationMetric::test_service_activation_metric: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_service_activation_metric.py::TestServiceActivationEnvVarMetric: v2.4.0 tests/appsec/test_service_activation_metric.py::TestServiceActivationEnvVarMetric::test_service_activation_metric: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_service_activation_metric.py::TestServiceActivationRemoteConfigMetric: v2.4.0 tests/appsec/test_service_activation_metric.py::TestServiceActivationRemoteConfigurationConfigMetric: v2.4.0 @@ -847,16 +847,16 @@ manifest: tests/appsec/test_trace_tagging.py::Test_TraceTaggingRules: v2.1.0-dev tests/appsec/test_trace_tagging.py::Test_TraceTaggingRules::test_rule_with_attributes_no_keep_event: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_trace_tagging.py::Test_TraceTaggingRules::test_rule_with_attributes_no_keep_no_event: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/test_trace_tagging.py::Test_TraceTaggingRulesRcCapability: v2.1.0-dev tests/appsec/test_traces.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/test_traces.py::Test_AppSecEventSpanTags: - weblog_declaration: @@ -864,7 +864,7 @@ manifest: gin: v1.37.0 tests/appsec/test_traces.py::Test_AppSecEventSpanTags::test_header_collection: - weblog_declaration: - envoy: irrelevant (test) + envoy, apim: irrelevant (test) haproxy: irrelevant (test) tests/appsec/test_traces.py::Test_AppSecObfuscator: v1.38.0 tests/appsec/test_traces.py::Test_CollectDefaultRequestHeader: @@ -877,7 +877,7 @@ manifest: gin: v1.37.0 tests/appsec/test_traces.py::Test_CollectRespondHeaders::test_header_collection: - weblog_declaration: - envoy: missing_feature (The endpoint /headers is not implemented in the weblog) + envoy, apim: missing_feature (The endpoint /headers is not implemented in the weblog) haproxy: missing_feature (The endpoint /headers is not implemented in the weblog) tests/appsec/test_traces.py::Test_ExternalWafRequestsIdentification: v1.63.0 tests/appsec/test_traces.py::Test_RetainTraces: @@ -889,28 +889,28 @@ manifest: - weblog_declaration: "*": v1.36.0 gin: v1.37.0 - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/appsec/waf/test_addresses.py::Test_BodyJson: v1.37.0 tests/appsec/waf/test_addresses.py::Test_BodyJson::test_json_array: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature tests/appsec/waf/test_addresses.py::Test_BodyJson::test_json_value: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature tests/appsec/waf/test_addresses.py::Test_BodyUrlEncoded: v1.37.0 tests/appsec/waf/test_addresses.py::Test_BodyUrlEncoded::test_body_value: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_addresses.py::Test_BodyXml: v1.37.0 tests/appsec/waf/test_addresses.py::Test_BodyXml::test_xml_attr_value: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_addresses.py::Test_BodyXml::test_xml_content: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_addresses.py::Test_Cookies: - weblog_declaration: @@ -934,7 +934,7 @@ manifest: gin: v1.37.0 tests/appsec/waf/test_addresses.py::Test_Headers::test_specific_key3: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature tests/appsec/waf/test_addresses.py::Test_IoFsFileWrite: "irrelevant (Java-only address: server.io.fs.file_write)" tests/appsec/waf/test_addresses.py::Test_PathParams: - weblog_declaration: @@ -945,7 +945,7 @@ manifest: net-http-span-pool: irrelevant (net-http doesn't handle path params) tests/appsec/waf/test_addresses.py::Test_PathParams::test_security_scanner: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_addresses.py::Test_ResponseStatus: - weblog_declaration: @@ -953,7 +953,7 @@ manifest: gin: v1.37.0 tests/appsec/waf/test_addresses.py::Test_ResponseStatus::test_basic: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_addresses.py::Test_UrlQuery: - weblog_declaration: @@ -969,7 +969,7 @@ manifest: tests/appsec/waf/test_addresses.py::Test_gRPC: v1.36.0 tests/appsec/waf/test_addresses.py::Test_gRPC::test_basic: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_blocking.py::Test_Blocking: v1.50.0-rc.1 tests/appsec/waf/test_blocking.py::Test_Blocking::test_accept_full_html: @@ -1000,7 +1000,7 @@ manifest: gin: v1.37.0 tests/appsec/waf/test_miscs.py::Test_404::test_404: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_miscs.py::Test_CorrectOptionProcessing: - weblog_declaration: @@ -1032,7 +1032,7 @@ manifest: gin: v1.37.0 tests/appsec/waf/test_rules.py::Test_DiscoveryScan::test_security_scan: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/appsec/waf/test_rules.py::Test_HttpProtocol: - weblog_declaration: @@ -1087,7 +1087,7 @@ manifest: tests/appsec/waf/test_truncation.py::Test_Truncation: v2.0.0 tests/appsec/waf/test_truncation.py::Test_Truncation::test_truncation: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/auto_inject/test_auto_inject_install.py::TestContainerAutoInjectInstallScriptAppsec: v2.0.0 tests/auto_inject/test_auto_inject_install.py::TestHostAutoInjectInstallScriptAppsec: v2.0.0 @@ -1516,7 +1516,7 @@ manifest: tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_record_exception_sets_handling_stack_in_go: - weblog_declaration: "*": v2.7.0-dev.1 - envoy: irrelevant + envoy, apim: irrelevant tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_set_attribute_remapping_httpresponsestatuscode: - declaration: missing_feature (Implemented in 1.65.0) component_version: <1.65.0 @@ -1654,26 +1654,26 @@ manifest: tests/test_config_consistency.py::Test_Config_ClientTagQueryString_Empty: v1.72.0-dev tests/test_config_consistency.py::Test_Config_ClientTagQueryString_Empty::test_query_string_redaction_unset: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_config_consistency.py::Test_Config_HttpClientErrorStatuses_Default: v1.69.0 tests/test_config_consistency.py::Test_Config_HttpClientErrorStatuses_Default::test_status_code_400: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_config_consistency.py::Test_Config_HttpClientErrorStatuses_Default::test_status_code_500: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_config_consistency.py::Test_Config_HttpClientErrorStatuses_FeatureFlagCustom: v1.69.0 tests/test_config_consistency.py::Test_Config_HttpServerErrorStatuses_Default: v1.67.0 tests/test_config_consistency.py::Test_Config_HttpServerErrorStatuses_Default::test_status_code_400: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_config_consistency.py::Test_Config_HttpServerErrorStatuses_Default::test_status_code_500: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_config_consistency.py::Test_Config_HttpServerErrorStatuses_FeatureFlagCustom: v1.69.0 tests/test_config_consistency.py::Test_Config_IntegrationEnabled_False: irrelevant (not applicable to Go because of how they do auto instrumentation) @@ -1738,7 +1738,7 @@ manifest: tests/test_config_consistency.py::Test_Config_UnifiedServiceTagging_CustomService: - weblog_declaration: "*": v1.67.0 - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/test_config_consistency.py::Test_Config_UnifiedServiceTagging_CustomService::test_specified_service_name: - weblog_declaration: @@ -1746,7 +1746,7 @@ manifest: tests/test_config_consistency.py::Test_Config_UnifiedServiceTagging_Default: v1.67.0 tests/test_config_consistency.py::Test_Config_UnifiedServiceTagging_Default::test_default_service_name: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_data_integrity.py::Test_LibraryHeaders: v1.60.0.dev0 tests/test_data_integrity.py::Test_LibraryHeaders::test_datadog_entity_id: irrelevant (implemented but not testable) @@ -1787,26 +1787,26 @@ manifest: - declaration: bug (APMRP-360) component_version: <=1.41.0 - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_identify.py::Test_Basic::test_identify_tags_with_attack: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_identify.py::Test_Propagate: v1.48.0-rc.1 tests/test_identify.py::Test_Propagate::test_identify_tags_outgoing: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_identify.py::Test_Propagate_Legacy: v1.41.0 tests/test_identify.py::Test_Propagate_Legacy::test_identify_tags_outgoing: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_library_conf.py::Test_ExtractBehavior_Default: - weblog_declaration: "*": v2.10.0-dev - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_library_conf.py::Test_ExtractBehavior_Ignore: v2.10.0-dev tests/test_library_conf.py::Test_ExtractBehavior_Restart: v2.10.0-dev @@ -1815,7 +1815,7 @@ manifest: tests/test_library_conf.py::Test_HeaderTags: v1.53.0 tests/test_library_conf.py::Test_HeaderTags::test_trace_header_tags_basic: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_library_conf.py::Test_HeaderTags_Colon_Leading: v1.53.0 tests/test_library_conf.py::Test_HeaderTags_Colon_Trailing: v1.70.0 @@ -1840,52 +1840,52 @@ manifest: tests/test_otel_tracestate_sampling.py::Test_ForceKeepClearsTh: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_ForwardInboundOtUnchanged: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_ForwardInboundOtUnchangedWhenDropped: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_MalformedOtHandling: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_PrecisionBoundaryDecision_Rate0_05: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_PrecisionBoundaryDecision_Rate0_1: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_PreserveDdAndOtherVendors: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_SampledWithoutOtNotFabricated: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_ThOnlyDoesNotFabricateRv: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otel_tracestate_sampling.py::Test_ThOnlyDoesNotFabricateRvWhenDropped: - weblog_declaration: "*": missing_feature (APMAPI-2171) - envoy: irrelevant (not applicable to proxies) + envoy, apim: irrelevant (not applicable to proxies) haproxy: irrelevant (not applicable to proxies) tests/test_otlp_runtime_metrics.py::Test_OtlpRuntimeMetrics: v2.10.0-dev tests/test_profiling.py::Test_Profile::test_process_tags_svc: missing_feature @@ -1906,27 +1906,27 @@ manifest: tests/test_sampling_rates.py::Test_SamplingRates: v1.36.2 tests/test_scrubbing.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/test_scrubbing.py::Test_UrlField::test_main: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_scrubbing.py::Test_UrlQuery: v1.40.0 tests/test_semantic_conventions.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/test_semantic_conventions.py::Test_Meta: v1.45.0 tests/test_semantic_conventions.py::Test_Meta::test_meta_component_tag: - weblog_declaration: - envoy: v2.3.0 + envoy, apim: v2.3.0 tests/test_semantic_conventions.py::Test_Meta::test_meta_http_url: - component_version: <1.69.0-dev declaration: bug (APMRP-360) tests/test_semantic_conventions.py::Test_MetaDatadogTags::test_meta_dd_tags: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_span_events.py::Test_SpanEvents_WithAgentSupport: - weblog_declaration: @@ -1944,7 +1944,7 @@ manifest: declaration: missing_feature (span got an event property set as an empty list) tests/test_standard_tags.py: - weblog_declaration: - envoy: v1.72.0 + envoy, apim: v1.72.0 haproxy: v2.4.0 tests/test_standard_tags.py::Test_StandardTagsClientIp: v1.46.0 tests/test_standard_tags.py::Test_StandardTagsClientIp::test_client_ip_vendor: @@ -1959,23 +1959,23 @@ manifest: tests/test_standard_tags.py::Test_StandardTagsRoute: v1.39.0 tests/test_standard_tags.py::Test_StandardTagsRoute::test_route: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_standard_tags.py::Test_StandardTagsStatusCode: v1.39.0 tests/test_standard_tags.py::Test_StandardTagsStatusCode::test_status_code: - weblog_declaration: - envoy: missing_feature + envoy, apim: missing_feature haproxy: missing_feature tests/test_standard_tags.py::Test_StandardTagsUrl: v1.40.0 tests/test_standard_tags.py::Test_StandardTagsUrl::test_multiple_matching_substring: - weblog_declaration: - envoy: irrelevant (tracer did not yet implemented the new version of query parameters obfuscation regex) + envoy, apim: irrelevant (tracer did not yet implemented the new version of query parameters obfuscation regex) haproxy: irrelevant (tracer did not yet implemented the new version of query parameters obfuscation regex) "*": missing_feature (tracer did not yet implemented the new version of query parameters obfuscation regex) tests/test_standard_tags.py::Test_StandardTagsUrl::test_url_with_sensitive_query_string: - weblog_declaration: haproxy: irrelevant (tracer did not yet implemented the new version of query parameters obfuscation regex) - envoy: irrelevant (tracer did not yet implemented the new version of query parameters obfuscation regex) + envoy, apim: irrelevant (tracer did not yet implemented the new version of query parameters obfuscation regex) "*": missing_feature (tracer did not yet implemented the new version of query parameters obfuscation regex) tests/test_standard_tags.py::Test_StandardTagsUserAgent: v1.39.0 tests/test_telemetry.py::Test_DependencyEnable: v1.73.0-dev @@ -2077,7 +2077,7 @@ manifest: tests/test_telemetry.py::Test_Telemetry::test_app_heartbeats_delays: - weblog_declaration: haproxy: missing_feature - envoy: missing_feature + envoy, apim: missing_feature tests/test_telemetry.py::Test_Telemetry::test_app_product_change: missing_feature (Weblog GET/enable_product and app-product-change event is not implemented yet.) tests/test_telemetry.py::Test_Telemetry::test_session_id_headers_across_forks: irrelevant (Go does not support fork; use test_session_id_headers_across_spawned instead) tests/test_telemetry.py::Test_Telemetry::test_session_id_headers_across_spawned: diff --git a/tests/external_processing/test_apm.py b/tests/external_processing/test_apm.py index 6190c6e599c..a10fac3a910 100644 --- a/tests/external_processing/test_apm.py +++ b/tests/external_processing/test_apm.py @@ -1,7 +1,7 @@ from utils import weblog, interfaces, features, context, irrelevant -@irrelevant(context.weblog_variant not in ("haproxy", "envoy")) +@irrelevant(context.weblog_variant not in ("haproxy", "envoy", "apim")) @features.go_proxies class Test_GoProxies_Tracing: def setup_correct_span_structure(self): diff --git a/tests/test_semantic_conventions.py b/tests/test_semantic_conventions.py index 11c504fcac2..326ece3ae08 100644 --- a/tests/test_semantic_conventions.py +++ b/tests/test_semantic_conventions.py @@ -36,6 +36,7 @@ "django-py3.13": "django", "python3.12": "django", "gin": "gin-gonic/gin", + "apim": "apim-callout", "haproxy": "haproxy-spoa", "gqlgen": "99designs/gqlgen", "graph-gophers": "graph-gophers/graphql-go", diff --git a/tests/test_the_test/test_ci_orchestrator.py b/tests/test_the_test/test_ci_orchestrator.py index 2add4b932c7..75b3a2dc9f4 100644 --- a/tests/test_the_test/test_ci_orchestrator.py +++ b/tests/test_the_test/test_ci_orchestrator.py @@ -286,7 +286,7 @@ def _is_uds_weblog(weblog: str) -> bool: return False # Go proxies - if weblog.name in ("envoy", "haproxy"): + if weblog.name in ("envoy", "haproxy", "apim"): if scenario.name not in ("DEFAULT", "APPSEC_BLOCKING"): return False diff --git a/utils/_context/containers.py b/utils/_context/containers.py index 5f60d546d42..f798513783e 100644 --- a/utils/_context/containers.py +++ b/utils/_context/containers.py @@ -995,7 +995,7 @@ def get_image_list(self, library: str | None, weblog: str | None) -> list[str]: if not library or not weblog: return result - if weblog in ("envoy", "haproxy"): + if weblog in ("envoy", "haproxy", "apim"): # Those are not based on a dockerfile. TODO : weblog abstraction return result @@ -1776,3 +1776,90 @@ def __init__( "retries": 10, }, ) + + +class ApimGatewayContainer(TestedContainer): + """Stand-in for the Azure APIM gateway: a stdlib-only Go shim compiled at container start""" + + def __init__(self) -> None: + super().__init__( + # already mirrored: mirror_images.lock.yaml:325, curated at mirror_images.yaml:90 + image_name="golang:1.25-alpine", + name="apim-gateway", + working_dir="/app", + # there is no dockerfile for this weblog (build_mode: none), the shim sources are + # mounted and compiled on start + command="go run .", + # GOTOOLCHAIN is deliberately not set: it is already `local` in this image + environment={"CGO_ENABLED": "0"}, + volumes={ + "./utils/build/docker/golang/apim/main.go": {"bind": "/app/main.go", "mode": "ro"}, + "./utils/build/docker/golang/apim/go.mod": {"bind": "/app/go.mod", "mode": "ro"}, + }, + ports={"80": ("127.0.0.1", weblog.port)}, + healthcheck={ + # golang:1.25-alpine ships busybox but no bash, so the /bin/bash + /dev/tcp + # healthcheck used by EnvoyContainer and HAProxyContainer is not usable here + "test": "wget -qO- http://localhost:80/", + # PROVISIONAL budget, to be tightened once a real cold start is measured: + # `go run .` compiles the shim from an empty build cache on every start, which is + # far slower than booting a prebuilt binary. interval and start_period are in + # NANOSECONDS (execute_command divides both by 1e9), so this is + # 15s + 46 * 3s ~= 153s ceiling, with a 3s readiness granularity. + "retries": 45, + "interval": 3_000_000_000, + "start_period": 15_000_000_000, + }, + ) + + +class ApimCalloutContainer(GoProcessorContainer): + """dd-trace-go apim-callout processor, driven by the apim-gateway shim""" + + def __init__( + self, + env: dict[str, str | None] | None = None, + volumes: dict[str, dict[str, str]] | None = None, + ) -> None: + try: + with open("binaries/golang-apim-callout-image", encoding="utf-8") as f: + image = f.read().strip() + logger.stdout(f"apim-callout image: {image} (from binaries/golang-apim-callout-image)") + except FileNotFoundError: + image = "ghcr.io/datadog/dd-trace-go/apim-callout:latest" + # make the downgrade loud: without the pointer file we test released code instead of + # the commit under test + logger.stdout( + f"WARNING: binaries/golang-apim-callout-image not found, falling back to released image {image}" + ) + + environment: dict[str, str | None] = { + "DD_APPSEC_ENABLED": "true", + "DD_SERVICE": "service_test", + "DD_ENV": "system-tests", + "DD_AGENT_HOST": "proxy", + "DD_TRACE_AGENT_PORT": str(ProxyPorts.weblog), + # not inherited: this lives in ExternalProcessingContainer, not in GoProcessorContainer + "DD_APPSEC_WAF_TIMEOUT": "1s", + # required, do not remove: the callout's own getDefaultEnvVars() defaults this to + # "false" whenever the variable is empty, which rate-limits ordinary traces to one per + # minute and makes APM assertions nondeterministic + "DD_APM_TRACING_ENABLED": "true", + } + + if env: + environment.update(env) + + if volumes is None: + volumes = {} + + super().__init__( + image_name=image, + name="apim-callout", + volumes=volumes, + environment=environment, + healthcheck={ + "test": "wget -qO- http://localhost:8081/", + "retries": 10, + }, + ) diff --git a/utils/_context/weblog_infrastructure.py b/utils/_context/weblog_infrastructure.py index 175ac6a95f5..799e476ddb3 100644 --- a/utils/_context/weblog_infrastructure.py +++ b/utils/_context/weblog_infrastructure.py @@ -17,12 +17,14 @@ DummyServerContainer, EnvoyContainer, HAProxyContainer, + ApimGatewayContainer, + ApimCalloutContainer, ExternalProcessingContainer, StreamProcessingOffloadContainer, GoProcessorContainer, ) -GoProxyWeblogs = Literal["envoy", "haproxy"] +GoProxyWeblogs = Literal["envoy", "haproxy", "apim"] class WeblogInfra(ABC): @@ -44,12 +46,13 @@ class EndToEndWeblogInfra(WeblogInfra): _go_proxy_weblog: GoProxyWeblogs | None = None _processor_container: GoProcessorContainer - """the Datadog library under test, running as an Envoy external processor - or an HAProxy SPOA agent. It intercepts HTTP traffic from the proxy runtime to apply - AppSec rules and emit traces. This is the "weblog" from the library's point of view.""" + """the Datadog library under test, running as a proxy-side callout: an Envoy external + processor, an HAProxy SPOA agent or an Azure APIM callout. It intercepts HTTP traffic from the + proxy runtime to apply AppSec rules and emit traces. This is the "weblog" from the library's + point of view.""" - _proxy_runtime_container: EnvoyContainer | HAProxyContainer - """the reverse proxy (Envoy or HAProxy) that sits in front of the + _proxy_runtime_container: TestedContainer + """the proxy runtime (Envoy, HAProxy or the APIM gateway) that sits in front of the dummy HTTP server and forwards requests through the processor. It is the actual HTTP entry point for test requests, exposing the weblog port to the test suite.""" @@ -124,6 +127,9 @@ def _configure_proxy_weblog(self, weblog: str) -> None: elif self._go_proxy_weblog == "haproxy": self._processor_container = StreamProcessingOffloadContainer() self._proxy_runtime_container = HAProxyContainer() + elif self._go_proxy_weblog == "apim": + self._processor_container = ApimCalloutContainer() + self._proxy_runtime_container = ApimGatewayContainer() self._processor_container.environment |= self._environment self._processor_container.volumes |= self._volumes diff --git a/utils/build/docker/golang/apim/go.mod b/utils/build/docker/golang/apim/go.mod new file mode 100644 index 00000000000..da5cb165a1e --- /dev/null +++ b/utils/build/docker/golang/apim/go.mod @@ -0,0 +1,3 @@ +module apimgateway + +go 1.25 diff --git a/utils/build/docker/golang/apim/main.go b/utils/build/docker/golang/apim/main.go new file mode 100644 index 00000000000..148a99db306 --- /dev/null +++ b/utils/build/docker/golang/apim/main.go @@ -0,0 +1,357 @@ +package main + +import ( + "bytes" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "time" +) + +const ( + calloutEndpoint = "http://apim-callout:8080/" + upstreamEndpoint = "http://http-app:8080" + bodyModeHeader = "X-Datadog-Apim-Body-Mode" + + calloutTimeout = 3 * time.Second + upstreamTimeout = 10 * time.Second + + phaseRequestHeaders = "" + phaseRequestBody = "" + phaseResponseHeaders = "" + phaseResponseBody = "" +) + +// calloutMessage represents the JSON body sent by the gateway on POST /. +// The Addresses field is phase-dependent and decoded separately. +type calloutMessage struct { + Addresses json.RawMessage `json:"addresses"` + Gateway string `json:"gateway,omitempty"` + RequestID string `json:"request-id,omitempty"` + Phase string `json:"phase,omitempty"` +} + +// calloutResult represents the JSON response returned to the gateway. +type calloutResult struct { + RequestID string `json:"request-id,omitempty"` + PropagateHeaders map[string][]string `json:"propagate-headers,omitempty"` + AllowedBodySize *int `json:"allowed-body-size,omitempty"` + Block *blockResult `json:"block,omitempty"` +} + +// blockResult represents a blocking decision sent back to the gateway. +type blockResult struct { + Status int `json:"status"` + Headers map[string][]string `json:"headers,omitempty"` + Content string `json:"content,omitempty"` +} + +// addressesRequestHeaders holds the phase-dependent addresses for the request headers phase. +type addressesRequestHeaders struct { + Method string `json:"method"` + Scheme string `json:"scheme"` + Authority string `json:"authority"` + Path string `json:"path"` + RemoteAddr string `json:"remote_addr"` + Headers map[string][]string `json:"headers"` + Body json.RawMessage `json:"body,omitempty"` +} + +// addressesResponseHeaders holds the phase-dependent addresses for the response headers phase. +type addressesResponseHeaders struct { + StatusCode int `json:"status_code"` + Headers map[string][]string `json:"headers"` + Body json.RawMessage `json:"body,omitempty"` +} + +// addressesBody holds the phase-dependent addresses for the body phase. +type addressesBody struct { + Body json.RawMessage `json:"body"` +} + +type gateway struct { + calloutURL string + upstreamURL string + calloutClient *http.Client + upstreamClient *http.Client + stderr io.Writer +} + +func main() { + if err := http.ListenAndServe(":80", newGateway()); err != nil { + fmt.Fprintln(os.Stderr, err) + } +} + +func newGateway() *gateway { + return &gateway{ + calloutURL: calloutEndpoint, + upstreamURL: upstreamEndpoint, + calloutClient: &http.Client{Timeout: calloutTimeout}, + upstreamClient: &http.Client{Timeout: upstreamTimeout}, + stderr: os.Stderr, + } +} + +func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { + requestBody, err := io.ReadAll(r.Body) + if err != nil { + g.failClosed(w) + return + } + defer r.Body.Close() + + inline := r.Header.Get(bodyModeHeader) == "inline" + r.Header.Del(bodyModeHeader) + + requestAddresses := addressesRequestHeaders{ + Method: r.Method, + Scheme: requestScheme(r), + Authority: r.Host, + Path: r.URL.RequestURI(), + RemoteAddr: r.RemoteAddr, + Headers: map[string][]string(r.Header.Clone()), + } + if inline { + requestAddresses.Body, err = encodedBody(requestBody, nil) + if err != nil { + g.failClosed(w) + return + } + } + + phase1, err := g.callout(phaseRequestHeaders, "", requestAddresses) + if err != nil { + g.failClosed(w) + return + } + if phase1.Block != nil { + if err := writeBlock(w, phase1.Block); err != nil { + g.failClosed(w) + } + return + } + + requestID := phase1.RequestID + applyHeaders(r.Header, phase1.PropagateHeaders) + if phase1.AllowedBodySize != nil { + body, err := encodedBody(requestBody, phase1.AllowedBodySize) + if err != nil { + g.failClosed(w) + return + } + phase2, err := g.callout(phaseRequestBody, requestID, addressesBody{Body: body}) + if err != nil { + g.failClosed(w) + return + } + if phase2.Block != nil { + if err := writeBlock(w, phase2.Block); err != nil { + g.failClosed(w) + } + return + } + } + + upstreamResponse, err := g.callUpstream(r, requestBody) + if err != nil { + g.failClosed(w) + return + } + defer upstreamResponse.Body.Close() + + responseBody, err := io.ReadAll(upstreamResponse.Body) + if err != nil { + g.failClosed(w) + return + } + + responseAddresses := addressesResponseHeaders{ + StatusCode: upstreamResponse.StatusCode, + Headers: map[string][]string(upstreamResponse.Header.Clone()), + } + if inline { + responseAddresses.Body, err = encodedBody(responseBody, nil) + if err != nil { + g.failClosed(w) + return + } + } + + phase3, err := g.callout(phaseResponseHeaders, requestID, responseAddresses) + if err != nil { + g.failClosed(w) + return + } + if phase3.Block != nil { + if err := writeBlock(w, phase3.Block); err != nil { + g.failClosed(w) + } + return + } + if phase3.AllowedBodySize != nil { + body, err := encodedBody(responseBody, phase3.AllowedBodySize) + if err != nil { + g.failClosed(w) + return + } + phase4, err := g.callout(phaseResponseBody, requestID, addressesBody{Body: body}) + if err != nil { + g.failClosed(w) + return + } + if phase4.Block != nil { + if err := writeBlock(w, phase4.Block); err != nil { + g.failClosed(w) + } + return + } + } + + writeUpstreamResponse(w, upstreamResponse, responseBody) +} + +func requestScheme(r *http.Request) string { + if r.TLS != nil { + return "https" + } + return "http" +} + +func (g *gateway) callout(phase, requestID string, addresses any) (calloutResult, error) { + addressesJSON, err := json.Marshal(addresses) + if err != nil { + g.logCallout(phase, requestID, err) + return calloutResult{}, err + } + payload, err := json.Marshal(calloutMessage{ + Addresses: addressesJSON, + RequestID: requestID, + Phase: phase, + }) + if err != nil { + g.logCallout(phase, requestID, err) + return calloutResult{}, err + } + + request, err := http.NewRequest(http.MethodPost, g.calloutURL, bytes.NewReader(payload)) + if err != nil { + g.logCallout(phase, requestID, err) + return calloutResult{}, err + } + request.Header.Set("Content-Type", "application/json") + + response, err := g.calloutClient.Do(request) + if err != nil { + g.logCallout(phase, requestID, err) + return calloutResult{}, err + } + defer response.Body.Close() + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + err := fmt.Errorf("callout returned %s", response.Status) + g.logCallout(phase, requestID, err) + return calloutResult{}, err + } + + var result calloutResult + if err := decodeJSON(response.Body, &result); err != nil { + g.logCallout(phase, requestID, err) + return calloutResult{}, err + } + if phase == phaseRequestHeaders && result.Block == nil && result.RequestID == "" { + err := fmt.Errorf("phase 1 callout response has no request-id") + g.logCallout(phase, result.RequestID, err) + return calloutResult{}, err + } + + logRequestID := requestID + if phase == phaseRequestHeaders { + logRequestID = result.RequestID + } + g.logCallout(phase, logRequestID, nil) + return result, nil +} + +func decodeJSON(body io.Reader, result *calloutResult) error { + decoder := json.NewDecoder(body) + if err := decoder.Decode(result); err != nil { + return err + } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return fmt.Errorf("callout response contains multiple JSON values") + } + return err + } + return nil +} + +func (g *gateway) callUpstream(r *http.Request, body []byte) (*http.Response, error) { + request, err := http.NewRequestWithContext(r.Context(), r.Method, g.upstreamURL+r.URL.RequestURI(), bytes.NewReader(body)) + if err != nil { + return nil, err + } + request.Header = r.Header.Clone() + request.Host = r.Host + return g.upstreamClient.Do(request) +} + +func encodedBody(body []byte, limit *int) (json.RawMessage, error) { + if limit != nil { + if *limit < 0 { + return nil, fmt.Errorf("negative allowed-body-size") + } + if len(body) > *limit { + body = body[:*limit] + } + } + return json.Marshal(base64.StdEncoding.EncodeToString(body)) +} + +func applyHeaders(headers http.Header, additions map[string][]string) { + for name, values := range additions { + headers[http.CanonicalHeaderKey(name)] = append([]string(nil), values...) + } +} + +func writeBlock(w http.ResponseWriter, block *blockResult) error { + content, err := base64.StdEncoding.DecodeString(block.Content) + if err != nil { + return err + } + if block.Status < http.StatusContinue || block.Status > 999 { + return fmt.Errorf("invalid block status %d", block.Status) + } + for name, values := range block.Headers { + w.Header()[name] = append([]string(nil), values...) + } + w.WriteHeader(block.Status) + _, err = w.Write(content) + return err +} + +func writeUpstreamResponse(w http.ResponseWriter, response *http.Response, body []byte) { + for name, values := range response.Header { + w.Header()[name] = append([]string(nil), values...) + } + w.WriteHeader(response.StatusCode) + _, _ = w.Write(body) +} + +func (g *gateway) logCallout(phase, requestID string, err error) { + if err != nil { + _, _ = fmt.Fprintf(g.stderr, "apim-gateway callout phase=%s request-id=%q outcome=error error=%q\n", phase, requestID, err.Error()) + return + } + _, _ = fmt.Fprintf(g.stderr, "apim-gateway callout phase=%s request-id=%q outcome=ok\n", phase, requestID) +} + +// The APIM policy uses ignore-error="true", but this shim deliberately fails closed on +// detectable callout failures so system-tests can exercise the mandatory D3 behavior. +func (g *gateway) failClosed(w http.ResponseWriter) { + http.Error(w, http.StatusText(http.StatusBadGateway), http.StatusBadGateway) +} diff --git a/utils/build/docker/golang/apim/main_test.go b/utils/build/docker/golang/apim/main_test.go new file mode 100644 index 00000000000..3d3d55d291f --- /dev/null +++ b/utils/build/docker/golang/apim/main_test.go @@ -0,0 +1,503 @@ +package main + +import ( + "bytes" + "encoding/base64" + "encoding/json" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" +) + +const fixtureRequestID = "request-123" + +type calloutStep struct { + phase string + requestID string + response string + assert func(*testing.T, map[string]json.RawMessage) +} + +func TestReplayFixtures(t *testing.T) { + t.Run("happy-four-call", testHappyFourCallFixture) + t.Run("block-at-request-headers", testRequestHeadersBlockFixture) + t.Run("block-at-request-body", testRequestBodyBlockFixture) +} + +func testHappyFourCallFixture(t *testing.T) { + callout := newScriptedCallout(t, []calloutStep{ + { + phase: "", + requestID: "", + response: `{"request-id":"request-123","propagate-headers":{"X-Test-Propagated":["yes"]},"allowed-body-size":3}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertGatewayOmitted(t, message) + assertRequestHeaderAddresses(t, message, nil) + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertBodyAddresses(t, message, "abc") + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{"allowed-body-size":8}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertResponseHeaderAddresses(t, message, nil) + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertBodyAddresses(t, message, "response") + }, + }, + }) + + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstreamCalls.Add(1) + if got := r.Header.Get("X-Test-Propagated"); got != "yes" { + t.Errorf("propagated header = %q, want yes", got) + } + if got := r.Header.Get(bodyModeHeader); got != "" { + t.Errorf("control header reached upstream as %q", got) + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read upstream request body: %v", err) + } + if got := string(body); got != "abcdef" { + t.Errorf("upstream request body = %q, want %q", got, "abcdef") + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-Upstream", "yes") + w.WriteHeader(http.StatusCreated) + _, _ = io.WriteString(w, "response-body") + })) + t.Cleanup(upstream.Close) + + var logs bytes.Buffer + gateway := newTestGateway(callout.URL, upstream.URL, &logs) + request := httptest.NewRequest(http.MethodPost, "http://example.test:7777/resource?x=1", strings.NewReader("abcdef")) + request.RemoteAddr = "198.51.100.4:5678" + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-Forwarded-For", "203.0.113.7") + request.Header.Set(bodyModeHeader, "deferred") + + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, request) + + if got := recorder.Code; got != http.StatusCreated { + t.Fatalf("status = %d, want %d", got, http.StatusCreated) + } + if got := recorder.Header().Get("X-Upstream"); got != "yes" { + t.Errorf("upstream response header = %q, want yes", got) + } + if got := recorder.Body.String(); got != "response-body" { + t.Errorf("response body = %q, want %q", got, "response-body") + } + if got := upstreamCalls.Load(); got != 1 { + t.Errorf("upstream calls = %d, want 1", got) + } + assertLogLines(t, logs.String(), []string{ + `apim-gateway callout phase= request-id="request-123" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" outcome=ok`, + }) +} + +func testRequestHeadersBlockFixture(t *testing.T) { + callout := newScriptedCallout(t, []calloutStep{{ + phase: "", + requestID: "", + response: `{"block":{"status":403,"headers":{"X-Block":["headers"]},"content":"YmxvY2tlZC1hdC1oZWFkZXJz"}}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertGatewayOmitted(t, message) + assertRequestHeaderAddresses(t, message, nil) + }, + }}) + + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + upstreamCalls.Add(1) + })) + t.Cleanup(upstream.Close) + + gateway := newTestGateway(callout.URL, upstream.URL, io.Discard) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, newFixtureRequest(http.MethodPost, nil)) + + assertBlockResponse(t, recorder, http.StatusForbidden, "headers", "blocked-at-headers") + if got := upstreamCalls.Load(); got != 0 { + t.Errorf("upstream calls = %d, want 0", got) + } +} + +func testRequestBodyBlockFixture(t *testing.T) { + callout := newScriptedCallout(t, []calloutStep{ + { + phase: "", + requestID: "", + response: `{"request-id":"request-123","allowed-body-size":1024}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertRequestHeaderAddresses(t, message, nil) + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{"block":{"status":406,"headers":{"X-Block":["body"]},"content":"YmxvY2tlZC1hdC1ib2R5"}}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertBodyAddresses(t, message, "payload") + }, + }, + }) + + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + upstreamCalls.Add(1) + })) + t.Cleanup(upstream.Close) + + gateway := newTestGateway(callout.URL, upstream.URL, io.Discard) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, newFixtureRequest(http.MethodPost, strings.NewReader("payload"))) + + assertBlockResponse(t, recorder, http.StatusNotAcceptable, "body", "blocked-at-body") + if got := upstreamCalls.Load(); got != 0 { + t.Errorf("upstream calls = %d, want 0", got) + } +} + +func TestResponsePhaseBlocksDiscardUpstreamResponse(t *testing.T) { + for _, fixture := range []struct { + name string + steps []calloutStep + }{ + { + name: "response-headers", + steps: []calloutStep{ + {phase: "", response: `{"request-id":"request-123"}`}, + {phase: "", requestID: fixtureRequestID, response: `{"block":{"status":451,"headers":{"X-Block":["response-headers"]},"content":"YmxvY2tlZA=="}}`}, + }, + }, + { + name: "response-body", + steps: []calloutStep{ + {phase: "", response: `{"request-id":"request-123"}`}, + {phase: "", requestID: fixtureRequestID, response: `{"allowed-body-size":1024}`}, + {phase: "", requestID: fixtureRequestID, response: `{"block":{"status":452,"headers":{"X-Block":["response-body"]},"content":"YmxvY2tlZA=="}}`}, + }, + }, + } { + t.Run(fixture.name, func(t *testing.T) { + callout := newScriptedCallout(t, fixture.steps) + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamCalls.Add(1) + w.WriteHeader(http.StatusCreated) + _, _ = io.WriteString(w, "upstream-response") + })) + t.Cleanup(upstream.Close) + + gateway := newTestGateway(callout.URL, upstream.URL, io.Discard) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "http://example.test:7777/", nil)) + + assertBlockResponse(t, recorder, fixture.steps[len(fixture.steps)-1].responseStatus(), fixture.name, "blocked") + if got := upstreamCalls.Load(); got != 1 { + t.Errorf("upstream calls = %d, want 1", got) + } + }) + } +} + +func TestInlineBodyModeUsesTwoCallouts(t *testing.T) { + requestBody := "request-body" + responseBody := "response-body" + callout := newScriptedCallout(t, []calloutStep{ + { + phase: "", + requestID: "", + response: `{"request-id":"request-123"}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertRequestHeaderAddresses(t, message, &requestBody) + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertResponseHeaderAddresses(t, message, &responseBody) + }, + }, + }) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get(bodyModeHeader); got != "" { + t.Errorf("control header reached upstream as %q", got) + } + w.Header().Set("Content-Type", "text/plain") + _, _ = io.WriteString(w, responseBody) + })) + t.Cleanup(upstream.Close) + + var logs bytes.Buffer + gateway := newTestGateway(callout.URL, upstream.URL, &logs) + request := newFixtureRequest(http.MethodPost, strings.NewReader(requestBody)) + request.Header.Set(bodyModeHeader, "inline") + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, request) + + if got := recorder.Code; got != http.StatusOK { + t.Fatalf("status = %d, want %d", got, http.StatusOK) + } + assertLogLines(t, logs.String(), []string{ + `apim-gateway callout phase= request-id="request-123" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" outcome=ok`, + }) +} + +func TestFailClosedWhenCalloutPortIsClosed(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + closedURL := "http://" + listener.Addr().String() + if err := listener.Close(); err != nil { + t.Fatal(err) + } + + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + upstreamCalls.Add(1) + })) + t.Cleanup(upstream.Close) + + var logs bytes.Buffer + gateway := newTestGateway(closedURL, upstream.URL, &logs) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "http://example.test:7777/", nil)) + + if got := recorder.Code; got != http.StatusBadGateway { + t.Fatalf("status = %d, want %d", got, http.StatusBadGateway) + } + if got := upstreamCalls.Load(); got != 0 { + t.Errorf("upstream calls = %d, want 0", got) + } + if got := logs.String(); !strings.Contains(got, `apim-gateway callout phase= request-id="" outcome=error`) { + t.Errorf("failure log = %q, want phase, empty request-id, and error outcome", got) + } +} + +func newTestGateway(calloutURL, upstreamURL string, stderr io.Writer) *gateway { + gateway := newGateway() + gateway.calloutURL = calloutURL + gateway.upstreamURL = upstreamURL + gateway.stderr = stderr + return gateway +} + +func newFixtureRequest(method string, body io.Reader) *http.Request { + request := httptest.NewRequest(method, "http://example.test:7777/resource?x=1", body) + request.RemoteAddr = "198.51.100.4:5678" + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-Forwarded-For", "203.0.113.7") + return request +} + +func newScriptedCallout(t *testing.T, steps []calloutStep) *httptest.Server { + t.Helper() + var next int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("callout method = %s, want POST", r.Method) + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + if next >= len(steps) { + t.Errorf("unexpected extra callout request") + w.WriteHeader(http.StatusInternalServerError) + return + } + step := steps[next] + next++ + + var message map[string]json.RawMessage + if err := json.NewDecoder(r.Body).Decode(&message); err != nil { + t.Errorf("decode callout request: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + assertRawString(t, message, "phase", step.phase) + if step.requestID == "" { + if _, ok := message["request-id"]; ok { + t.Errorf("phase %s unexpectedly sent request-id", step.phase) + } + } else { + assertRawString(t, message, "request-id", step.requestID) + } + if step.assert != nil { + step.assert(t, message) + } + + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, step.response) + })) + t.Cleanup(func() { + server.Close() + if next != len(steps) { + t.Errorf("callout calls = %d, want %d", next, len(steps)) + } + }) + return server +} + +func (step calloutStep) responseStatus() int { + var response struct { + Block struct { + Status int `json:"status"` + } `json:"block"` + } + if err := json.Unmarshal([]byte(step.response), &response); err != nil { + panic(err) + } + return response.Block.Status +} + +func assertGatewayOmitted(t *testing.T, message map[string]json.RawMessage) { + t.Helper() + if _, ok := message["gateway"]; ok { + t.Error("callout request included gateway") + } +} + +func assertRequestHeaderAddresses(t *testing.T, message map[string]json.RawMessage, body *string) { + t.Helper() + addresses := decodeAddresses(t, message) + assertRawString(t, addresses, "method", http.MethodPost) + assertRawString(t, addresses, "scheme", "http") + assertRawString(t, addresses, "authority", "example.test:7777") + assertRawString(t, addresses, "path", "/resource?x=1") + assertRawString(t, addresses, "remote_addr", "198.51.100.4:5678") + + var headers map[string][]string + if err := json.Unmarshal(addresses["headers"], &headers); err != nil { + t.Fatalf("decode request headers: %v", err) + } + if got := headers[bodyModeHeader]; len(got) != 0 { + t.Errorf("control header included in callout addresses: %v", got) + } + if got := headers["X-Forwarded-For"]; len(got) != 1 || got[0] != "203.0.113.7" { + t.Errorf("forwarding header = %v, want [203.0.113.7]", got) + } + assertAddressBody(t, addresses, body) +} + +func assertResponseHeaderAddresses(t *testing.T, message map[string]json.RawMessage, body *string) { + t.Helper() + addresses := decodeAddresses(t, message) + var status int + if err := json.Unmarshal(addresses["status_code"], &status); err != nil { + t.Fatalf("decode response status: %v", err) + } + if status != http.StatusCreated && status != http.StatusOK { + t.Errorf("response status = %d, want 201 or 200", status) + } + assertAddressBody(t, addresses, body) +} + +func assertBodyAddresses(t *testing.T, message map[string]json.RawMessage, want string) { + t.Helper() + assertAddressBody(t, decodeAddresses(t, message), &want) +} + +func assertAddressBody(t *testing.T, addresses map[string]json.RawMessage, want *string) { + t.Helper() + raw, ok := addresses["body"] + if want == nil { + if ok { + t.Errorf("body = %s, want omitted", raw) + } + return + } + if !ok { + t.Fatal("body missing") + } + var encoded string + if err := json.Unmarshal(raw, &encoded); err != nil { + t.Fatalf("decode body: %v", err) + } + decoded, err := base64.StdEncoding.DecodeString(encoded) + if err != nil { + t.Fatalf("decode body base64: %v", err) + } + if got := string(decoded); got != *want { + t.Errorf("body = %q, want %q", got, *want) + } +} + +func decodeAddresses(t *testing.T, message map[string]json.RawMessage) map[string]json.RawMessage { + t.Helper() + var addresses map[string]json.RawMessage + if err := json.Unmarshal(message["addresses"], &addresses); err != nil { + t.Fatalf("decode addresses: %v", err) + } + return addresses +} + +func assertRawString(t *testing.T, values map[string]json.RawMessage, key, want string) { + t.Helper() + raw, ok := values[key] + if !ok { + t.Errorf("missing %q", key) + return + } + var got string + if err := json.Unmarshal(raw, &got); err != nil { + t.Errorf("decode %q: %v", key, err) + return + } + if got != want { + t.Errorf("%s = %q, want %q", key, got, want) + } +} + +func assertBlockResponse(t *testing.T, recorder *httptest.ResponseRecorder, wantStatus int, wantHeader, wantBody string) { + t.Helper() + if got := recorder.Code; got != wantStatus { + t.Errorf("status = %d, want %d", got, wantStatus) + } + if got := recorder.Header().Get("X-Block"); got != wantHeader { + t.Errorf("X-Block = %q, want %q", got, wantHeader) + } + if got := recorder.Body.String(); got != wantBody { + t.Errorf("body = %q, want %q", got, wantBody) + } +} + +func assertLogLines(t *testing.T, logs string, want []string) { + t.Helper() + got := strings.FieldsFunc(strings.TrimSpace(logs), func(r rune) bool { return r == '\n' }) + if len(got) != len(want) { + t.Fatalf("log lines = %q, want %q", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("log line %d = %q, want %q", i, got[i], want[i]) + } + } +} diff --git a/utils/build/docker/golang/go.work b/utils/build/docker/golang/go.work index bcade5e527e..39618382d24 100644 --- a/utils/build/docker/golang/go.work +++ b/utils/build/docker/golang/go.work @@ -1,6 +1,7 @@ go 1.25.0 use ( + ./apim ./app ./app/net-http-orchestrion ./parametric diff --git a/utils/build/docker/golang/weblog_metadata.yml b/utils/build/docker/golang/weblog_metadata.yml index 4a1f6ae278b..51f16fe6e1d 100644 --- a/utils/build/docker/golang/weblog_metadata.yml +++ b/utils/build/docker/golang/weblog_metadata.yml @@ -6,6 +6,9 @@ haproxy: envoy: build_mode: none supported_scenarios: [APPSEC_BLOCKING, DEFAULT] +apim: + build_mode: none + supported_scenarios: [APPSEC_BLOCKING, DEFAULT] graphql-go: categories: [dd_trace_graphql] excluded_scenarios: [GRAPHQL_ERROR_TRACKING] # why ??? diff --git a/utils/scripts/load-binary.sh b/utils/scripts/load-binary.sh index f063797e299..378da8b92ea 100755 --- a/utils/scripts/load-binary.sh +++ b/utils/scripts/load-binary.sh @@ -308,6 +308,10 @@ elif [ "$TARGET" = "golang" ]; then echo "Using ghcr.io/datadog/dd-trace-go/haproxy-spoa:dev" echo "ghcr.io/datadog/dd-trace-go/haproxy-spoa:dev" > golang-haproxy-spoa-image + # azure apim integration + echo "Using ghcr.io/datadog/dd-trace-go/apim-callout:dev" + echo "ghcr.io/datadog/dd-trace-go/apim-callout:dev" > golang-apim-callout-image + elif [ "$TARGET" = "cpp" ]; then assert_version_is_dev # PROFILER: The main version is stored in s3, though we can not access this in CI From 390e4c3ec65bf6f8f311ff72407e77f0774d2971 Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Thu, 13 Aug 2026 11:08:32 +0200 Subject: [PATCH 2/9] [golang] Add apim inline-mode tests, correlate callout logs by path Cover the apim shim's two body modes with system-tests: - default (deferred) drives 4 callout calls sharing one request-id - inline drives exactly 2 - span structure and component are identical in both modes - inline mode closes cached state, so no orphaned-span warning appears in apim-callout stderr after the 30s TTL The gateway now logs the request path alongside the phase and request-id. Without it these tests cannot isolate their own traffic: the gateway healthcheck is a bodiless GET, which is deferred mode, so it drives a 4-call flow of its own and the container only reports healthy once such a request succeeds. Asserting merely that "some request-id somewhere made 4 calls" therefore could not fail. Each test now uses a unique probe path and asserts exactly one correlated request-id group for it. The path is logged to stderr rather than returned as a response header so it stays invisible to the WAF, the upstream and the client. --- manifests/golang.yml | 16 +++ .../external_processing/test_apim_callout.py | 107 ++++++++++++++++++ utils/build/docker/golang/apim/main.go | 35 +++--- utils/build/docker/golang/apim/main_test.go | 16 +-- 4 files changed, 149 insertions(+), 25 deletions(-) create mode 100644 tests/external_processing/test_apim_callout.py diff --git a/manifests/golang.yml b/manifests/golang.yml index 3af448dd57a..92344073e59 100644 --- a/manifests/golang.yml +++ b/manifests/golang.yml @@ -1155,6 +1155,22 @@ manifest: tests/debugger/test_debugger_symdb.py::Test_Debugger_SymDb::test_event_metadata: missing_feature (extended event schema not yet shipped) tests/debugger/test_debugger_telemetry.py::Test_Debugger_Telemetry: missing_feature tests/docker_ssi/test_docker_ssi_appsec.py::TestDockerSSIAppsecFeatures::test_telemetry_source_ssi: v2.0.0 + tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_default_body_mode_uses_four_callouts: + - weblog_declaration: + "*": irrelevant (only applies to apim) + apim: v2.7.1 + tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_inline_body_mode_closes_request_state: + - weblog_declaration: + "*": irrelevant (only applies to apim) + apim: v2.7.1 + tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_inline_body_mode_preserves_trace_structure: + - weblog_declaration: + "*": irrelevant (only applies to apim) + apim: v2.7.1 + tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_inline_body_mode_uses_two_callouts: + - weblog_declaration: + "*": irrelevant (only applies to apim) + apim: v2.7.1 tests/ffe/test_agentless_configuration.py: missing_feature (FFL-2695 tracks Go agentless configuration-source implementation; FFL-2731 tracks the system-tests contract) tests/ffe/test_dynamic_evaluation.py::Test_FFE_Flag_Parse_Error_Isolation::test_valid_flag_unaffected: bug (FFL-2184) tests/ffe/test_dynamic_evaluation.py::Test_FFE_RC_Down_From_Start: v2.4.0 diff --git a/tests/external_processing/test_apim_callout.py b/tests/external_processing/test_apim_callout.py new file mode 100644 index 00000000000..97adc6d20f9 --- /dev/null +++ b/tests/external_processing/test_apim_callout.py @@ -0,0 +1,107 @@ +import re +import time +from collections import defaultdict +from pathlib import Path + +from utils import context, features, interfaces, irrelevant, weblog +from utils._weblog import HttpResponse +from utils.dd_types import DataDogLibrarySpan + + +CALLOUT_LOG_PATTERN = re.compile( + r'apim-gateway callout phase=(?P<[^>]+>) request-id="(?P[^"]+)" path="(?P[^"]+)" outcome=ok' +) +DEFERRED_PHASES = ("", "", "", "") +INLINE_PHASES = ("", "") +APIM_SPAN = ("web", "server", "apim-callout") +DEFAULT_PROBE_PATH = "/?probe=default-four-call" +INLINE_PROBE_PATH = "/?probe=inline-two-call" +TRACE_DEFAULT_PROBE_PATH = "/?probe=trace-default" +TRACE_INLINE_PROBE_PATH = "/?probe=trace-inline" +STATE_CLOSURE_PROBE_PATH = "/?probe=inline-state-closure" + + +def _container_stderr(container_name: str) -> str: + log_path = Path(context.scenario.host_log_folder) / "docker" / container_name / "stderr.log" + return log_path.read_text(encoding="utf-8") + + +def _callout_phases_by_request_id_and_path() -> dict[tuple[str, str], list[str]]: + phases_by_request_id_and_path: defaultdict[tuple[str, str], list[str]] = defaultdict(list) + for line in _container_stderr("apim-gateway").splitlines(): + if match := CALLOUT_LOG_PATTERN.fullmatch(line): + key = match.group("request_id"), match.group("path") + phases_by_request_id_and_path[key].append(match.group("phase")) + + return dict(phases_by_request_id_and_path) + + +def _assert_probe_phases(probe_path: str, expected_phases: tuple[str, ...]) -> None: + matching_groups = { + request_id: phases + for (request_id, path), phases in _callout_phases_by_request_id_and_path().items() + if path == probe_path + } + assert len(matching_groups) == 1 + assert next(iter(matching_groups.values())) == list(expected_phases) + + +def _span_structure(span: DataDogLibrarySpan) -> tuple[str, str, str]: + return span["type"], span["meta"]["span.kind"], span["meta"]["component"] + + +def _assert_apim_span(span: DataDogLibrarySpan) -> None: + assert _span_structure(span) == APIM_SPAN + + +def _trace_structure(request: HttpResponse) -> list[tuple[str, str, str]]: + interfaces.library.assert_trace_exists(request=request) + assert _span_structure(interfaces.library.get_root_span(request=request)) == APIM_SPAN + interfaces.library.validate_all_spans(request=request, validator=_assert_apim_span) + return sorted(_span_structure(span) for _, _, span in interfaces.library.get_spans(request=request)) + + +@irrelevant(context.weblog_variant != "apim") +@features.go_proxies +class Test_ApimCallout: + def setup_default_body_mode_uses_four_callouts(self): + self.r = weblog.post(DEFAULT_PROBE_PATH, json={"body": "default"}) + + def test_default_body_mode_uses_four_callouts(self): + assert self.r.status_code == 200 + _assert_probe_phases(DEFAULT_PROBE_PATH, DEFERRED_PHASES) + + def setup_inline_body_mode_uses_two_callouts(self): + self.r = weblog.post(INLINE_PROBE_PATH, json={"body": "inline"}, headers={"X-Datadog-Apim-Body-Mode": "inline"}) + + def test_inline_body_mode_uses_two_callouts(self): + assert self.r.status_code == 200 + _assert_probe_phases(INLINE_PROBE_PATH, INLINE_PHASES) + + def setup_inline_body_mode_preserves_trace_structure(self): + self.default_response = weblog.post(TRACE_DEFAULT_PROBE_PATH, json={"body": "default trace"}) + self.inline_response = weblog.post( + TRACE_INLINE_PROBE_PATH, + json={"body": "inline trace"}, + headers={"X-Datadog-Apim-Body-Mode": "inline"}, + ) + + def test_inline_body_mode_preserves_trace_structure(self): + assert self.default_response.status_code == 200 + assert self.inline_response.status_code == 200 + _assert_probe_phases(TRACE_DEFAULT_PROBE_PATH, DEFERRED_PHASES) + _assert_probe_phases(TRACE_INLINE_PROBE_PATH, INLINE_PHASES) + assert _trace_structure(self.default_response) == _trace_structure(self.inline_response) + + def setup_inline_body_mode_closes_request_state(self): + self.r = weblog.post( + STATE_CLOSURE_PROBE_PATH, + json={"body": "inline state"}, + headers={"X-Datadog-Apim-Body-Mode": "inline"}, + ) + time.sleep(31) + + def test_inline_body_mode_closes_request_state(self): + assert self.r.status_code == 200 + _assert_probe_phases(STATE_CLOSURE_PROBE_PATH, INLINE_PHASES) + assert "closing orphaned span" not in _container_stderr("apim-callout") diff --git a/utils/build/docker/golang/apim/main.go b/utils/build/docker/golang/apim/main.go index 148a99db306..d7a995c132a 100644 --- a/utils/build/docker/golang/apim/main.go +++ b/utils/build/docker/golang/apim/main.go @@ -104,6 +104,7 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { } defer r.Body.Close() + requestPath := r.URL.RequestURI() inline := r.Header.Get(bodyModeHeader) == "inline" r.Header.Del(bodyModeHeader) @@ -111,7 +112,7 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { Method: r.Method, Scheme: requestScheme(r), Authority: r.Host, - Path: r.URL.RequestURI(), + Path: requestPath, RemoteAddr: r.RemoteAddr, Headers: map[string][]string(r.Header.Clone()), } @@ -123,7 +124,7 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } - phase1, err := g.callout(phaseRequestHeaders, "", requestAddresses) + phase1, err := g.callout(phaseRequestHeaders, "", requestPath, requestAddresses) if err != nil { g.failClosed(w) return @@ -143,7 +144,7 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { g.failClosed(w) return } - phase2, err := g.callout(phaseRequestBody, requestID, addressesBody{Body: body}) + phase2, err := g.callout(phaseRequestBody, requestID, requestPath, addressesBody{Body: body}) if err != nil { g.failClosed(w) return @@ -181,7 +182,7 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } - phase3, err := g.callout(phaseResponseHeaders, requestID, responseAddresses) + phase3, err := g.callout(phaseResponseHeaders, requestID, requestPath, responseAddresses) if err != nil { g.failClosed(w) return @@ -198,7 +199,7 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { g.failClosed(w) return } - phase4, err := g.callout(phaseResponseBody, requestID, addressesBody{Body: body}) + phase4, err := g.callout(phaseResponseBody, requestID, requestPath, addressesBody{Body: body}) if err != nil { g.failClosed(w) return @@ -221,10 +222,10 @@ func requestScheme(r *http.Request) string { return "http" } -func (g *gateway) callout(phase, requestID string, addresses any) (calloutResult, error) { +func (g *gateway) callout(phase, requestID, requestPath string, addresses any) (calloutResult, error) { addressesJSON, err := json.Marshal(addresses) if err != nil { - g.logCallout(phase, requestID, err) + g.logCallout(phase, requestID, requestPath, err) return calloutResult{}, err } payload, err := json.Marshal(calloutMessage{ @@ -233,37 +234,37 @@ func (g *gateway) callout(phase, requestID string, addresses any) (calloutResult Phase: phase, }) if err != nil { - g.logCallout(phase, requestID, err) + g.logCallout(phase, requestID, requestPath, err) return calloutResult{}, err } request, err := http.NewRequest(http.MethodPost, g.calloutURL, bytes.NewReader(payload)) if err != nil { - g.logCallout(phase, requestID, err) + g.logCallout(phase, requestID, requestPath, err) return calloutResult{}, err } request.Header.Set("Content-Type", "application/json") response, err := g.calloutClient.Do(request) if err != nil { - g.logCallout(phase, requestID, err) + g.logCallout(phase, requestID, requestPath, err) return calloutResult{}, err } defer response.Body.Close() if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { err := fmt.Errorf("callout returned %s", response.Status) - g.logCallout(phase, requestID, err) + g.logCallout(phase, requestID, requestPath, err) return calloutResult{}, err } var result calloutResult if err := decodeJSON(response.Body, &result); err != nil { - g.logCallout(phase, requestID, err) + g.logCallout(phase, requestID, requestPath, err) return calloutResult{}, err } if phase == phaseRequestHeaders && result.Block == nil && result.RequestID == "" { err := fmt.Errorf("phase 1 callout response has no request-id") - g.logCallout(phase, result.RequestID, err) + g.logCallout(phase, result.RequestID, requestPath, err) return calloutResult{}, err } @@ -271,7 +272,7 @@ func (g *gateway) callout(phase, requestID string, addresses any) (calloutResult if phase == phaseRequestHeaders { logRequestID = result.RequestID } - g.logCallout(phase, logRequestID, nil) + g.logCallout(phase, logRequestID, requestPath, nil) return result, nil } @@ -342,12 +343,12 @@ func writeUpstreamResponse(w http.ResponseWriter, response *http.Response, body _, _ = w.Write(body) } -func (g *gateway) logCallout(phase, requestID string, err error) { +func (g *gateway) logCallout(phase, requestID, requestPath string, err error) { if err != nil { - _, _ = fmt.Fprintf(g.stderr, "apim-gateway callout phase=%s request-id=%q outcome=error error=%q\n", phase, requestID, err.Error()) + _, _ = fmt.Fprintf(g.stderr, "apim-gateway callout phase=%s request-id=%q path=%q outcome=error error=%q\n", phase, requestID, requestPath, err.Error()) return } - _, _ = fmt.Fprintf(g.stderr, "apim-gateway callout phase=%s request-id=%q outcome=ok\n", phase, requestID) + _, _ = fmt.Fprintf(g.stderr, "apim-gateway callout phase=%s request-id=%q path=%q outcome=ok\n", phase, requestID, requestPath) } // The APIM policy uses ignore-error="true", but this shim deliberately fails closed on diff --git a/utils/build/docker/golang/apim/main_test.go b/utils/build/docker/golang/apim/main_test.go index 3d3d55d291f..46bf0d9c17f 100644 --- a/utils/build/docker/golang/apim/main_test.go +++ b/utils/build/docker/golang/apim/main_test.go @@ -112,10 +112,10 @@ func testHappyFourCallFixture(t *testing.T) { t.Errorf("upstream calls = %d, want 1", got) } assertLogLines(t, logs.String(), []string{ - `apim-gateway callout phase= request-id="request-123" outcome=ok`, - `apim-gateway callout phase= request-id="request-123" outcome=ok`, - `apim-gateway callout phase= request-id="request-123" outcome=ok`, - `apim-gateway callout phase= request-id="request-123" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, }) } @@ -266,8 +266,8 @@ func TestInlineBodyModeUsesTwoCallouts(t *testing.T) { t.Fatalf("status = %d, want %d", got, http.StatusOK) } assertLogLines(t, logs.String(), []string{ - `apim-gateway callout phase= request-id="request-123" outcome=ok`, - `apim-gateway callout phase= request-id="request-123" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, }) } @@ -298,8 +298,8 @@ func TestFailClosedWhenCalloutPortIsClosed(t *testing.T) { if got := upstreamCalls.Load(); got != 0 { t.Errorf("upstream calls = %d, want 0", got) } - if got := logs.String(); !strings.Contains(got, `apim-gateway callout phase= request-id="" outcome=error`) { - t.Errorf("failure log = %q, want phase, empty request-id, and error outcome", got) + if got := logs.String(); !strings.Contains(got, `apim-gateway callout phase= request-id="" path="/" outcome=error`) { + t.Errorf("failure log = %q, want phase, empty request-id, path, and error outcome", got) } } From 3434ea875fe2c29141a34db9bb594770423793c7 Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Thu, 13 Aug 2026 17:08:33 +0200 Subject: [PATCH 3/9] [golang] Tighten apim-gateway healthcheck to the measured cold start AC1 measured ~11.93s from container start to the first successful health probe. Replace the provisional 153s ceiling with ~21s (15s start_period + 6 * 1s), keeping a cushion because `go run .` still compiles the shim from an empty build cache on every container start. --- utils/_context/containers.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/utils/_context/containers.py b/utils/_context/containers.py index f798513783e..263de478dca 100644 --- a/utils/_context/containers.py +++ b/utils/_context/containers.py @@ -1801,13 +1801,11 @@ def __init__(self) -> None: # golang:1.25-alpine ships busybox but no bash, so the /bin/bash + /dev/tcp # healthcheck used by EnvoyContainer and HAProxyContainer is not usable here "test": "wget -qO- http://localhost:80/", - # PROVISIONAL budget, to be tightened once a real cold start is measured: - # `go run .` compiles the shim from an empty build cache on every start, which is - # far slower than booting a prebuilt binary. interval and start_period are in - # NANOSECONDS (execute_command divides both by 1e9), so this is - # 15s + 46 * 3s ~= 153s ceiling, with a 3s readiness granularity. - "retries": 45, - "interval": 3_000_000_000, + # AC1 measured ~11.93s from container start to first successful health probe. + # `go run .` still compiles on every start, so keep a small cushion: ~21s + # ceiling (15s + 6 * 1s). + "retries": 5, + "interval": 1_000_000_000, "start_period": 15_000_000_000, }, ) From ee345502b7cb1d4f0ca414aa8059a88d64a86f3d Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Thu, 13 Aug 2026 17:43:37 +0200 Subject: [PATCH 4/9] [golang] Address final review: fail-closed diagnostics, span-links gap, cuts Fail-closed paths now emit a diagnostic. D3 requires "502 + stderr", but failClosed discarded the error, so 11 paths returned a bare 502 with nothing in the log: request/upstream body reads, inline and deferred body encodes, the upstream call, and block writes at all four phases. Each now logs `apim-gateway fail-closed stage= error=...`. Callout-error sites use a separate non-logging variant, since callout() already logged them and a single logging helper would double-log. Adds tests for the upstream-closed-port and malformed-block paths, and tightens the existing closed-port test to assert an exact line count rather than a substring, which previously tolerated the very double-logging this splits apart. Record the span-links gap for apim. apim-callout defaults DD_TRACE_PROPAGATION_STYLE to datadog, so a conflicting W3C traceparent is never extracted and no terminated-context span link is created. Envoy's processor leaves the variable unset and gets the tracer default, which is why it passes. haproxy-spoa ships the same datadog-only default and is already missing_feature on this row, so apim is marked beside it rather than having the harness override a shipped default to force a pass. Drop three pieces of dead or gold-plated code: - calloutMessage.Gateway was never assigned. With omitempty and no writer it could not serialize, so removal is wire-neutral, and it was the footgun for the rule against sending a gateway at all. - decodeJSON's trailing-JSON rejection was untested and guarded nothing: the first Decode had already produced the correct result. It only manufactured a 502 the real callout cannot trigger, adding a third candidate cause to a shim whose main risk is undiagnosable failures. The two sibling guards stay, because each converts a handler panic into a clean 502. - responseStatus() reflected over fixture JSON and panicked to recover values already present as literals; replaced by a wantStatus field. Also drops the four manifest entries for the new tests. The @irrelevant decorator already produces the "*" outcome, and the sibling test_apm.py has no manifest rows at all, so the entries duplicated activation across two mechanisms and the version floor was unreachable. --- manifests/golang.yml | 17 +- .../external_processing/test_apim_callout.py | 50 ++++-- utils/build/docker/golang/apim/main.go | 60 ++++--- utils/build/docker/golang/apim/main_test.go | 150 ++++++++++++++---- 4 files changed, 181 insertions(+), 96 deletions(-) diff --git a/manifests/golang.yml b/manifests/golang.yml index 92344073e59..365a4fd76de 100644 --- a/manifests/golang.yml +++ b/manifests/golang.yml @@ -1155,22 +1155,6 @@ manifest: tests/debugger/test_debugger_symdb.py::Test_Debugger_SymDb::test_event_metadata: missing_feature (extended event schema not yet shipped) tests/debugger/test_debugger_telemetry.py::Test_Debugger_Telemetry: missing_feature tests/docker_ssi/test_docker_ssi_appsec.py::TestDockerSSIAppsecFeatures::test_telemetry_source_ssi: v2.0.0 - tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_default_body_mode_uses_four_callouts: - - weblog_declaration: - "*": irrelevant (only applies to apim) - apim: v2.7.1 - tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_inline_body_mode_closes_request_state: - - weblog_declaration: - "*": irrelevant (only applies to apim) - apim: v2.7.1 - tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_inline_body_mode_preserves_trace_structure: - - weblog_declaration: - "*": irrelevant (only applies to apim) - apim: v2.7.1 - tests/external_processing/test_apim_callout.py::Test_ApimCallout::test_inline_body_mode_uses_two_callouts: - - weblog_declaration: - "*": irrelevant (only applies to apim) - apim: v2.7.1 tests/ffe/test_agentless_configuration.py: missing_feature (FFL-2695 tracks Go agentless configuration-source implementation; FFL-2731 tracks the system-tests contract) tests/ffe/test_dynamic_evaluation.py::Test_FFE_Flag_Parse_Error_Isolation::test_valid_flag_unaffected: bug (FFL-2184) tests/ffe/test_dynamic_evaluation.py::Test_FFE_RC_Down_From_Start: v2.4.0 @@ -1775,6 +1759,7 @@ manifest: tests/test_distributed.py::Test_Span_Links_From_Conflicting_Contexts: v2.2.3 tests/test_distributed.py::Test_Span_Links_From_Conflicting_Contexts::test_span_links_from_conflicting_contexts: - weblog_declaration: + apim: missing_feature (apim-callout defaults DD_TRACE_PROPAGATION_STYLE to datadog, so the conflicting W3C context is not extracted) haproxy: missing_feature tests/test_distributed.py::Test_Span_Links_Omit_Tracestate_From_Conflicting_Contexts: missing_feature (implementation specs have not been determined) tests/test_distributed.py::Test_Synthetics_APM_Datadog: # Modified by easy win activation script diff --git a/tests/external_processing/test_apim_callout.py b/tests/external_processing/test_apim_callout.py index 97adc6d20f9..ea03167eb72 100644 --- a/tests/external_processing/test_apim_callout.py +++ b/tests/external_processing/test_apim_callout.py @@ -42,8 +42,14 @@ def _assert_probe_phases(probe_path: str, expected_phases: tuple[str, ...]) -> N for (request_id, path), phases in _callout_phases_by_request_id_and_path().items() if path == probe_path } - assert len(matching_groups) == 1 - assert next(iter(matching_groups.values())) == list(expected_phases) + assert len(matching_groups) == 1, ( + f"expected exactly one callout request-id group for probe {probe_path}, " + f"got {len(matching_groups)}: {sorted(matching_groups)}" + ) + request_id, phases = next(iter(matching_groups.items())) + assert phases == list(expected_phases), ( + f"probe {probe_path} (request-id {request_id}) hit callout phases {phases}, expected {list(expected_phases)}" + ) def _span_structure(span: DataDogLibrarySpan) -> tuple[str, str, str]: @@ -64,21 +70,21 @@ def _trace_structure(request: HttpResponse) -> list[tuple[str, str, str]]: @irrelevant(context.weblog_variant != "apim") @features.go_proxies class Test_ApimCallout: - def setup_default_body_mode_uses_four_callouts(self): + def setup_default_body_mode_uses_four_callouts(self) -> None: self.r = weblog.post(DEFAULT_PROBE_PATH, json={"body": "default"}) - def test_default_body_mode_uses_four_callouts(self): - assert self.r.status_code == 200 + def test_default_body_mode_uses_four_callouts(self) -> None: + assert self.r.status_code == 200, f"deferred probe returned {self.r.status_code}, expected 200" _assert_probe_phases(DEFAULT_PROBE_PATH, DEFERRED_PHASES) - def setup_inline_body_mode_uses_two_callouts(self): + def setup_inline_body_mode_uses_two_callouts(self) -> None: self.r = weblog.post(INLINE_PROBE_PATH, json={"body": "inline"}, headers={"X-Datadog-Apim-Body-Mode": "inline"}) - def test_inline_body_mode_uses_two_callouts(self): - assert self.r.status_code == 200 + def test_inline_body_mode_uses_two_callouts(self) -> None: + assert self.r.status_code == 200, f"inline probe returned {self.r.status_code}, expected 200" _assert_probe_phases(INLINE_PROBE_PATH, INLINE_PHASES) - def setup_inline_body_mode_preserves_trace_structure(self): + def setup_inline_body_mode_preserves_trace_structure(self) -> None: self.default_response = weblog.post(TRACE_DEFAULT_PROBE_PATH, json={"body": "default trace"}) self.inline_response = weblog.post( TRACE_INLINE_PROBE_PATH, @@ -86,14 +92,22 @@ def setup_inline_body_mode_preserves_trace_structure(self): headers={"X-Datadog-Apim-Body-Mode": "inline"}, ) - def test_inline_body_mode_preserves_trace_structure(self): - assert self.default_response.status_code == 200 - assert self.inline_response.status_code == 200 + def test_inline_body_mode_preserves_trace_structure(self) -> None: + assert self.default_response.status_code == 200, ( + f"deferred probe returned {self.default_response.status_code}, expected 200" + ) + assert self.inline_response.status_code == 200, ( + f"inline probe returned {self.inline_response.status_code}, expected 200" + ) _assert_probe_phases(TRACE_DEFAULT_PROBE_PATH, DEFERRED_PHASES) _assert_probe_phases(TRACE_INLINE_PROBE_PATH, INLINE_PHASES) - assert _trace_structure(self.default_response) == _trace_structure(self.inline_response) + default_spans = _trace_structure(self.default_response) + inline_spans = _trace_structure(self.inline_response) + assert default_spans == inline_spans, ( + f"inline body delivery changed the trace: deferred spans {default_spans}, inline spans {inline_spans}" + ) - def setup_inline_body_mode_closes_request_state(self): + def setup_inline_body_mode_closes_request_state(self) -> None: self.r = weblog.post( STATE_CLOSURE_PROBE_PATH, json={"body": "inline state"}, @@ -101,7 +115,9 @@ def setup_inline_body_mode_closes_request_state(self): ) time.sleep(31) - def test_inline_body_mode_closes_request_state(self): - assert self.r.status_code == 200 + def test_inline_body_mode_closes_request_state(self) -> None: + assert self.r.status_code == 200, f"inline probe returned {self.r.status_code}, expected 200" _assert_probe_phases(STATE_CLOSURE_PROBE_PATH, INLINE_PHASES) - assert "closing orphaned span" not in _container_stderr("apim-callout") + assert "closing orphaned span" not in _container_stderr("apim-callout"), ( + "apim-callout logged an orphaned span, so inline mode did not close the cached request state" + ) diff --git a/utils/build/docker/golang/apim/main.go b/utils/build/docker/golang/apim/main.go index d7a995c132a..e953f5922b4 100644 --- a/utils/build/docker/golang/apim/main.go +++ b/utils/build/docker/golang/apim/main.go @@ -29,7 +29,6 @@ const ( // The Addresses field is phase-dependent and decoded separately. type calloutMessage struct { Addresses json.RawMessage `json:"addresses"` - Gateway string `json:"gateway,omitempty"` RequestID string `json:"request-id,omitempty"` Phase string `json:"phase,omitempty"` } @@ -83,6 +82,7 @@ type gateway struct { func main() { if err := http.ListenAndServe(":80", newGateway()); err != nil { fmt.Fprintln(os.Stderr, err) + os.Exit(1) } } @@ -99,7 +99,7 @@ func newGateway() *gateway { func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { requestBody, err := io.ReadAll(r.Body) if err != nil { - g.failClosed(w) + g.failClosed(w, "read-request-body", err) return } defer r.Body.Close() @@ -119,19 +119,19 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { if inline { requestAddresses.Body, err = encodedBody(requestBody, nil) if err != nil { - g.failClosed(w) + g.failClosed(w, "encode-inline-request-body", err) return } } phase1, err := g.callout(phaseRequestHeaders, "", requestPath, requestAddresses) if err != nil { - g.failClosed(w) + g.writeFailClosed(w) return } if phase1.Block != nil { if err := writeBlock(w, phase1.Block); err != nil { - g.failClosed(w) + g.failClosed(w, "write-block-request-headers", err) } return } @@ -141,17 +141,17 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { if phase1.AllowedBodySize != nil { body, err := encodedBody(requestBody, phase1.AllowedBodySize) if err != nil { - g.failClosed(w) + g.failClosed(w, "encode-request-body", err) return } phase2, err := g.callout(phaseRequestBody, requestID, requestPath, addressesBody{Body: body}) if err != nil { - g.failClosed(w) + g.writeFailClosed(w) return } if phase2.Block != nil { if err := writeBlock(w, phase2.Block); err != nil { - g.failClosed(w) + g.failClosed(w, "write-block-request-body", err) } return } @@ -159,14 +159,14 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { upstreamResponse, err := g.callUpstream(r, requestBody) if err != nil { - g.failClosed(w) + g.failClosed(w, "call-upstream", err) return } defer upstreamResponse.Body.Close() responseBody, err := io.ReadAll(upstreamResponse.Body) if err != nil { - g.failClosed(w) + g.failClosed(w, "read-upstream-body", err) return } @@ -177,36 +177,36 @@ func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { if inline { responseAddresses.Body, err = encodedBody(responseBody, nil) if err != nil { - g.failClosed(w) + g.failClosed(w, "encode-inline-response-body", err) return } } phase3, err := g.callout(phaseResponseHeaders, requestID, requestPath, responseAddresses) if err != nil { - g.failClosed(w) + g.writeFailClosed(w) return } if phase3.Block != nil { if err := writeBlock(w, phase3.Block); err != nil { - g.failClosed(w) + g.failClosed(w, "write-block-response-headers", err) } return } if phase3.AllowedBodySize != nil { body, err := encodedBody(responseBody, phase3.AllowedBodySize) if err != nil { - g.failClosed(w) + g.failClosed(w, "encode-response-body", err) return } phase4, err := g.callout(phaseResponseBody, requestID, requestPath, addressesBody{Body: body}) if err != nil { - g.failClosed(w) + g.writeFailClosed(w) return } if phase4.Block != nil { if err := writeBlock(w, phase4.Block); err != nil { - g.failClosed(w) + g.failClosed(w, "write-block-response-body", err) } return } @@ -258,7 +258,7 @@ func (g *gateway) callout(phase, requestID, requestPath string, addresses any) ( } var result calloutResult - if err := decodeJSON(response.Body, &result); err != nil { + if err := json.NewDecoder(response.Body).Decode(&result); err != nil { g.logCallout(phase, requestID, requestPath, err) return calloutResult{}, err } @@ -276,21 +276,6 @@ func (g *gateway) callout(phase, requestID, requestPath string, addresses any) ( return result, nil } -func decodeJSON(body io.Reader, result *calloutResult) error { - decoder := json.NewDecoder(body) - if err := decoder.Decode(result); err != nil { - return err - } - var trailing any - if err := decoder.Decode(&trailing); err != io.EOF { - if err == nil { - return fmt.Errorf("callout response contains multiple JSON values") - } - return err - } - return nil -} - func (g *gateway) callUpstream(r *http.Request, body []byte) (*http.Response, error) { request, err := http.NewRequestWithContext(r.Context(), r.Method, g.upstreamURL+r.URL.RequestURI(), bytes.NewReader(body)) if err != nil { @@ -353,6 +338,15 @@ func (g *gateway) logCallout(phase, requestID, requestPath string, err error) { // The APIM policy uses ignore-error="true", but this shim deliberately fails closed on // detectable callout failures so system-tests can exercise the mandatory D3 behavior. -func (g *gateway) failClosed(w http.ResponseWriter) { +// failClosed reports the causing error on stderr, tagged with the stage that failed, then +// returns the fail-closed status. +func (g *gateway) failClosed(w http.ResponseWriter, stage string, err error) { + _, _ = fmt.Fprintf(g.stderr, "apim-gateway fail-closed stage=%s error=%q\n", stage, err.Error()) + g.writeFailClosed(w) +} + +// writeFailClosed returns the fail-closed status for a failure that logCallout has already +// reported, so a callout failure is diagnosed exactly once. +func (g *gateway) writeFailClosed(w http.ResponseWriter) { http.Error(w, http.StatusText(http.StatusBadGateway), http.StatusBadGateway) } diff --git a/utils/build/docker/golang/apim/main_test.go b/utils/build/docker/golang/apim/main_test.go index 46bf0d9c17f..ef251dc20eb 100644 --- a/utils/build/docker/golang/apim/main_test.go +++ b/utils/build/docker/golang/apim/main_test.go @@ -184,18 +184,21 @@ func testRequestBodyBlockFixture(t *testing.T) { func TestResponsePhaseBlocksDiscardUpstreamResponse(t *testing.T) { for _, fixture := range []struct { - name string - steps []calloutStep + name string + wantStatus int + steps []calloutStep }{ { - name: "response-headers", + name: "response-headers", + wantStatus: 451, steps: []calloutStep{ {phase: "", response: `{"request-id":"request-123"}`}, {phase: "", requestID: fixtureRequestID, response: `{"block":{"status":451,"headers":{"X-Block":["response-headers"]},"content":"YmxvY2tlZA=="}}`}, }, }, { - name: "response-body", + name: "response-body", + wantStatus: 452, steps: []calloutStep{ {phase: "", response: `{"request-id":"request-123"}`}, {phase: "", requestID: fixtureRequestID, response: `{"allowed-body-size":1024}`}, @@ -217,7 +220,7 @@ func TestResponsePhaseBlocksDiscardUpstreamResponse(t *testing.T) { recorder := httptest.NewRecorder() gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "http://example.test:7777/", nil)) - assertBlockResponse(t, recorder, fixture.steps[len(fixture.steps)-1].responseStatus(), fixture.name, "blocked") + assertBlockResponse(t, recorder, fixture.wantStatus, fixture.name, "blocked") if got := upstreamCalls.Load(); got != 1 { t.Errorf("upstream calls = %d, want 1", got) } @@ -272,15 +275,6 @@ func TestInlineBodyModeUsesTwoCallouts(t *testing.T) { } func TestFailClosedWhenCalloutPortIsClosed(t *testing.T) { - listener, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - closedURL := "http://" + listener.Addr().String() - if err := listener.Close(); err != nil { - t.Fatal(err) - } - var upstreamCalls atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { upstreamCalls.Add(1) @@ -288,7 +282,7 @@ func TestFailClosedWhenCalloutPortIsClosed(t *testing.T) { t.Cleanup(upstream.Close) var logs bytes.Buffer - gateway := newTestGateway(closedURL, upstream.URL, &logs) + gateway := newTestGateway(closedListenerAddress(t), upstream.URL, &logs) recorder := httptest.NewRecorder() gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "http://example.test:7777/", nil)) @@ -298,9 +292,98 @@ func TestFailClosedWhenCalloutPortIsClosed(t *testing.T) { if got := upstreamCalls.Load(); got != 0 { t.Errorf("upstream calls = %d, want 0", got) } - if got := logs.String(); !strings.Contains(got, `apim-gateway callout phase= request-id="" path="/" outcome=error`) { - t.Errorf("failure log = %q, want phase, empty request-id, path, and error outcome", got) + // Exactly one line: logCallout reports the callout failure, and failing closed must not + // duplicate it. + assertLogPrefixes(t, logs.String(), []string{ + `apim-gateway callout phase= request-id="" path="/" outcome=error`, + }) +} + +func TestFailClosedWhenUpstreamPortIsClosed(t *testing.T) { + callout := newScriptedCallout(t, []calloutStep{{ + phase: "", + requestID: "", + response: `{"request-id":"request-123"}`, + }}) + + var logs bytes.Buffer + gateway := newTestGateway(callout.URL, closedListenerAddress(t), &logs) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "http://example.test:7777/", nil)) + + if got := recorder.Code; got != http.StatusBadGateway { + t.Fatalf("status = %d, want %d", got, http.StatusBadGateway) } + // The callout succeeded, so the upstream stage was reached and is what failed. + assertLogPrefixes(t, logs.String(), []string{ + `apim-gateway callout phase= request-id="request-123" path="/" outcome=ok`, + `apim-gateway fail-closed stage=call-upstream error="`, + }) +} + +func TestFailClosedOnMalformedBlock(t *testing.T) { + for _, fixture := range []struct { + name string + response string + wantLogged string + }{ + { + name: "undecodable-content", + response: `{"block":{"status":403,"headers":{"X-Block":["headers"]},"content":"!!!"}}`, + wantLogged: `apim-gateway fail-closed stage=write-block-request-headers error="illegal base64 data`, + }, + { + name: "out-of-range-status", + response: `{"block":{"status":42,"headers":{"X-Block":["headers"]},"content":"YmxvY2tlZA=="}}`, + wantLogged: `apim-gateway fail-closed stage=write-block-request-headers error="invalid block status 42"`, + }, + } { + t.Run(fixture.name, func(t *testing.T) { + callout := newScriptedCallout(t, []calloutStep{{ + phase: "", + requestID: "", + response: fixture.response, + }}) + + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + upstreamCalls.Add(1) + })) + t.Cleanup(upstream.Close) + + var logs bytes.Buffer + gateway := newTestGateway(callout.URL, upstream.URL, &logs) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "http://example.test:7777/", nil)) + + if got := recorder.Code; got != http.StatusBadGateway { + t.Fatalf("status = %d, want %d", got, http.StatusBadGateway) + } + if got := recorder.Header().Get("X-Block"); got != "" { + t.Errorf("X-Block = %q, want no block header on a fail-closed response", got) + } + if got := upstreamCalls.Load(); got != 0 { + t.Errorf("upstream calls = %d, want 0", got) + } + assertLogPrefixes(t, logs.String(), []string{ + `apim-gateway callout phase= request-id="" path="/" outcome=ok`, + fixture.wantLogged, + }) + }) + } +} + +func closedListenerAddress(t *testing.T) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + address := "http://" + listener.Addr().String() + if err := listener.Close(); err != nil { + t.Fatal(err) + } + return address } func newTestGateway(calloutURL, upstreamURL string, stderr io.Writer) *gateway { @@ -366,18 +449,6 @@ func newScriptedCallout(t *testing.T, steps []calloutStep) *httptest.Server { return server } -func (step calloutStep) responseStatus() int { - var response struct { - Block struct { - Status int `json:"status"` - } `json:"block"` - } - if err := json.Unmarshal([]byte(step.response), &response); err != nil { - panic(err) - } - return response.Block.Status -} - func assertGatewayOmitted(t *testing.T, message map[string]json.RawMessage) { t.Helper() if _, ok := message["gateway"]; ok { @@ -491,7 +562,7 @@ func assertBlockResponse(t *testing.T, recorder *httptest.ResponseRecorder, want func assertLogLines(t *testing.T, logs string, want []string) { t.Helper() - got := strings.FieldsFunc(strings.TrimSpace(logs), func(r rune) bool { return r == '\n' }) + got := splitLogLines(logs) if len(got) != len(want) { t.Fatalf("log lines = %q, want %q", got, want) } @@ -501,3 +572,22 @@ func assertLogLines(t *testing.T, logs string, want []string) { } } } + +// assertLogPrefixes is assertLogLines for records whose tail carries a platform-dependent +// error string. The line count is still exact. +func assertLogPrefixes(t *testing.T, logs string, want []string) { + t.Helper() + got := splitLogLines(logs) + if len(got) != len(want) { + t.Fatalf("log lines = %q, want %d lines starting with %q", got, len(want), want) + } + for i := range want { + if !strings.HasPrefix(got[i], want[i]) { + t.Errorf("log line %d = %q, want prefix %q", i, got[i], want[i]) + } + } +} + +func splitLogLines(logs string) []string { + return strings.FieldsFunc(strings.TrimSpace(logs), func(r rune) bool { return r == '\n' }) +} From a72a1c25030956bda1e06479fddd8a4a99bf3b61 Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Thu, 13 Aug 2026 18:11:25 +0200 Subject: [PATCH 5/9] [golang] Cover nil-vs-zero and negative allowed-body-size allowed-body-size is a pointer, and a returned 0 is a limit rather than an absence, so both body phases must still fire with a fully truncated body. Every existing fixture returned a positive size, so narrowing the check from "is set" to "is positive" left the whole suite green -- the shim's single most important semantic rule was unguarded. Adds a four-phase fixture returning 0 on both header phases, asserting the callout receives an empty body while upstream and the client still receive the untruncated payloads. Under a mutation to a positive-value check, every pre-existing test still passes and only this one fails, which is what makes it worth having. Also covers the negative-size guard. Deleting it panics with slice bounds out of range [:-1] at the body[:*limit] site, so the guard's rationale is now verified rather than asserted; the shim returns 502 with a fail-closed diagnostic instead. --- utils/build/docker/golang/apim/main_test.go | 116 ++++++++++++++++++++ 1 file changed, 116 insertions(+) diff --git a/utils/build/docker/golang/apim/main_test.go b/utils/build/docker/golang/apim/main_test.go index ef251dc20eb..b96f21bc682 100644 --- a/utils/build/docker/golang/apim/main_test.go +++ b/utils/build/docker/golang/apim/main_test.go @@ -274,6 +274,88 @@ func TestInlineBodyModeUsesTwoCallouts(t *testing.T) { }) } +// TestZeroAllowedBodySizeStillSendsBodyPhases pins the pointer semantics of allowed-body-size: +// a returned 0 is a limit, not an absence, so both body phases must still fire with a fully +// truncated body. Every other fixture returns a positive size, so a regression from +// `!= nil` to a positive-value test would leave them all green. +func TestZeroAllowedBodySizeStillSendsBodyPhases(t *testing.T) { + const requestBody = "request-payload" + const responseBody = "response-payload" + + callout := newScriptedCallout(t, []calloutStep{ + { + phase: "", + requestID: "", + response: `{"request-id":"request-123","allowed-body-size":0}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertRequestHeaderAddresses(t, message, nil) + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertBodyAddresses(t, message, "") + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{"allowed-body-size":0}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertResponseHeaderAddresses(t, message, nil) + }, + }, + { + phase: "", + requestID: fixtureRequestID, + response: `{}`, + assert: func(t *testing.T, message map[string]json.RawMessage) { + assertBodyAddresses(t, message, "") + }, + }, + }) + + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstreamCalls.Add(1) + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read upstream request body: %v", err) + } + // Truncation applies to the callout copy only; the proxied request keeps the original. + if got := string(body); got != requestBody { + t.Errorf("upstream request body = %q, want %q", got, requestBody) + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, responseBody) + })) + t.Cleanup(upstream.Close) + + var logs bytes.Buffer + gateway := newTestGateway(callout.URL, upstream.URL, &logs) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, newFixtureRequest(http.MethodPost, strings.NewReader(requestBody))) + + if got := recorder.Code; got != http.StatusOK { + t.Fatalf("status = %d, want %d", got, http.StatusOK) + } + // Same asymmetry on the response side: the client still receives the whole payload. + if got := recorder.Body.String(); got != responseBody { + t.Errorf("response body = %q, want %q", got, responseBody) + } + if got := upstreamCalls.Load(); got != 1 { + t.Errorf("upstream calls = %d, want 1", got) + } + assertLogLines(t, logs.String(), []string{ + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + }) +} + func TestFailClosedWhenCalloutPortIsClosed(t *testing.T) { var upstreamCalls atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { @@ -373,6 +455,40 @@ func TestFailClosedOnMalformedBlock(t *testing.T) { } } +// TestFailClosedOnNegativeAllowedBodySize covers the guard that keeps a hostile or buggy +// allowed-body-size from reaching the body[:*limit] slice expression. Failing closed is the +// contract; panicking would take the whole gateway down. +func TestFailClosedOnNegativeAllowedBodySize(t *testing.T) { + callout := newScriptedCallout(t, []calloutStep{{ + phase: "", + requestID: "", + response: `{"request-id":"request-123","allowed-body-size":-1}`, + }}) + + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + upstreamCalls.Add(1) + })) + t.Cleanup(upstream.Close) + + var logs bytes.Buffer + gateway := newTestGateway(callout.URL, upstream.URL, &logs) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, newFixtureRequest(http.MethodPost, strings.NewReader("payload"))) + + if got := recorder.Code; got != http.StatusBadGateway { + t.Fatalf("status = %d, want %d", got, http.StatusBadGateway) + } + if got := upstreamCalls.Load(); got != 0 { + t.Errorf("upstream calls = %d, want 0", got) + } + // The diagnostic text is ours, so the whole line is pinned. + assertLogLines(t, logs.String(), []string{ + `apim-gateway callout phase= request-id="request-123" path="/resource?x=1" outcome=ok`, + `apim-gateway fail-closed stage=encode-request-body error="negative allowed-body-size"`, + }) +} + func closedListenerAddress(t *testing.T) string { t.Helper() listener, err := net.Listen("tcp", "127.0.0.1:0") From f0ce68972ecd6c402ccf9d6ea82e1e7ec7cd6180 Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Fri, 14 Aug 2026 11:42:11 +0200 Subject: [PATCH 6/9] [golang] Never write to stdout from a container constructor The apim end-to-end jobs hung at "Pull images" while every sibling golang variant passed. .github/actions/pull_images redirects get-image-list.py's stdout into compose.yaml and feeds that to `docker compose`, and container objects are constructed during that call -- so ApimCalloutContainer's logger.stdout() lines landed inside the compose document as a bogus top-level key: {'WARNING': 'binaries/golang-apim-callout-image not found...', 'services': {}} Both branches logged, which is why both the prod job (fallback) and the dev job (pointer found) hung. The sibling processor containers are unaffected only because they swallow the same branch silently. Switch to logger.info/logger.warning. The resolved tag still reaches stdout on a real run via GoProcessorContainer.post_start, which prints "Processor image: ", so the released-vs-commit-under-test distinction stays visible where it matters. Adds tests/test_the_test/test_get_image_list.py asserting that the script's stdout parses as a compose document whose only top-level key is `services`, parametrised over all three go-proxy weblogs and over both pointer states. Reintroducing either logger.stdout call fails it. It deliberately does not pin services == {}, since a pointer to an image absent from the local cache legitimately adds an entry. --- tests/test_the_test/test_get_image_list.py | 98 ++++++++++++++++++++++ utils/_context/containers.py | 14 ++-- 2 files changed, 106 insertions(+), 6 deletions(-) create mode 100644 tests/test_the_test/test_get_image_list.py diff --git a/tests/test_the_test/test_get_image_list.py b/tests/test_the_test/test_get_image_list.py new file mode 100644 index 00000000000..ba7aa059656 --- /dev/null +++ b/tests/test_the_test/test_get_image_list.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from contextlib import contextmanager +import os +from pathlib import Path +import subprocess +import sys +from typing import TYPE_CHECKING + +import pytest +import yaml + +from utils import scenarios + + +if TYPE_CHECKING: + from collections.abc import Iterator + + +SCRIPT = Path("utils/scripts/get-image-list.py") +SCENARIOS = "APPSEC_BLOCKING,DEFAULT" + +# .github/actions/pull_images redirects the script's stdout into compose.yaml, then feeds that file +# to `docker compose`. Container objects are built during that call, so anything a container +# constructor writes on stdout becomes part of the compose document. +# +# The go proxy weblogs are the ones exposed to this: each of them resolves its processor image from +# an optional pointer file in binaries/, a branch that has historically been tempting to report on +# stdout. +GO_PROXY_POINTER_FILES = { + "apim": Path("binaries/golang-apim-callout-image"), + "envoy": Path("binaries/golang-service-extensions-callout-image"), + "haproxy": Path("binaries/golang-haproxy-spoa-image"), +} + + +@contextmanager +def _pointer_file(path: Path, *, present: bool) -> Iterator[None]: + """Force the presence or the absence of a binaries/ image pointer file. + + A pointer file already sitting there is a local build artifact: move it aside and put it back, + so both states can be tested whatever the checkout looks like. + """ + + backup = path.parent / f"{path.name}.test_the_test_backup" + existed = path.is_file() + + if existed: + path.rename(backup) + + try: + if present: + path.write_text("ghcr.io/datadog/system-tests/fake-processor:test-the-test\n", encoding="utf-8") + + yield + finally: + path.unlink(missing_ok=True) + + if existed: + backup.rename(path) + + +def _run_get_image_list(weblog: str) -> str: + result = subprocess.run( + [sys.executable, str(SCRIPT), SCENARIOS, "-l=golang", f"-w={weblog}"], + check=False, + capture_output=True, + text=True, + env={**os.environ, "PYTHONPATH": "."}, + ) + + assert result.returncode == 0, result.stderr + + return result.stdout + + +@scenarios.test_the_test +class Test_GetImageList: + @pytest.mark.parametrize("weblog", sorted(GO_PROXY_POINTER_FILES)) + @pytest.mark.parametrize("pointer_present", [False, True], ids=["pointer_absent", "pointer_present"]) + def test_stdout_is_only_a_compose_document(self, weblog: str, pointer_present: bool): # noqa: FBT001 + """get-image-list.py stdout is a compose file, it must not carry anything else. + + A container constructor calling logger.stdout() would inject its message as an extra + top-level key, and `docker compose` would then reject the generated compose.yaml. + """ + + with _pointer_file(GO_PROXY_POINTER_FILES[weblog], present=pointer_present): + stdout = _run_get_image_list(weblog) + + document = yaml.safe_load(stdout) + + assert isinstance(document, dict), f"stdout for weblog {weblog} is not a YAML mapping:\n{stdout}" + assert sorted(document) == ["services"], ( + f"stdout for weblog {weblog} must only contain the services key. " + f"A container constructor is writing on stdout instead of logging:\n{stdout}" + ) + assert isinstance(document["services"], dict), f"services must be a mapping:\n{stdout}" diff --git a/utils/_context/containers.py b/utils/_context/containers.py index 263de478dca..d0bf2135077 100644 --- a/utils/_context/containers.py +++ b/utils/_context/containers.py @@ -1822,14 +1822,16 @@ def __init__( try: with open("binaries/golang-apim-callout-image", encoding="utf-8") as f: image = f.read().strip() - logger.stdout(f"apim-callout image: {image} (from binaries/golang-apim-callout-image)") + # never logger.stdout() from a container constructor: containers are also built by + # utils/scripts/get-image-list.py, whose stdout IS the compose document consumed by + # `docker compose` in .github/actions/pull_images. The resolved tag still reaches + # stdout on a real run, via GoProcessorContainer.post_start(). + logger.info(f"apim-callout image: {image} (from binaries/golang-apim-callout-image)") except FileNotFoundError: image = "ghcr.io/datadog/dd-trace-go/apim-callout:latest" - # make the downgrade loud: without the pointer file we test released code instead of - # the commit under test - logger.stdout( - f"WARNING: binaries/golang-apim-callout-image not found, falling back to released image {image}" - ) + # the downgrade is significant: without the pointer file we test released code instead + # of the commit under test + logger.warning(f"binaries/golang-apim-callout-image not found, falling back to released image {image}") environment: dict[str, str | None] = { "DD_APPSEC_ENABLED": "true", From fe965689a299fc27469ab03220f5e38a4e99f819 Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Fri, 14 Aug 2026 12:12:50 +0200 Subject: [PATCH 7/9] [golang] Treat the response-body callout phase as upstream-dependent The deferred probes asserted four callout phases and failed in CI with three: never fired. The gateway only makes that call when the callout answers with allowed-body-size, and the callout only asks for a response body it can parse. The stock http-app (jasonrm/dummy-server) has a single 34-line handler that answers every request with the status code as text/plain, so there is no request that can make it return a parseable body -- pointing the probes elsewhere is not an option. Across both CI jobs, 1164 requests produced zero phases under two different callout versions, while the 20 phases line up exactly with the JSON-body POSTs, so the request side works and the response side is upstream-driven. Local validation missed this because the amd64-only http-app cannot run on arm64 and was substituted with a JSON-returning stand-in, which makes the fourth phase fire. The comment on UPSTREAM_DEPENDENT_PHASE says so, to stop someone re-requiring it after validating the same way. So the deferred assertion now requires the three guaranteed phases and tolerates a single trailing . What matters is preserved: the deferred flow must contain and the inline flow must not, which is the distinction these tests exist to prove. Renames the test and probe path that claimed four calls. --- .../external_processing/test_apim_callout.py | 69 +++++++++++++++---- 1 file changed, 56 insertions(+), 13 deletions(-) diff --git a/tests/external_processing/test_apim_callout.py b/tests/external_processing/test_apim_callout.py index ea03167eb72..420784b459b 100644 --- a/tests/external_processing/test_apim_callout.py +++ b/tests/external_processing/test_apim_callout.py @@ -11,10 +11,23 @@ CALLOUT_LOG_PATTERN = re.compile( r'apim-gateway callout phase=(?P<[^>]+>) request-id="(?P[^"]+)" path="(?P[^"]+)" outcome=ok' ) -DEFERRED_PHASES = ("", "", "", "") +# Deferred (default) body mode: the bodies are not inlined, so the callout answers +# with `allowed-body-size` and the gateway makes a separate call. These three phases +# are guaranteed, in this order. +DEFERRED_PHASES = ("", "", "") +# Inline body mode: both bodies ride along on the header calls, which is exactly what suppresses +# `allowed-body-size`, so no body phase is ever requested on either side. INLINE_PHASES = ("", "") +# A fourth phase is possible but NOT guaranteed, so it is tolerated and never +# required. The gateway only makes it when the callout answers with +# `allowed-body-size`, and the callout only asks for the response body when the upstream returned +# one it can parse. That is a property of the upstream, not of the gateway: the stock `http-app` +# (jasonrm/dummy-server) answers every request with `text/plain` containing the status code, so +# this phase never fires in CI. Do not turn it back into a required phase after validating against +# a substituted upstream -- a JSON-returning stand-in makes it fire and hides this distinction. +UPSTREAM_DEPENDENT_PHASE = "" APIM_SPAN = ("web", "server", "apim-callout") -DEFAULT_PROBE_PATH = "/?probe=default-four-call" +DEFAULT_PROBE_PATH = "/?probe=default-deferred-body" INLINE_PROBE_PATH = "/?probe=inline-two-call" TRACE_DEFAULT_PROBE_PATH = "/?probe=trace-default" TRACE_INLINE_PROBE_PATH = "/?probe=trace-inline" @@ -36,7 +49,15 @@ def _callout_phases_by_request_id_and_path() -> dict[tuple[str, str], list[str]] return dict(phases_by_request_id_and_path) -def _assert_probe_phases(probe_path: str, expected_phases: tuple[str, ...]) -> None: +def _probe_phase_group(probe_path: str) -> tuple[str, list[str]]: + """Return the (request-id, phases) of the single callout group that belongs to `probe_path`. + + Every request reaching the gateway -- including its own bodiless healthcheck -- appends to the + same stderr log, so phases are correlated by (request-id, path) and never counted globally. + Each probe path is requested exactly once, so exactly one group must match. Zero groups means + the gateway log format drifted away from CALLOUT_LOG_PATTERN, which would otherwise make every + phase assertion below pass vacuously. + """ matching_groups = { request_id: phases for (request_id, path), phases in _callout_phases_by_request_id_and_path().items() @@ -46,9 +67,29 @@ def _assert_probe_phases(probe_path: str, expected_phases: tuple[str, ...]) -> N f"expected exactly one callout request-id group for probe {probe_path}, " f"got {len(matching_groups)}: {sorted(matching_groups)}" ) - request_id, phases = next(iter(matching_groups.items())) - assert phases == list(expected_phases), ( - f"probe {probe_path} (request-id {request_id}) hit callout phases {phases}, expected {list(expected_phases)}" + return next(iter(matching_groups.items())) + + +def _assert_deferred_probe_phases(probe_path: str) -> None: + """Assert `probe_path` was served in deferred body mode: a separate callout. + + A single trailing is accepted because it depends on the upstream returning a + parseable response body (see UPSTREAM_DEPENDENT_PHASE); everything before it is required. + """ + request_id, phases = _probe_phase_group(probe_path) + required_phases = phases[:-1] if phases[-1:] == [UPSTREAM_DEPENDENT_PHASE] else phases + assert required_phases == list(DEFERRED_PHASES), ( + f"deferred probe {probe_path} (request-id {request_id}) hit callout phases {phases}, " + f"expected {list(DEFERRED_PHASES)} optionally followed by {UPSTREAM_DEPENDENT_PHASE}" + ) + + +def _assert_inline_probe_phases(probe_path: str) -> None: + """Assert `probe_path` was served in inline body mode: header phases only, no body phase.""" + request_id, phases = _probe_phase_group(probe_path) + assert phases == list(INLINE_PHASES), ( + f"inline probe {probe_path} (request-id {request_id}) hit callout phases {phases}, " + f"expected {list(INLINE_PHASES)}" ) @@ -70,19 +111,21 @@ def _trace_structure(request: HttpResponse) -> list[tuple[str, str, str]]: @irrelevant(context.weblog_variant != "apim") @features.go_proxies class Test_ApimCallout: - def setup_default_body_mode_uses_four_callouts(self) -> None: + def setup_default_body_mode_defers_request_body_callout(self) -> None: self.r = weblog.post(DEFAULT_PROBE_PATH, json={"body": "default"}) - def test_default_body_mode_uses_four_callouts(self) -> None: + def test_default_body_mode_defers_request_body_callout(self) -> None: + """Without the inline header, the request body is fetched by its own callout.""" assert self.r.status_code == 200, f"deferred probe returned {self.r.status_code}, expected 200" - _assert_probe_phases(DEFAULT_PROBE_PATH, DEFERRED_PHASES) + _assert_deferred_probe_phases(DEFAULT_PROBE_PATH) def setup_inline_body_mode_uses_two_callouts(self) -> None: self.r = weblog.post(INLINE_PROBE_PATH, json={"body": "inline"}, headers={"X-Datadog-Apim-Body-Mode": "inline"}) def test_inline_body_mode_uses_two_callouts(self) -> None: + """Inlining the body on the header calls removes the body phases, leaving exactly two calls.""" assert self.r.status_code == 200, f"inline probe returned {self.r.status_code}, expected 200" - _assert_probe_phases(INLINE_PROBE_PATH, INLINE_PHASES) + _assert_inline_probe_phases(INLINE_PROBE_PATH) def setup_inline_body_mode_preserves_trace_structure(self) -> None: self.default_response = weblog.post(TRACE_DEFAULT_PROBE_PATH, json={"body": "default trace"}) @@ -99,8 +142,8 @@ def test_inline_body_mode_preserves_trace_structure(self) -> None: assert self.inline_response.status_code == 200, ( f"inline probe returned {self.inline_response.status_code}, expected 200" ) - _assert_probe_phases(TRACE_DEFAULT_PROBE_PATH, DEFERRED_PHASES) - _assert_probe_phases(TRACE_INLINE_PROBE_PATH, INLINE_PHASES) + _assert_deferred_probe_phases(TRACE_DEFAULT_PROBE_PATH) + _assert_inline_probe_phases(TRACE_INLINE_PROBE_PATH) default_spans = _trace_structure(self.default_response) inline_spans = _trace_structure(self.inline_response) assert default_spans == inline_spans, ( @@ -117,7 +160,7 @@ def setup_inline_body_mode_closes_request_state(self) -> None: def test_inline_body_mode_closes_request_state(self) -> None: assert self.r.status_code == 200, f"inline probe returned {self.r.status_code}, expected 200" - _assert_probe_phases(STATE_CLOSURE_PROBE_PATH, INLINE_PHASES) + _assert_inline_probe_phases(STATE_CLOSURE_PROBE_PATH) assert "closing orphaned span" not in _container_stderr("apim-callout"), ( "apim-callout logged an orphaned span, so inline mode did not close the cached request state" ) From 43cefc84453afef8b9c7dc17a31233c15fee8c93 Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Fri, 14 Aug 2026 16:22:19 +0200 Subject: [PATCH 8/9] [golang] Widen apim-gateway health budget, canonicalize block headers The health budget was tightened to a ~21s ceiling off a single 11.93s measurement, which was too aggressive. `go run .` rebuilds net/http and crypto/tls from source into an empty GOCACHE on every container start -- ~12s wall but ~13s of CPU locally -- so a contended 2-vCPU runner can exceed 21s, and the failure mode is a dead scenario rather than a slow one. execute_command stops at the first success, so a high retry count costs nothing when the container is already up. It also treats start_period as an unconditional sleep rather than a Docker grace period, so dropping it makes a fast start detected sooner than the previous blind 15s wait. Every other container in the file uses retries 10-60 with no start_period; 5 was the lowest value present. Also route block headers through applyHeaders. It already canonicalizes, and writeBlock was the one place writing callout-supplied header names verbatim. Not currently reachable, since dd-trace-go builds those headers via Header().Set and they arrive canonical, but the failure mode is quiet: a lowercase content-type would be missed by Go's internal Header.get, so the server would content-sniff and emit a second Content-Type. --- utils/_context/containers.py | 16 +++++++++++----- utils/build/docker/golang/apim/main.go | 6 +++--- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/utils/_context/containers.py b/utils/_context/containers.py index d0bf2135077..df797ab0536 100644 --- a/utils/_context/containers.py +++ b/utils/_context/containers.py @@ -1801,12 +1801,18 @@ def __init__(self) -> None: # golang:1.25-alpine ships busybox but no bash, so the /bin/bash + /dev/tcp # healthcheck used by EnvoyContainer and HAProxyContainer is not usable here "test": "wget -qO- http://localhost:80/", - # AC1 measured ~11.93s from container start to first successful health probe. - # `go run .` still compiles on every start, so keep a small cushion: ~21s - # ceiling (15s + 6 * 1s). - "retries": 5, + # `go run .` compiles the shim on every container start, and since Go 1.20 ships + # no prebuilt std it rebuilds net/http and crypto/tls from source into an empty + # GOCACHE. That measured ~12s wall locally but ~13s of CPU, so a contended + # 2-vCPU runner can take substantially longer. + # + # execute_command retries until the first success and stops, so a high retry + # count costs nothing on a fast start but keeps a slow runner from failing the + # whole scenario. start_period is an unconditional sleep here, not a Docker + # grace period, so it is omitted: probing immediately detects a fast start + # sooner than any blind wait would. + "retries": 60, "interval": 1_000_000_000, - "start_period": 15_000_000_000, }, ) diff --git a/utils/build/docker/golang/apim/main.go b/utils/build/docker/golang/apim/main.go index e953f5922b4..20a627d39fa 100644 --- a/utils/build/docker/golang/apim/main.go +++ b/utils/build/docker/golang/apim/main.go @@ -312,9 +312,9 @@ func writeBlock(w http.ResponseWriter, block *blockResult) error { if block.Status < http.StatusContinue || block.Status > 999 { return fmt.Errorf("invalid block status %d", block.Status) } - for name, values := range block.Headers { - w.Header()[name] = append([]string(nil), values...) - } + // canonicalize: a lowercase name from the callout would not be found by Go's internal + // Header.get, so the server would content-sniff and emit a second Content-Type + applyHeaders(w.Header(), block.Headers) w.WriteHeader(block.Status) _, err = w.Write(content) return err From ec42fca26c1d272ce61f9d3e0e6af3c953665b3d Mon Sep 17 00:00:00 2001 From: Eliott Bouhana Date: Fri, 14 Aug 2026 17:00:06 +0200 Subject: [PATCH 9/9] [golang] Drop the sleep from apim setup, scope the orphan assertion pr-review.mdc forbids time.sleep() inside setup_ methods in end-to-end scenarios, and the 31s sleep also added that much unconditional wall time to every DEFAULT run. The wait was unnecessary. Every setup_ runs before any test_, and the containers are stopped and their logs collected in between, so far more than the callout's 30s state TTL has already elapsed by the time the assertion reads the log. Also scope the assertion to the probe's own request-id, which _assert_inline_probe_phases already computed and discarded. Matching the bare warning text would fail on any unrelated request that legitimately orphaned state and blame inline mode for it. --- .../external_processing/test_apim_callout.py | 30 ++++++++++++++----- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/tests/external_processing/test_apim_callout.py b/tests/external_processing/test_apim_callout.py index 420784b459b..802e750bfd0 100644 --- a/tests/external_processing/test_apim_callout.py +++ b/tests/external_processing/test_apim_callout.py @@ -1,5 +1,4 @@ import re -import time from collections import defaultdict from pathlib import Path @@ -84,13 +83,17 @@ def _assert_deferred_probe_phases(probe_path: str) -> None: ) -def _assert_inline_probe_phases(probe_path: str) -> None: - """Assert `probe_path` was served in inline body mode: header phases only, no body phase.""" +def _assert_inline_probe_phases(probe_path: str) -> str: + """Assert `probe_path` was served in inline body mode: header phases only, no body phase. + + Returns the request-id, so a caller can scope further assertions to this exchange. + """ request_id, phases = _probe_phase_group(probe_path) assert phases == list(INLINE_PHASES), ( f"inline probe {probe_path} (request-id {request_id}) hit callout phases {phases}, " f"expected {list(INLINE_PHASES)}" ) + return request_id def _span_structure(span: DataDogLibrarySpan) -> tuple[str, str, str]: @@ -156,11 +159,24 @@ def setup_inline_body_mode_closes_request_state(self) -> None: json={"body": "inline state"}, headers={"X-Datadog-Apim-Body-Mode": "inline"}, ) - time.sleep(31) def test_inline_body_mode_closes_request_state(self) -> None: assert self.r.status_code == 200, f"inline probe returned {self.r.status_code}, expected 200" - _assert_inline_probe_phases(STATE_CLOSURE_PROBE_PATH) - assert "closing orphaned span" not in _container_stderr("apim-callout"), ( - "apim-callout logged an orphaned span, so inline mode did not close the cached request state" + request_id = _assert_inline_probe_phases(STATE_CLOSURE_PROBE_PATH) + # The callout evicts cached request state after a 30s TTL and logs one warning naming the + # request-id. Inline mode deletes the state on the response-headers call, so that warning + # must never appear for this exchange. No wait is needed here: every setup_ runs before any + # test_, and the containers are stopped and their logs collected in between, so far more + # than the TTL has elapsed by the time this assertion reads the log. + # + # Scoped to this request-id on purpose. Matching the bare warning text would also fail on + # an unrelated request that legitimately orphaned state, and attribute it to inline mode. + orphaned = [ + line + for line in _container_stderr("apim-callout").splitlines() + if "closing orphaned span" in line and request_id in line + ] + assert not orphaned, ( + f"apim-callout orphaned the cached state for inline request-id {request_id}, " + f"so inline mode did not close it: {orphaned}" )