[v3-1-test] fix: always include kid in JWT header for symmetric key tokens (#62883)

airflow-core/src/airflow/api_fastapi/auth/tokens.py

@@ -372,12 +372,13 @@ def _load_key_from_configured_file() -> AllowedPrivateKeys | None:
 
 
 def _generate_kid(gen) -> str:
-    if not gen._private_key:
-        return "not-used"
-
+    # Always check config first — both symmetric and asymmetric keys can have a configured kid
     if kid := _conf_factory("api_auth", "jwt_kid", fallback=None)():
         return kid
 
+    if not gen._private_key:
+        return "not-used"
+
     # Generate it from the thumbprint of the private key
     info = key_to_jwk_dict(gen._private_key)
     return info["kid"]
@@ -457,8 +458,7 @@ def generate(self, extras: dict[str, Any] | None = None, headers: dict[str, Any]
         if extras is not None:
             claims = extras | claims
         headers = {"alg": self.algorithm, **(headers or {})}
-        if self._private_key:
-            headers["kid"] = self.kid
+        headers["kid"] = self.kid
         return jwt.encode(claims, self.signing_arg, algorithm=self.algorithm, headers=headers)
 
 

airflow-core/tests/unit/api_fastapi/auth/test_tokens.py

@@ -126,6 +126,30 @@ def test_with_secret_key():
     assert generator.signing_arg == "abc"
 
 
+def test_secret_key_token_includes_kid_in_header():
+    """Symmetric (secret_key) tokens must include 'kid' in the JWT header so the validator accepts them."""
+    generator = JWTGenerator(secret_key="test-secret", audience="test", valid_for=60)
+    token = generator.generate({"sub": "user"})
+    header = jwt.get_unverified_header(token)
+    assert "kid" in header, "kid must always be present in the JWT header"
+    assert header["kid"] == "not-used"
+
+
+def test_secret_key_with_configured_kid():
+    """When jwt_kid is configured, symmetric key generators should use it."""
+    from unittest.mock import patch
+
+    with patch.dict(
+        "os.environ",
+        {"AIRFLOW__API_AUTH__JWT_KID": "my-custom-kid"},
+    ):
+        generator = JWTGenerator(secret_key="test-secret", audience="test", valid_for=60)
+        assert generator.kid == "my-custom-kid"
+        token = generator.generate({"sub": "user"})
+        header = jwt.get_unverified_header(token)
+        assert header["kid"] == "my-custom-kid"
+
+
 @pytest.fixture
 def jwt_generator(ed25519_private_key: Ed25519PrivateKey):
     key = ed25519_private_key