test(auth): cover validation and concurrent failure paths

Exercise malformed inputs, shared failures, waiter deadlines, and persistent HTTPS connections. Collect fresh-process import coverage through coverage.py's subprocess patch for pytest-cov 7.

Author: bfreiberg

poetry.lock

@@ -87,7 +87,7 @@ kafka-consumer-avro = ["avro"]
 kafka-consumer-protobuf = ["protobuf"]
 
 [tool.poetry.group.dev.dependencies]
-coverage = { extras = ["toml"], version = "^7.6" }
+coverage = { extras = ["toml"], version = "^7.10.6" }
 pytest = ">=8.3.4,<10.0.0"
 boto3 = "^1.26.164"
 isort = ">=5.13.2,<10.0.0"
@@ -148,6 +148,8 @@ omit = [
   "aws_lambda_powertools/metrics/metric.py"  # barrel import (export-only)
 ]
 branch = true
+# pytest-cov 7 delegates subprocess measurement to coverage.py.
+patch = ["subprocess"]
 
 [tool.coverage.html]
 directory = "test_report"

pyproject.toml

@@ -190,3 +190,35 @@ def test_invalid_request_arns_raise_instead_of_returning_an_invalid_policy(jwks,
     event["methodArn"] = arn
     with pytest.raises(ValueError, match="concrete API Gateway"):
         verifier(jwks).authorize(event)
+
+
+@pytest.mark.parametrize("malformed", [False, True])
+@pytest.mark.parametrize("field", ["headers", "multiValueHeaders"])
+def test_authorizer_denies_malformed_or_ambiguous_header_maps(jwks, issue_token, malformed, field):
+    token = "Bearer " + issue_token()
+    value = [token] if field == "multiValueHeaders" else token
+    headers = [("Authorization", value)] if malformed else {"Authorization": value, "authorization": value}
+    event = {"type": "REQUEST", "methodArn": ARN, field: headers}
+    response = verifier(jwks).authorize(event)
+    assert response["principalId"] == "unauthorized"
+    assert response["policyDocument"]["Statement"][0]["Effect"] == "Deny"
+    assert "context" not in response
+
+
+@pytest.mark.parametrize("event", [None, [], {}, {"type": "OTHER"}])
+def test_authorizer_rejects_unsupported_events(jwks, event):
+    with pytest.raises(ValueError, match="TOKEN or REQUEST"):
+        verifier(jwks).authorize(event)
+
+
+@pytest.mark.parametrize(
+    "options,message",
+    [
+        ({"response_format": "unsupported"}, "response_format"),
+        ({"context_claims": ["claims"]}, "claims is reserved"),
+    ],
+)
+def test_authorizer_rejects_invalid_response_configuration(jwks, issue_token, options, message):
+    event = {"type": "TOKEN", "methodArn": ARN, "authorizationToken": "Bearer " + issue_token()}
+    with pytest.raises(ValueError, match=message):
+        verifier(jwks).authorize(event, **options)

tests/functional/auth/test_authorizer.py

@@ -127,7 +127,10 @@ def test_invalid_discovery_never_falls_back_or_fetches_untrusted_keys(http, issu
     assert len(http.requests) == 1
 
 
-@pytest.mark.parametrize("body", [{}, {"keys": None}, {"keys": ["bad-key"]}, b"not json", b"x" * (1024 * 1024 + 1)])
+@pytest.mark.parametrize(
+    "body",
+    [{}, {"keys": None}, {"keys": ["bad-key"]}, [], None, b"not json", b"x" * (1024 * 1024 + 1)],
+)
 def test_malformed_key_sets_fail_closed(http, issue_token, body):
     http.serve(JWKS_URL, body)
 
@@ -213,3 +216,32 @@ def test_failed_unknown_key_refresh_preserves_only_still_fresh_keys(http, jwks,
     clock.advance(299)
     with pytest.raises(JWKSFetchError):
         subject.verify(issue_token())
+
+
+def test_waiting_verifier_timeout_does_not_cancel_the_shared_key_fetch(http, jwks, issue_token):
+    entered = threading.Event()
+    release = threading.Event()
+
+    def fetch():
+        entered.set()
+        assert release.wait(5)
+        return jwks
+
+    http.serve(JWKS_URL, fetch)
+    owner = verifier(jwks_uri=JWKS_URL, timeout_seconds=5)
+    waiter = verifier(jwks_uri=JWKS_URL, timeout_seconds=0.1)
+    token = issue_token()
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        result = executor.submit(owner.verify, token)
+        try:
+            assert entered.wait(5)
+            with pytest.raises(JWKSFetchError) as error:
+                waiter.verify(token)
+            assert error.value.__context__ is None
+            assert not result.done()
+        finally:
+            release.set()
+        assert result.result(timeout=5)["sub"] == "user-123"
+
+    assert waiter.verify(token)["sub"] == "user-123"
+    assert len(http.requests) == 1

tests/functional/auth/test_jwks_cache.py

@@ -230,6 +230,9 @@ def load_secret():
     "options",
     [
         {"audience": "one", "resource": "two"},
+        {"audience": " "},
+        {"resource": ""},
+        {"resource": 42},
         {"token_url": "http://idp.example.com/token"},
         {"token_url": "https://user:secret@idp.example.com/token"},
         {"client_id": ""},
@@ -273,6 +276,7 @@ def test_request_attaches_resource_token_without_forwarding_client_credentials(h
         ("https://api.example.com/orders", {"redirect": True}),
         ("https://api.example.com/orders", {"retries": 3}),
         ("https://api.example.com/orders", {"timeout": 0}),
+        ("https://api.example.com/orders", {"headers": [("Accept", "application/json")]}),
     ],
 )
 def test_request_rejects_unsafe_overrides_before_acquiring_credentials(http, url, options):
@@ -290,3 +294,107 @@ def test_request_does_not_follow_redirects_or_retry_downstream_failures(http):
     http.serve("https://api.example.com/orders", {}, status=503)
     assert subject.request("GET", "https://api.example.com/orders").status == 503
     assert len(http.requests) == 3
+
+
+@pytest.mark.parametrize("method", [None, "", "GET /", "GET\r\nInjected"])
+def test_invalid_http_methods_are_rejected_before_loading_credentials(http, method):
+    calls = []
+
+    def load_secret():
+        calls.append(True)
+        return "test-secret"
+
+    with pytest.raises(ValueError, match="HTTP method"):
+        client(client_secret=load_secret).request(method, "https://api.example.com/orders")
+    assert calls == []
+    assert http.requests == []
+
+
+@pytest.mark.parametrize("secret", [None, "", 42])
+def test_invalid_secret_loader_results_are_rejected_before_sending_credentials(http, secret):
+    with pytest.raises(TokenExchangeError) as error:
+        client(client_secret=lambda: secret).auth_headers()
+    assert error.value.__context__ is None
+    assert http.requests == []
+
+
+def test_retry_stops_when_the_backoff_exceeds_the_remaining_budget(http, clock, monkeypatch):
+    sleeps = []
+    monkeypatch.setattr(time, "sleep", sleeps.append)
+    http.serve(TOKEN_URL, {}, status=503, method="POST")
+
+    with pytest.raises(TokenExchangeError):
+        client(timeout_seconds=0.05).auth_headers()
+    assert sleeps == []
+    assert len(http.requests) == 1
+
+
+def test_waiting_callers_share_a_failed_exchange_and_can_recover(http, monkeypatch):
+    entered = threading.Event()
+    release = threading.Event()
+    joined = threading.Event()
+
+    def exchange():
+        entered.set()
+        assert release.wait(5)
+        return {"error": "invalid_client"}
+
+    http.serve(TOKEN_URL, exchange, status=401, method="POST")
+    subject = client()
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        owner = executor.submit(subject.auth_headers)
+        try:
+            assert entered.wait(5)
+            flight = subject._flight
+            assert flight is not None
+            wait = flight.done.wait
+
+            def observe_wait(timeout):
+                joined.set()
+                return wait(timeout)
+
+            # Keep the real Event; observe it so the provider is released only
+            # after the second caller has joined the active exchange.
+            monkeypatch.setattr(flight.done, "wait", observe_wait)
+            waiter = executor.submit(subject.auth_headers)
+            assert joined.wait(5)
+        finally:
+            release.set()
+        for result in (owner, waiter):
+            with pytest.raises(TokenExchangeError) as error:
+                result.result(timeout=5)
+            assert error.value.__context__ is None
+    assert len(http.requests) == 1
+
+    http.serve(TOKEN_URL, {"access_token": "recovered", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    assert subject.auth_headers() == {"Authorization": "Bearer recovered"}
+    assert len(http.requests) == 2
+
+
+def test_waiting_callers_timeout_without_returning_the_late_token(http):
+    entered = threading.Event()
+    release = threading.Event()
+
+    def exchange():
+        entered.set()
+        assert release.wait(5)
+        return {"access_token": "too-late", "token_type": "Bearer", "expires_in": 100}
+
+    http.serve(TOKEN_URL, exchange, method="POST")
+    subject = client(timeout_seconds=0.1)
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        owner = executor.submit(subject.auth_headers)
+        try:
+            assert entered.wait(5)
+            with pytest.raises(TokenExchangeError):
+                subject.auth_headers()
+            assert not owner.done()
+            assert len(http.requests) == 1
+        finally:
+            release.set()
+        with pytest.raises(TokenExchangeError):
+            owner.result(timeout=5)
+
+    http.serve(TOKEN_URL, {"access_token": "recovered", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    assert subject.auth_headers() == {"Authorization": "Bearer recovered"}
+    assert len(http.requests) == 2

tests/functional/auth/test_oauth2.py

@@ -114,3 +114,31 @@ def test_any_of_rejects_ambiguous_issuer_configuration(jwks):
 
     with pytest.raises(ValueError):
         JWTVerifier.any_of(verifier, verifier)
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        {"user_pool_id": "invalid"},
+        {"user_pool_id": None},
+        {"client_id": " "},
+        {"client_id": None},
+        {"issuer": "https://untrusted.example.com"},
+        {"algorithms": ["HS256"]},
+        {"jwks_uri": "https://untrusted.example.com/keys"},
+    ],
+)
+def test_cognito_rejects_invalid_or_overridden_trust_configuration(options):
+    config = {
+        "user_pool_id": "us-east-1_pool",
+        "client_id": "desktop-client",
+        "audience": "https://api.example.com",
+    }
+    with pytest.raises(ValueError):
+        JWTVerifier.cognito(**{**config, **options})
+
+
+@pytest.mark.parametrize("verifiers", [(), (None,), ("https://idp.example.com",)])
+def test_issuer_groups_require_explicit_verifier_instances(verifiers):
+    with pytest.raises(ValueError):
+        JWTVerifier.any_of(*verifiers)

tests/functional/auth/test_profiles.py

@@ -65,6 +65,7 @@ def test_expired_token_is_rejected(jwks, claims, issue_token):
         ("exp", float("inf")),
         ("exp", float("nan")),
         ("exp", True),
+        ("exp", 10**400),
         ("nbf", "0"),
         ("nbf", 9999999999),
     ],
@@ -88,6 +89,9 @@ def test_invalid_claim_values_are_rejected(jwks, claims, issue_token, claim, val
         ("issuer", "http://idp.example.com"),
         ("issuer", "https://user:secret@idp.example.com"),
         ("issuer", "https://idp.example.com/#fragment"),
+        ("issuer", "https://idp.example.com:invalid"),
+        ("issuer", "https://[invalid"),
+        ("issuer", 42),
         ("audience", ""),
         ("audience", []),
         ("algorithms", []),
@@ -96,6 +100,8 @@ def test_invalid_claim_values_are_rejected(jwks, claims, issue_token, claim, val
         ("algorithms", ["RS256", "HS256"]),
         ("clock_skew_seconds", -1),
         ("clock_skew_seconds", float("inf")),
+        ("clock_skew_seconds", 10**400),
+        ("jwks_uri", "https://idp.example.com/keys"),
         ("required_claims", ""),
     ],
 )
@@ -112,15 +118,18 @@ def test_invalid_verifier_configuration_is_rejected(jwks, option, value):
         JWTVerifier(**options)
 
 
+@pytest.mark.parametrize("issuer_group", [False, True])
 @pytest.mark.parametrize("token", ["", "not-a-jwt", "a.b.c", None, 42])
-def test_malformed_tokens_raise_redacted_errors(jwks, token):
+def test_malformed_tokens_raise_redacted_errors(jwks, token, issuer_group):
     verifier = JWTVerifier(
         issuer="https://idp.example.com/",
         audience="https://api.example.com",
         algorithms=["RS256"],
         jwks=jwks,
     )
 
+    if issuer_group:
+        verifier = JWTVerifier.any_of(verifier)
     with pytest.raises(InvalidTokenError) as error:
         verifier.verify(token)
     assert str(error.value) == "Invalid access token"

tests/functional/auth/test_verifier.py

@@ -70,13 +70,20 @@ def tls_files(tmp_path_factory):
     return certificate_path, key_path
 
 
-@pytest.fixture
-def https_server(tls_files, monkeypatch):
+@pytest.fixture(params=[False, True], ids=["connection-close", "keep-alive"])
+def https_server(tls_files, monkeypatch, request):
     endpoint = LocalHTTPS()
 
     class Handler(BaseHTTPRequestHandler):
         protocol_version = "HTTP/1.1"
 
+        def handle(self):
+            try:
+                super().handle()
+            except (ConnectionResetError, ssl.SSLEOFError):
+                # The client may reject a response without draining its body.
+                self.close_connection = True
+
         def do_GET(self):  # noqa: N802
             self.respond()
 
@@ -90,7 +97,7 @@ def respond(self):
             self.send_response(reply.status)
             self.send_header("Content-Type", "application/json")
             self.send_header("Content-Length", str(len(reply.body)))
-            self.send_header("Connection", "close")
+            self.send_header("Connection", "keep-alive" if request.param else "close")
             for name, value in reply.headers.items():
                 self.send_header(name, value)
             self.end_headers()
@@ -109,7 +116,7 @@ def respond(self):
                 # Timeout and oversized-body tests deliberately close early.
                 pass
             finally:
-                self.close_connection = True
+                self.close_connection = not request.param
 
         def log_message(self, format, *args):  # noqa: A002
             pass