dlactin commented on code in PR #72978:
URL: https://github.com/apache/airflow/pull/72978#discussion_r3992914153
##########
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:
I was going to add separate error handling for connectivity errors, but
realized the other grant methods also just return 403s.
If that's something that makes sense to add to the airflow auth manager, I'm
happy to submit another PR for all grant types, or just leave it as-is.
--
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]
