fix: widen verify_broker_token except clause to jwt.PyJWTError

src/krb5_token_service/identity.py

@@ -156,7 +156,13 @@ async def verify_broker_token(token: str, settings: Settings) -> dict[str, Any]:
                     "require": ["exp", "iat", "sub"],
                 },
             )
-    except jwt.InvalidTokenError as exc:
+    except jwt.PyJWTError as exc:
+        # PyJWTError (not just InvalidTokenError) so that
+        # RSAAlgorithm.from_jwk's InvalidKeyError — raised for a malformed
+        # or non-RSA JWKS entry — is classified as a 401 like any other
+        # verification failure, rather than escaping uncaught as an
+        # unaudited 500. InvalidKeyError is a PyJWTError but not an
+        # InvalidTokenError.
         error = exc
     except (ValueError, KeyError) as exc:
         error = exc

tests/test_identity.py

@@ -75,6 +75,28 @@ async def test_unknown_kid_is_401(
             await identity.verify_broker_token(token, settings)
         assert excinfo.value.status_code == 401
 
+    async def test_malformed_jwks_key_is_401(
+        self,
+        make_token: Callable[..., str],
+        settings: Settings,
+        stub_jwks_fetch: JwksFetchStub,
+    ) -> None:
+        # A JWKS entry with no n/e is what a broker key-rotation glitch (or
+        # a stray non-RSA key) can publish. RSAAlgorithm.from_jwk raises
+        # jwt.InvalidKeyError for it, which must be classified as a 401
+        # like any other verification failure, not escape as an unhandled
+        # 500. Reassign .keys rather than appending in place — jwks is
+        # session-scoped, so mutating it here would leak into other tests.
+        malformed_kid = "malformed-key"
+        stub_jwks_fetch.keys = [
+            *stub_jwks_fetch.keys,
+            {"kid": malformed_kid, "kty": "RSA", "use": "sig"},
+        ]
+        token = make_token(kid=malformed_kid)
+        with pytest.raises(HTTPException) as excinfo:
+            await identity.verify_broker_token(token, settings)
+        assert excinfo.value.status_code == 401
+
     @pytest.mark.parametrize("claim", ["exp", "iat", "sub"])
     async def test_missing_required_claim_is_401(
         self, make_token: Callable[..., str], settings: Settings, claim: str