vincbeck commented on code in PR #72978:
URL: https://github.com/apache/airflow/pull/72978#discussion_r3992150702
##########
providers/keycloak/src/airflow/providers/keycloak/auth_manager/services/token.py:
##########
@@ -61,6 +66,83 @@ def create_token_for(
return get_auth_manager().generate_api_jwt(user,
expiration_time_in_seconds=expiration_time_in_seconds)
+def create_jwt_federated_token(
+ assertion: str,
+ expiration_time_in_seconds: int = conf.getint("api_auth",
"jwt_expiration_time"),
+) -> str:
+ """
+ Create a token from a Keycloak access token obtained outside of Airflow.
+
+ This authentication flow accepts an access token issued by Keycloak
through any
+ Keycloak-native mechanism (e.g. a "Signed JWT - Federated" client bound to
an
+ external OIDC identity provider such as a Kubernetes ServiceAccount
issuer, or AWS
+ IAM outbound identity federation). Airflow never contacts Keycloak itself
here; it
+ only verifies a token that was already issued, so the caller must have
obtained it
+ directly from Keycloak's token endpoint.
+
+ The token's signature, issuer, and audience are verified against this
realm's JWKS.
+ The ``aud`` claim (a string or a list) must include this Airflow client's
id, which
+ requires an Audience mapper on the federated client's scope in Keycloak.
The calling
+ client (``azp``) must also appear in the ``jwt_federated_client_ids``
allow-list
+ below -- an ``aud`` match alone only proves the token was meant for
Airflow, not
+ that the issuing client has been vetted for machine auth.
+ """
+ realm = conf.get(CONF_SECTION_NAME, CONF_REALM_KEY)
+ server_url = conf.get(CONF_SECTION_NAME, CONF_SERVER_URL_KEY)
+ client_id = conf.get(CONF_SECTION_NAME, CONF_CLIENT_ID_KEY)
+ issuer = f"{server_url.rstrip('/')}/realms/{realm}"
+
+ try:
+ jwks_client = PyJWKClient(f"{issuer}/protocol/openid-connect/certs")
+ signing_key = jwks_client.get_signing_key_from_jwt(assertion)
+ claims = jwt.decode(
+ assertion,
+ signing_key.key,
+ algorithms=["RS256"],
+ audience=client_id,
+ issuer=issuer,
+ )
+ except jwt.PyJWTError:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="Invalid Keycloak assertion",
+ )
+
+ allowed_client_ids = {
+ allowed.strip()
+ for allowed in conf.get(CONF_SECTION_NAME,
CONF_JWT_FEDERATED_CLIENT_IDS_KEY, fallback="").split(",")
+ if allowed.strip()
+ }
+ federated_client_id = claims.get("azp") or claims.get("client_id")
+ if not federated_client_id or federated_client_id not in
allowed_client_ids:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="Invalid Keycloak assertion",
+ )
+
+ # Confirm the assertion is still live (not revoked) and fetch the same
shape of
+ # user info create_client_credentials_token uses, rather than trusting the
JWT's
+ # own claims alone.
+ client = KeycloakAuthManager.get_keycloak_client()
+ try:
+ userinfo_raw: dict | bytes = client.userinfo(assertion)
+ except KeycloakAuthenticationError:
Review Comment:
`python-keycloak` only maps HTTP 401 to `KeycloakAuthenticationError`;
everything else raises `KeycloakGetError`, a sibling class. The PR's own docs
call out the "assertion without openid scope → bare 403" case — that escapes
the handler and returns 500 + traceback instead of 403. The new test hides it
by mocking `side_effect = KeycloakAuthenticationError()`.
Catch `KeycloakError`.
##########
providers/keycloak/src/airflow/providers/keycloak/auth_manager/services/token.py:
##########
@@ -61,6 +66,83 @@ def create_token_for(
return get_auth_manager().generate_api_jwt(user,
expiration_time_in_seconds=expiration_time_in_seconds)
+def create_jwt_federated_token(
+ assertion: str,
+ expiration_time_in_seconds: int = conf.getint("api_auth",
"jwt_expiration_time"),
+) -> str:
+ """
+ Create a token from a Keycloak access token obtained outside of Airflow.
+
+ This authentication flow accepts an access token issued by Keycloak
through any
+ Keycloak-native mechanism (e.g. a "Signed JWT - Federated" client bound to
an
+ external OIDC identity provider such as a Kubernetes ServiceAccount
issuer, or AWS
+ IAM outbound identity federation). Airflow never contacts Keycloak itself
here; it
+ only verifies a token that was already issued, so the caller must have
obtained it
+ directly from Keycloak's token endpoint.
+
+ The token's signature, issuer, and audience are verified against this
realm's JWKS.
+ The ``aud`` claim (a string or a list) must include this Airflow client's
id, which
+ requires an Audience mapper on the federated client's scope in Keycloak.
The calling
+ client (``azp``) must also appear in the ``jwt_federated_client_ids``
allow-list
+ below -- an ``aud`` match alone only proves the token was meant for
Airflow, not
+ that the issuing client has been vetted for machine auth.
+ """
+ realm = conf.get(CONF_SECTION_NAME, CONF_REALM_KEY)
+ server_url = conf.get(CONF_SECTION_NAME, CONF_SERVER_URL_KEY)
+ client_id = conf.get(CONF_SECTION_NAME, CONF_CLIENT_ID_KEY)
+ issuer = f"{server_url.rstrip('/')}/realms/{realm}"
+
+ try:
+ jwks_client = PyJWKClient(f"{issuer}/protocol/openid-connect/certs")
Review Comment:
We might want to cache that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
