fix(api): sign S3 report download URLs with Signature Version 4

api/changelog.d/s3-report-download-sigv4.fixed.md

@@ -0,0 +1 @@
+Scan report downloads from an S3 bucket with default SSE-KMS encryption no longer fail with an `InvalidArgument` error, by signing presigned download URLs with AWS Signature Version 4

api/src/backend/tasks/jobs/export.py

@@ -6,6 +6,7 @@
 import config.django.base as base
 from api.db_utils import rls_transaction
 from api.models import Scan
+from botocore.config import Config
 from botocore.exceptions import ClientError, NoCredentialsError, ParamValidationError
 from celery.utils.log import get_task_logger
 from django.conf import settings
@@ -215,6 +216,9 @@ def get_s3_client():
     Raises:
         ClientError, NoCredentialsError, or ParamValidationError if both attempts to create a client fail.
     """
+    # Pinning SigV4 is required: boto3's default query-string signer for S3 falls back to
+    # SigV2, which S3 rejects for objects encrypted with SSE-KMS.
+    s3_config = Config(signature_version="s3v4")
     s3_client = None
     try:
         s3_client = boto3.client(
@@ -223,10 +227,11 @@ def get_s3_client():
             aws_secret_access_key=settings.DJANGO_OUTPUT_S3_AWS_SECRET_ACCESS_KEY,
             aws_session_token=settings.DJANGO_OUTPUT_S3_AWS_SESSION_TOKEN,
             region_name=settings.DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION,
+            config=s3_config,
         )
         s3_client.list_buckets()
     except (ClientError, NoCredentialsError, ParamValidationError, ValueError):
-        s3_client = boto3.client("s3")
+        s3_client = boto3.client("s3", config=s3_config)
         s3_client.list_buckets()
 
     return s3_client

api/src/backend/tasks/tests/test_export.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import boto3
 import pytest
 from botocore.exceptions import ClientError
 from tasks.jobs.export import (
@@ -49,13 +50,65 @@ def test_get_s3_client_success(self, mock_settings, mock_boto_client):
 
     @patch("tasks.jobs.export.boto3.client")
     @patch("tasks.jobs.export.settings")
-    def test_get_s3_client_fallback(self, mock_settings, mock_boto_client):
-        mock_boto_client.side_effect = [
-            ClientError({"Error": {"Code": "403"}}, "ListBuckets"),
-            MagicMock(),
-        ]
+    def test_get_s3_client_generates_sigv4_presigned_url(
+        self, mock_settings, mock_boto_client
+    ):
+        mock_settings.DJANGO_OUTPUT_S3_AWS_ACCESS_KEY_ID = "test-access-key"
+        mock_settings.DJANGO_OUTPUT_S3_AWS_SECRET_ACCESS_KEY = "test-secret-key"
+        mock_settings.DJANGO_OUTPUT_S3_AWS_SESSION_TOKEN = ""
+        mock_settings.DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION = "us-east-1"
+
+        def create_client(service_name, **kwargs):
+            # Build a real boto3 client so signing runs for real, only faking the
+            # network call used to validate the credentials. Use boto3.Session()
+            # rather than boto3.client() directly, since the latter is patched
+            # above and would recurse into this same side effect.
+            real_client = boto3.Session().client(service_name, **kwargs)
+            real_client.list_buckets = MagicMock()
+            return real_client
+
+        mock_boto_client.side_effect = create_client
+
         client = get_s3_client()
-        assert client is not None
+        url = client.generate_presigned_url(
+            "get_object",
+            Params={"Bucket": "test-bucket", "Key": "report.zip"},
+            ExpiresIn=300,
+        )
+
+        # SSE-KMS objects require SigV4; the default query signer falls back to SigV2.
+        assert "X-Amz-Algorithm=AWS4-HMAC-SHA256" in url
+
+    @patch("tasks.jobs.export.boto3.client")
+    @patch("tasks.jobs.export.settings")
+    def test_get_s3_client_fallback(self, mock_settings, mock_boto_client, monkeypatch):
+        # The fallback branch relies on boto3's default credential chain (e.g. an
+        # IAM role), so provide env credentials for the real client to sign with.
+        monkeypatch.setenv("AWS_ACCESS_KEY_ID", "fallback-access-key")
+        monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "fallback-secret-key")
+        monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
+
+        calls = {"count": 0}
+
+        def create_client(service_name, **kwargs):
+            calls["count"] += 1
+            if calls["count"] == 1:
+                raise ClientError({"Error": {"Code": "403"}}, "ListBuckets")
+            real_client = boto3.Session().client(service_name, **kwargs)
+            real_client.list_buckets = MagicMock()
+            return real_client
+
+        mock_boto_client.side_effect = create_client
+
+        client = get_s3_client()
+        url = client.generate_presigned_url(
+            "get_object",
+            Params={"Bucket": "test-bucket", "Key": "report.zip"},
+            ExpiresIn=300,
+        )
+
+        # The credential-less fallback path must also pin SigV4.
+        assert "X-Amz-Algorithm=AWS4-HMAC-SHA256" in url
 
     @patch("tasks.jobs.export.get_s3_client")
     @patch("tasks.jobs.export.base")