chore: add integration tests around Serverless ASM

.github/workflows/serverless-integration.yml

@@ -12,8 +12,11 @@ jobs:
   test:
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
         architecture: [amd64, arm64]
+        suite: [metric, log, trace, appsec]
+    name: ${{ matrix.suite }} on ${{ matrix.architecture }}
     steps:
       - name: Checkout datadog-agent repository
         uses: actions/checkout@v3
@@ -26,7 +29,7 @@ jobs:
           node-version: 14
 
       - name: Install Serverless Framework
-        run: sudo yarn global add serverless@^3.7.9 --prefix /usr/local
+        run: sudo yarn global add serverless@^3.36.0 --prefix /usr/local
 
       - name: Checkout the datadog-lambda-extension repository
         uses: actions/checkout@v3
@@ -44,6 +47,12 @@ jobs:
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v2
 
+      - name: Create raw logs directory
+        id: rawlogs
+        run: |-
+          DIR=$(mktemp -d)
+          echo "dir=${DIR}" >> $GITHUB_OUTPUT
+
       - name: Run tests if AWS credentials are available
         id: test
         uses: nick-fields/retry@v2
@@ -54,15 +63,14 @@ jobs:
           timeout_minutes: 45
           max_attempts: 2
           command: |
-            export RAWLOGS_DIR=$(mktemp -d)
-            echo "rawlogs=$RAWLOGS_DIR" >> $GITHUB_OUTPUT
+            RAWLOGS_DIR="${{ steps.rawlogs.outputs.dir }}/${{ matrix.architecture }}}"
             cd go/src/github.com/DataDog/datadog-agent
             ARCHITECTURE=${{ matrix.architecture }} RAWLOGS_DIR=$RAWLOGS_DIR \
-                ./test/integration/serverless/run.sh
+                ./test/integration/serverless/run.sh ${{ matrix.suite }}
 
       - name: Archive raw logs
         if: always()
         uses: actions/upload-artifact@v3
         with:
           name: rawlogs
-          path: ${{ steps.test.outputs.rawlogs }}
+          path: ${{ steps.rawlogs.outputs.dir }}

test/integration/serverless/appsec-payload.json

@@ -0,0 +1,70 @@
+{
+  "version": "1.0",
+  "resource": "/my/path",
+  "path": "/my/path",
+  "httpMethod": "GET",
+  "headers": {
+    "User-Agent": "Arachni/v1"
+  },
+  "multiValueHeaders": {
+    "User-Agent": [
+      "Arachni/v1"
+    ]
+  },
+  "queryStringParameters": {
+    "sql": "UNION SELECT password FROM users"
+  },
+  "multiValueQueryStringParameters": {
+    "sql": [
+      "UNION SELECT password FROM users"
+    ]
+  },
+  "requestContext": {
+    "accountId": "123456789012",
+    "apiId": "id",
+    "authorizer": {
+      "claims": null,
+      "scopes": null
+    },
+    "domainName": "id.execute-api.us-east-1.amazonaws.com",
+    "domainPrefix": "id",
+    "extendedRequestId": "request-id",
+    "httpMethod": "GET",
+    "identity": {
+      "accessKey": null,
+      "accountId": null,
+      "caller": null,
+      "cognitoAuthenticationProvider": null,
+      "cognitoAuthenticationType": null,
+      "cognitoIdentityId": null,
+      "cognitoIdentityPoolId": null,
+      "principalOrgId": null,
+      "sourceIp": "192.0.2.1",
+      "user": null,
+      "userAgent": "user-agent",
+      "userArn": null,
+      "clientCert": {
+        "clientCertPem": "CERT_CONTENT",
+        "subjectDN": "www.example.com",
+        "issuerDN": "Example issuer",
+        "serialNumber": "a1:a1:a1:a1:a1:a1:a1:a1:a1:a1:a1:a1:a1:a1:a1:a1",
+        "validity": {
+          "notBefore": "May 28 12:30:02 2019 GMT",
+          "notAfter": "Aug  5 09:36:04 2021 GMT"
+        }
+      }
+    },
+    "path": "/my/path",
+    "protocol": "HTTP/1.1",
+    "requestId": "id=",
+    "requestTime": "04/Mar/2020:19:15:17 +0000",
+    "requestTimeEpoch": 1583349317135,
+    "resourceId": null,
+    "resourcePath": "/my/path",
+    "stage": "$default"
+  },
+  "pathParameters": null,
+  "stageVariables": null,
+  "body": "Hello from Lambda!",
+  "isBase64Encoded": false
+}

test/integration/serverless/build_csharp_functions.sh

@@ -14,4 +14,4 @@ else
     CONVERTED_ARCH="x86_64"
 fi
 
-dotnet lambda package --configuration Release --framework net6.0 --verbosity quiet --output-package bin/Release/net6.0/handler.zip --function-architecture $CONVERTED_ARCH
+PATH="${PATH}:${HOME}/.dotnet/tools" dotnet lambda package --configuration Release --framework net6.0 --verbosity quiet --output-package bin/Release/net6.0/handler.zip --function-architecture $CONVERTED_ARCH

test/integration/serverless/build_go_functions.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
 echo "Building Go functions for $ARCHITECTURE architecture"
-go_test_dirs=("metric" "log" "timeout" "trace")
+go_test_dirs=("metric" "log" "timeout" "trace" "appsec")
 cd src
 for go_dir in "${go_test_dirs[@]}"; do
     env GOOS=linux CGO_ENABLED=0 GOARCH=$ARCHITECTURE go build -ldflags="-s -w" -o bin/"$go_dir" go-tests/"$go_dir"/main.go

test/integration/serverless/build_java_functions.sh

@@ -1,8 +1,8 @@
 #!/bin/bash
 
 echo "Building Java Lambda Functions"
-java_test_dirs=("metric" "trace" "log" "timeout" "error")
+java_test_dirs=("metric" "trace" "log" "timeout" "error" "appsec")
 cd src
 for java_dir in "${java_test_dirs[@]}"; do
     mvn package -q -f java-tests/"${java_dir}"/pom.xml
-done
+done

test/integration/serverless/log_normalize.py

@@ -1,6 +1,9 @@
 import argparse
 import json
+import os
 import re
+import traceback
+from typing import Union
 
 
 def normalize_metrics(stage):
@@ -85,6 +88,7 @@ def sort__dd_tags_container(log):
         tags = tags.split(',')
         tags.sort()
         log["tags"]["_dd.tags.container"] = ','.join(tags)
+        return log
 
     return [
         require(r'BEGINTRACE.*ENDTRACE'),
@@ -104,13 +108,44 @@ def sort__dd_tags_container(log):
         replace(r'("otel.trace_id":")[a-zA-Z0-9]+"', r'\1null"'),
         replace(r'("faas.execution":")[a-zA-Z0-9-]+"', r'\1null"'),
         replace(r'("faas.instance":")[a-zA-Z0-9-/]+\[\$LATEST\][a-zA-Z0-9]+"', r'\1null"'),
+        replace(r'("_dd.tracer_hostname":)"\d{1,3}(?:.\d{1,3}){3}"+', r'\1"<redacted>"'),
         replace(stage, 'XXXXXX'),
         exclude(r'[ ]$'),
         foreach(sort__dd_tags_container),
         sort_by(trace_sort_key),
     ]
 
 
+def normalize_appsec(stage):
+    def select__dd_appsec_json(log):
+        """Selects the content of spans.*.meta.[_dd.appsec.json] which is
+        unfortunately an embedded JSON string value, so it's parsed out.
+        """
+
+        entries = []
+
+        for chunk in log["chunks"]:
+            for span in chunk.get("spans") or []:
+                meta = span.get("meta") or {}
+                data = meta.get("_dd.appsec.json")
+                if data is None:
+                    continue
+                parsed = json.loads(data, strict=False)
+                # The triggers may appear in any order, so we sort them by rule ID
+                parsed["triggers"] = sorted(parsed["triggers"], key=lambda x: x["rule"]["id"])
+                entries.append(parsed)
+
+        return entries
+
+    return [
+        require(r'BEGINTRACE.*ENDTRACE'),
+        exclude(r'BEGINTRACE'),
+        exclude(r'ENDTRACE'),
+        flatmap(select__dd_appsec_json),
+        replace(stage, 'XXXXXX'),
+    ]
+
+
 #####################
 # BEGIN NORMALIZERS #
 #####################
@@ -159,11 +194,28 @@ def _foreach(log):
         logs = json.loads(log, strict=False)
         for log_item in logs:
             fn(log_item)
-        return json.dumps(logs)
+        return json.dumps(logs, sort_keys=True)
 
     return _foreach
 
 
+def flatmap(fn):
+    """
+    Execute fn with each element of the list in order, flatten the results.
+    """
+
+    def _flat_map(log):
+        logs = json.loads(log, strict=False)
+
+        mapped = []
+        for log_item in logs:
+            mapped.extend(fn(log_item))
+
+        return json.dumps(mapped, sort_keys=True)
+
+    return _flat_map
+
+
 def sort_by(key):
     """
     Sort the json entries using the given key function, requires the log string
@@ -211,12 +263,17 @@ def get_normalizers(typ, stage):
         return normalize_logs(stage)
     elif typ == 'traces':
         return normalize_traces(stage)
+    elif typ == 'appsec':
+        return normalize_appsec(stage)
     else:
         raise ValueError(f'invalid type "{typ}"')
 
 
 def format_json(log):
-    return json.dumps(json.loads(log, strict=False), indent=2)
+    try:
+        return json.dumps(json.loads(log, strict=False), indent=2)
+    except json.JSONDecodeError:
+        return log
 
 
 def parse_args():
@@ -230,9 +287,21 @@ def parse_args():
 if __name__ == '__main__':
     try:
         args = parse_args()
+
+        if args.logs.startswith('file:'):
+            with open(args.logs[5:], 'r') as f:
+                args.logs = f.read()
+
         print(normalize(args.logs, args.type, args.stage))
-    except Exception:
-        err = {"error": "normalization raised exception"}
+    except Exception as e:
+        err: dict[str, Union[str, list[str]]] = {
+            "error": "normalization raised exception",
+        }
+        # Unless explicitly specified, perform as it did historically
+        if os.environ.get("TRACEBACK") == "true":
+            err["message"] = str(e)
+            err["backtrace"] = traceback.format_exception(type(e), e, e.__traceback__)
+
         err_json = json.dumps(err, indent=2)
         print(err_json)
         exit(1)