chore: adding fuzzing tests (#7903)

* chore: adding fuzzing tests

* chore: adding fuzzing tests

* chore: adding fuzzing tests

Author: leandrodamascena

.clusterfuzzlite/Dockerfile

@@ -0,0 +1,12 @@
+FROM gcr.io/oss-fuzz-base/base-builder-python
+
+# Copy project source
+COPY . $SRC/powertools
+
+WORKDIR $SRC/powertools
+
+# Install project dependencies
+RUN pip3 install -e ".[all]"
+
+# Copy build script
+COPY .clusterfuzzlite/build.sh $SRC/

.clusterfuzzlite/build.sh

@@ -0,0 +1,6 @@
+#!/bin/bash -eu
+
+# Build fuzz targets from tests/fuzz/
+for fuzzer in $(find $SRC/powertools/tests/fuzz -name 'fuzz_*.py'); do
+    compile_python_fuzzer "$fuzzer"
+done

.clusterfuzzlite/project.yaml

@@ -0,0 +1,4 @@
+language: python
+main_repo: https://github.com/aws-powertools/powertools-lambda-python
+sanitizers:
+  - address

.github/workflows/cflite_scheduled.yml

@@ -0,0 +1,34 @@
+name: ClusterFuzzLite fuzzing
+
+on:
+  schedule:
+    # Run daily at 8 AM UTC
+    - cron: "0 8 * * *"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  PR:
+    runs-on: ubuntu-latest
+    concurrency:
+      group: ${{ github.workflow }}-${{ github.ref }}
+      cancel-in-progress: true
+    steps:
+      - name: Build Fuzzers
+        id: build
+        uses: google/clusterfuzzlite/actions/build_fuzzers@884713a6c30a92e5e8544c39945cd7cb630abcd1 # v1
+        with:
+          language: python
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          sanitizer: address
+
+      - name: Run Fuzzers
+        id: run
+        uses: google/clusterfuzzlite/actions/run_fuzzers@884713a6c30a92e5e8544c39945cd7cb630abcd1 # v1
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          fuzz-seconds: 30
+          mode: code-change
+          sanitizer: address

ruff.toml

@@ -101,3 +101,4 @@ runtime-evaluated-base-classes = ["pydantic.BaseModel"]
 "examples/*" = ["FA100", "TCH"]
 "tests/*" = ["FA100"]
 "aws_lambda_powertools/utilities/parser/models/*" = ["FA100"]
+"tests/fuzz/*" = ["FA100", "F401", "E402"]

tests/fuzz/__init__.py

@@ -0,0 +1 @@
+"""Fuzz testing targets for ClusterFuzzLite."""

tests/fuzz/fuzz_event_sources.py

@@ -0,0 +1,77 @@
+"""Fuzz target for Event Source Data Classes - SQS, SNS, API Gateway, Kinesis parsing."""
+
+from __future__ import annotations
+
+import json
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from aws_lambda_powertools.utilities.data_classes import (
+        APIGatewayProxyEvent,
+        KinesisStreamEvent,
+        SNSEvent,
+        SQSEvent,
+    )
+
+
+def fuzz_sqs_event(data: bytes) -> None:
+    """Fuzz SQS event parsing."""
+    try:
+        event_dict = json.loads(data.decode("utf-8", errors="ignore"))
+        SQSEvent(event_dict)
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError):
+        pass
+    except Exception:
+        pass
+
+
+def fuzz_sns_event(data: bytes) -> None:
+    """Fuzz SNS event parsing."""
+    try:
+        event_dict = json.loads(data.decode("utf-8", errors="ignore"))
+        SNSEvent(event_dict)
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError):
+        pass
+    except Exception:
+        pass
+
+
+def fuzz_api_gateway_event(data: bytes) -> None:
+    """Fuzz API Gateway event parsing."""
+    try:
+        event_dict = json.loads(data.decode("utf-8", errors="ignore"))
+        APIGatewayProxyEvent(event_dict)
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError):
+        pass
+    except Exception:
+        pass
+
+
+def fuzz_kinesis_event(data: bytes) -> None:
+    """Fuzz Kinesis event parsing."""
+    try:
+        event_dict = json.loads(data.decode("utf-8", errors="ignore"))
+        KinesisStreamEvent(event_dict)
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError):
+        pass
+    except Exception:
+        pass
+
+
+def fuzz_all_events(data: bytes) -> None:
+    """Fuzz all event sources."""
+    fuzz_sqs_event(data)
+    fuzz_sns_event(data)
+    fuzz_api_gateway_event(data)
+    fuzz_kinesis_event(data)
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, fuzz_all_events)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

tests/fuzz/fuzz_parser.py

@@ -0,0 +1,36 @@
+"""Fuzz target for Parser - Pydantic event validation."""
+
+from __future__ import annotations
+
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from pydantic import BaseModel, ValidationError
+
+    from aws_lambda_powertools.utilities.parser import parse
+
+
+class SimpleModel(BaseModel):
+    name: str
+    value: int
+
+
+def fuzz_parser(data: bytes) -> None:
+    """Fuzz the parser with arbitrary JSON-like data."""
+    try:
+        parse(event=data.decode("utf-8", errors="ignore"), model=SimpleModel)
+    except (ValidationError, ValueError, TypeError, KeyError):
+        pass
+    except Exception:
+        pass
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, fuzz_parser)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

tests/fuzz/fuzz_validation.py

@@ -0,0 +1,42 @@
+"""Fuzz target for Validation - JSON Schema validation."""
+
+from __future__ import annotations
+
+import json
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from aws_lambda_powertools.utilities.validation import validate
+    from aws_lambda_powertools.utilities.validation.exceptions import SchemaValidationError
+
+SIMPLE_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+        "name": {"type": "string"},
+        "age": {"type": "integer"},
+    },
+    "required": ["name"],
+}
+
+
+def fuzz_validation(data: bytes) -> None:
+    """Fuzz JSON Schema validation."""
+    try:
+        event = json.loads(data.decode("utf-8", errors="ignore"))
+        validate(event=event, schema=SIMPLE_SCHEMA)
+    except (json.JSONDecodeError, SchemaValidationError, TypeError, ValueError):
+        pass
+    except Exception:
+        pass
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, fuzz_validation)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()