"""Tests for SigV4 signing of internal .NET API Function URL calls (v1 PR1). The API Lambda Function URL uses authType=AWS_IAM, so the workload Lambdas must SigV4-sign their requests or every call 403s. The X-Internal-Api-Key app-layer header must be preserved alongside the transport-layer signature. Verified on the suggestions Lambda as the representative implementation — all four workload Lambdas (suggestions, pdf-extract, pdf-generate, library-ingest) share the same _sign_request_headers / _retry_request code. """ import importlib.util import json import os import sys from unittest.mock import MagicMock, patch _dir = os.path.join(os.path.dirname(__file__), "..", "suggestions") _spec = importlib.util.spec_from_file_location( "suggestions_app_signing", os.path.join(_dir, "app.py") ) app = importlib.util.module_from_spec(_spec) sys.modules["suggestions_app_signing"] = app _spec.loader.exec_module(app) def _ok_response(): resp = MagicMock() resp.status_code = 200 return resp class TestInternalApiSigning: def test_post_request_is_sigv4_signed(self): with patch.object( app.httpx, "request", return_value=_ok_response() ) as mock_req: app._retry_request( "POST", "https://fn.lambda-url.us-east-1.on.aws/api/proposals/x/line-items", json={"items": [1, 2, 3]}, headers={ "X-Internal-Api-Key": "secret-123", "Content-Type": "application/json", }, ) assert mock_req.call_count == 1 _, kwargs = mock_req.call_args headers = kwargs["headers"] # Transport-layer SigV4 auth present assert headers["Authorization"].startswith("AWS4-HMAC-SHA256") assert "X-Amz-Date" in headers # App-layer key preserved alongside the signature assert headers["X-Internal-Api-Key"] == "secret-123" # Body sent as the exact signed bytes (not re-serialized via json=) assert kwargs["content"] == json.dumps({"items": [1, 2, 3]}).encode() assert "json" not in kwargs def test_session_token_included_when_present(self): # conftest provides AWS_SESSION_TOKEN -> SigV4 must add the security-token header with patch.object( app.httpx, "request", return_value=_ok_response() ) as mock_req: app._retry_request( "POST", "https://fn.lambda-url.us-east-1.on.aws/x", json={"a": 1} ) _, kwargs = mock_req.call_args assert "X-Amz-Security-Token" in kwargs["headers"] def test_unsigned_fallback_when_no_credentials(self): with ( patch.object(app._boto_session, "get_credentials", return_value=None), patch.object(app.httpx, "request", return_value=_ok_response()) as mock_req, ): app._retry_request( "POST", "https://fn.lambda-url.us-east-1.on.aws/x", json={"a": 1}, headers={"X-Internal-Api-Key": "secret-123"}, ) _, kwargs = mock_req.call_args # No AWS creds (local/dev) -> no signature, but app-layer key + body still sent assert "Authorization" not in kwargs["headers"] assert kwargs["headers"]["X-Internal-Api-Key"] == "secret-123" assert kwargs["content"] == json.dumps({"a": 1}).encode() def test_bodyless_request_signs_empty_payload(self): with patch.object( app.httpx, "request", return_value=_ok_response() ) as mock_req: app._retry_request( "GET", "https://fn.lambda-url.us-east-1.on.aws/api/proposals/x", headers={"X-Internal-Api-Key": "secret-123"}, ) _, kwargs = mock_req.call_args assert kwargs["content"] == b"" assert kwargs["headers"]["Authorization"].startswith("AWS4-HMAC-SHA256")