potiuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3978917452
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,21 +31,69 @@
InvalidSignatureError,
)
-from airflow.api_fastapi.auth.tokens import JWTValidator
+from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
from airflow.providers.common.compat.sdk import conf
log = logging.getLogger(__name__)
-@cache
-def jwt_validator() -> JWTValidator:
+def _trusted_jwks_url() -> str:
+ """Return the configured trusted JWKS URL, or an empty string when
unset."""
+ return conf.get("edge", "trusted_jwks_url", fallback="") or ""
+
+
+def _jwt_algorithms() -> list[str]:
+ """Return the accepted signing algorithms for OIDC worker tokens."""
+ configured = conf.get("edge", "jwt_algorithm", fallback="RS256") or "RS256"
+ return [algorithm.strip() for algorithm in configured.split(",") if
algorithm.strip()]
+
+
+def _jwt_audience() -> str | None:
+ """Return the expected token audience, or None to skip audience
verification."""
+ return conf.get("edge", "jwt_audience", fallback="") or None
+
+
+def _jwt_leeway() -> int:
+ """Return the clock-skew leeway (seconds) for OIDC worker tokens."""
+ return conf.getint("edge", "jwt_leeway", fallback=30)
+
+
+def _shared_secret_validator() -> JWTValidator:
+ """Build a validator for worker tokens signed with the shared ``[api_auth]
jwt_secret``."""
return JWTValidator(
secret_key=conf.get("api_auth", "jwt_secret"),
leeway=conf.getint("api_auth", "jwt_leeway", fallback=30),
audience="api",
)
+def _oidc_validator(jwks_url: str) -> JWTValidator:
+ """
+ Build a validator for worker tokens issued by a trusted OIDC provider.
+
+ Verifies the token signature against the provider JWKS and checks the
+ ``iss`` and (optionally) ``aud`` claims. Used when ``[edge]
trusted_jwks_url``
+ is configured, so workers can authenticate with tokens minted by an
+ external identity provider instead of the shared secret.
+ """
+ return JWTValidator(
+ jwks=JWKS(url=jwks_url),
+ issuer=conf.get("edge", "jwt_issuer", fallback=None),
+ audience=cast("str", _jwt_audience()),
Review Comment:
These two lines advertise behaviour the code does not implement, and the
result is a 403 for tokens that should work.
`_jwt_audience()` returns `None` when `[edge] jwt_audience` is empty, and
the `cast("str", ...)` here asserts non-`None` to satisfy
`JWTValidator.audience: str | Sequence[str]` — precisely where `None` is the
intended, load-bearing value. `JWTValidator` passes it straight to `jwt.decode`
and leaves PyJWT's `verify_aud` enabled, and PyJWT does this (`api_jwt.py:520`):
```python
if audience is None:
if "aud" not in payload or not payload["aud"]:
return
# Application did not specify an audience, but
# the token has the 'aud' claim
raise InvalidAudienceError("Invalid audience")
```
So an empty `jwt_audience` does not skip audience verification. It requires
the token to carry **no** `aud` at all, and rejects any token that has one. The
`provider.yaml` help says "Leave empty to skip audience verification, which is
required for provider tokens that do not carry an ``aud`` claim" — the second
half is right, the first half says the opposite of what happens. An operator
who reads that and points a normal IdP-issued token at it gets a 403 and an
anonymised error id with no indication that `aud` was the reason.
Two coherent ways out:
- **Preferred:** require a configured `jwt_audience` for tokens that carry
`aud`, and reword the option to say an empty value accepts only audience-less
tokens. That matches what the code does and keeps the audience bound.
- If genuinely skipping verification is wanted, implement it explicitly
(disable `verify_aud`) rather than relying on `audience=None`, and say so in
the docs — though note that removing the audience check widens the same trust
boundary discussed on the `_check_method_claim` thread.
Either way, please drop the `cast`. If `None` is a legitimate value here,
the honest fix is widening the core annotation to `str | Sequence[str] | None`;
a cast that contradicts the value being passed hides exactly the behaviour that
bites above.
Worth calling out that the current tests cannot catch this: they assert
validator attributes rather than validating a signed token, so an audience
rejection never surfaces. A pair of signed-token tests — with and without
`aud`, under both an empty and a configured `jwt_audience` — would pin
whichever semantics you choose.
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -59,17 +108,31 @@ def _forbidden_response(message: str):
)
+def _check_method_claim(method: str, payload: dict) -> None:
+ """
+ Verify the signed ``method`` claim for shared-secret tokens.
+
+ Tokens minted by the Edge API carry the request ``method`` they are valid
+ for. Tokens issued by an external OIDC provider do not, so the check is
+ skipped when OIDC verification is enabled.
+ """
+ if _trusted_jwks_url():
Review Comment:
This early return is the heart of my concern with the design.
The signed `method` claim is currently the only thing that scopes a worker
token: a token minted for one endpoint cannot be replayed against another. OIDC
mode removes that check and substitutes nothing — no subject check, no client
check, no scope check.
What remains for an OIDC-mode request is: the token was signed by a key in
the configured JWKS, and it carries `iat` and `exp`. That establishes **who
signed the token**, not **that this identity is permitted to act as an edge
worker**. Those are different questions, and only the first is being asked.
The consequence, if the issuer serves any other workload — which is the
normal case for a corporate IdP shared across services — is that an unrelated
identity holding an `aud`-less token from that issuer passes this function.
`jwt_token_authorization` is the sole authentication dependency on the worker
routes, so that identity can register workers, fetch queued jobs, and change
job states. That is strictly broader access than a holder of the existing
shared secret has, which inverts the usual expectation that adding an IdP
tightens authentication.
It compounds with the issuer default. `[edge] jwt_issuer` has no default, so
the minimal documented configuration — set `trusted_jwks_url`, leave the rest
alone — passes `issuer=None`, and PyJWT short-circuits on that
(`api_jwt.py:566`):
```python
def _validate_iss(self, payload, issuer) -> None:
if issuer is None:
return
```
So in its simplest supported form this binds neither the issuer nor the
identity.
What I'd like to see before this lands:
1. An explicit worker-authorization policy — a validated `sub`, client id,
or scope claim that the deployment declares and this code enforces. Whatever
shape it takes, something has to answer "may this identity act as a worker?"
2. `jwt_issuer` required when `trusted_jwks_url` is set, failing closed at
validator construction rather than silently skipping the check.
3. A test proving a token signed by the trusted issuer but belonging to an
unrelated identity is rejected. That test is the real specification here.
Separately, and much smaller: "is OIDC enabled?" is answered in two places
with different lifetimes. `jwt_validator()` is `@cache`d, so the mode is fixed
at first use, while this function re-reads config on every request. They can
disagree — a validator cached in shared-secret mode plus a config that now
reports a JWKS URL yields shared-secret validation with the method check
disabled, i.e. a shared-secret token accepted for every method. Narrow, and
your own cache-stickiness test documents the asymmetry, but a single cached
source of truth costs nothing on an auth path.
For transparency: this PR got two independent reviews, and both landed on
this function. The framing above — that a trusted signature is not
authorization — is the second reviewer's, and I think it is the more accurate
way to describe the gap than my own initial reading, which treated it as a weak
default.
##########
providers/edge3/provider.yaml:
##########
@@ -106,6 +106,57 @@ config:
type: string
example: https://airflow.hosting.org/edge_worker/v1/rpcapi
default: ~
+ trusted_jwks_url:
+ description: |
+ When set, edge worker tokens are verified against this JSON Web Key
Set
+ (JWKS) URL of a trusted OpenID Connect provider, using asymmetric
+ signatures, instead of the shared ``[api_auth] jwt_secret``. This
lets
+ edge workers authenticate with tokens minted by an external identity
+ provider.
+
+ When empty (the default) the shared-secret (symmetric) verification
is
+ used, so existing deployments are unaffected.
+ version_added: 4.4.0
+ type: string
+ example: https://idp.example.com/oauth2/v1/keys
+ default: ~
+ jwt_issuer:
Review Comment:
The five new options are documented clearly one by one, but the config
reference is where a Deployment Manager decides whether to turn this on, and it
is missing the two things that change their threat model.
1. **Enabling `trusted_jwks_url` disables per-method token scoping.** The
signed `method` claim is what stops a worker token minted for one endpoint
being replayed against another, and OIDC mode skips it. That is currently
stated only in the PR description, which nobody reads six months from now while
editing `airflow.cfg`.
2. **This option being unset means the issuer is not verified at all.** The
description says "Expected ``iss`` claim of edge worker tokens when
``trusted_jwks_url`` is set", which reads as though the check always happens
once OIDC is on. With no value configured, `issuer=None` reaches PyJWT and
`_validate_iss` returns without checking anything — so the option is silently
load-bearing.
If the outcome of the other threads is that `jwt_issuer` becomes required
whenever `trusted_jwks_url` is set, point 2 resolves itself and the description
just needs to say it is mandatory in OIDC mode. Point 1 needs a sentence either
way — ideally on `trusted_jwks_url` itself, since that is the option someone
reads first when deciding to enable the feature.
While you are in here: `jwt_audience`'s "Leave empty to skip audience
verification" needs the same correction as the code thread — an empty value
does not skip the check, it requires the token to have no ``aud``.
--
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]
