You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "robertpofuk (via GitHub)" <gi...@apache.org> on 2026/08/29 16:12:29 UTC
[PR] Add OIDC worker token verification to Edge3 [airflow]
robertpofuk opened a new pull request, #72262:
URL: https://github.com/apache/airflow/pull/72262
## Why
The Edge worker API currently only verifies worker tokens signed with the shared `[api_auth] jwt_secret` (symmetric HS512). This adds **optional** verification against a trusted OpenID Connect provider, so edge workers can authenticate with tokens minted by an external identity provider instead of the shared secret.
## What
When `[edge] oidc_jwks_url` is set, worker tokens are verified against the provider JWKS using asymmetric signatures, checking the `iss` and (optionally) `aud` claims. New `[edge]` options:
- `oidc_jwks_url` — provider JWKS URL; enables OIDC verification when set
- `oidc_issuer` — expected `iss` claim
- `oidc_audience` — expected `aud` claim; leave empty to skip audience verification (for tokens that carry no `aud`)
- `oidc_algorithms` — accepted signing algorithms, comma-separated (default `RS256`)
When `oidc_jwks_url` is empty (the default), the existing shared-secret verification is used unchanged, so existing deployments are unaffected. The per-request signed `method` claim check is skipped in OIDC mode because externally issued tokens do not carry it.
Uses the existing core `JWKS` / `JWTValidator`; no new dependencies. Adds unit tests for validator selection, the config options, and the method-claim behavior (the module had no test module before).
## Known limitation / follow-up
Some identity providers publish JWKS keys without an `alg` field. PyJWT then defaults such keys to `RS256` and rejects tokens signed with another algorithm even when `algorithms` is passed to `jwt.decode`. That looks like it should be addressed in core `JWTValidator` rather than here, so it is intentionally out of scope for this PR. Happy to follow up separately if maintainers agree.
---
^ Add meaningful description above
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines) for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3922402215
##########
providers/edge3/provider.yaml:
##########
@@ -106,6 +106,47 @@ config:
type: string
example: https://airflow.hosting.org/edge_worker/v1/rpcapi
default: ~
+ oidc_jwks_url:
Review Comment:
Done! Adjusted naming
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3980304006
##########
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:
Done — `trusted_jwks_url` now notes it drops the per-method check, `jwt_issuer`
notes that empty skips the issuer check (and needs a `jwt_verifier`), and the
`jwt_audience` wording is corrected. Also fixed the old `oidc_*` names in the PR
description.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5463467354
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our [Contributors' Guide](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example Dag that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5470513595
Fixed CI failures and rebased on main.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3958250738
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,21 +31,64 @@
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 _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()),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=conf.getint("api_auth", "jwt_leeway", fallback=30),
Review Comment:
If we do edge3.jwt_leeway then i would not fallback. Its is then configuration cross reference. This one would be worst one. I think if I open edge3 config and I see there jwt_leeway and other JWT settings in one place it is fine and clear. So current impl with edge3.jwt_leeway would i think fit the best and be clean and isolated and most fleksible.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5681286383
Added 2 new types: _OidcAuthenticator and WorkerTokenContext following same logic. This will add for now "method" as param but we can in future add more info theoretically neede (i.e. queryParams or whatever and extend), so both intput and oput are externdable.
Made validator async so should also be future proof.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4091852923
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -52,7 +53,15 @@ class WorkerTokenAuthorization(TypedDict, total=False):
authorized: bool
-def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+@dataclass(frozen=True)
+class WorkerTokenContext:
+ """Request context passed to a ``[edge] jwt_verifier`` alongside the token claims."""
Review Comment:
Why not pass the full FastAPI Request object? My (quick) read of Jarek's comment led me to expect the full Request object to be passed down.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4016214535
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,25 +31,146 @@
InvalidSignatureError,
)
-from airflow.api_fastapi.auth.tokens import JWTValidator
-from airflow.providers.common.compat.sdk import conf
+from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
log = logging.getLogger(__name__)
+class WorkerTokenAuthorization(TypedDict, total=False):
+ """
+ Result of authorizing an OIDC worker token beyond signature verification.
+
+ Returned by a ``[edge] jwt_verifier`` callable to answer "may this token act
+ as an edge worker?". ``authorized`` must be ``True`` for the request to
+ proceed; a falsy result (or a raised exception) rejects it.
+ """
+
+ authorized: bool
+
+
+def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+ """Authorize any token that passed signature, issuer and audience verification."""
+ return {"authorized": True}
+
+
+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 configured audience, or None to accept only tokens without an ``aud`` claim."""
+ return conf.get("edge", "jwt_audience", fallback="") or None
+
+
+def _jwt_issuer() -> str | None:
+ """Return the expected issuer, or None to skip issuer verification when left empty."""
+ return conf.get("edge", "jwt_issuer", 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 _jwt_verifier() -> Callable[[dict], WorkerTokenAuthorization | None]:
+ """Return the configured worker-authorization callable, or the permissive default."""
+ return conf.getimport("edge", "jwt_verifier", fallback=None) or _default_jwt_verifier
+
+
+def _jwt_verifier_configured() -> bool:
+ """Return whether an explicit ``[edge] jwt_verifier`` is set."""
+ return bool(conf.get("edge", "jwt_verifier", fallback=""))
+
+
@cache
-def jwt_validator() -> JWTValidator:
+def _oidc_enabled() -> bool:
+ """
+ Return whether OIDC verification is enabled, decided once and cached.
+
+ The validator is also cached, so the mode must be read from a single place;
+ otherwise a request-time re-read could disagree with the cached validator and
+ skip the ``method``-claim check for a shared-secret token.
+ """
+ return bool(_trusted_jwks_url())
+
+
+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.
+
+ Rejects the configuration when issuer verification is skipped (empty
+ ``jwt_issuer``) without a ``jwt_verifier``: that combination would accept any
+ token signed by a key in the JWKS. The validator is built lazily on the first
+ request, so this surfaces as a rejected request (403) rather than a startup
+ failure.
+ """
+ if not _jwt_issuer() and not _jwt_verifier_configured():
+ raise AirflowConfigException(
+ "[edge] jwt_verifier must be set when trusted_jwks_url is configured "
+ "without jwt_issuer, otherwise any token signed by the JWKS is accepted."
+ )
+ return JWTValidator(
+ jwks=JWKS(url=jwks_url),
+ issuer=_jwt_issuer(),
+ audience=_jwt_audience(),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=_jwt_leeway(),
+ )
+
+
+@cache
+def jwt_validator() -> JWTValidator:
+ if _oidc_enabled():
+ return _oidc_validator(_trusted_jwks_url())
+ return _shared_secret_validator()
+
+
def jwt_validate(authorization: str) -> dict:
return jwt_validator().validated_claims(authorization)
+def _check_worker_authorization(payload: dict) -> None:
Review Comment:
this i would then extend with "WorkerTokenContext" param, I can imagine people in future maybe wanting more request information passed in. this way its both in and out extendable
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3980292732
##########
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:
Addressed the issuer bit differently on purpose: my IdP doesn't provide `iss` or
other identifiable info, so I bind identity in a custom `jwt_verifier` instead. When
`jwt_issuer` is empty a `jwt_verifier` is required, so some authorization is always
there. Happy to add more fixed checks, but I'd keep `jwt_verifier`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3957723553
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,21 +31,64 @@
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 _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()),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=conf.getint("api_auth", "jwt_leeway", fallback=30),
Review Comment:
I'm not saying either one is right or wrong.
Maybe we want `conf.getint("edge3", "jwt_leeway", fallback=conf.getint("api_auth", "jwt_leeway", fallback=30),`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3980125261
##########
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:
Addressed this bit differently. My IDP does not provide iss or other idenfiable info so in my vendored impl i did custom fuction so I didn't make `jwt_issuer` strictly required — it can
be empty *if* you set a `jwt_verifier`. Some IdPs are per-deployment and binding on
`sub`/`scope` in the verifier is more useful than on `iss`. I can add other fixed checks but would keep jwt_verifier. When iss jwt_issuer is not defined jwt_verifier is required so some kind of authorzataion must ber provided
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4014102509
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,25 +31,146 @@
InvalidSignatureError,
)
-from airflow.api_fastapi.auth.tokens import JWTValidator
-from airflow.providers.common.compat.sdk import conf
+from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
log = logging.getLogger(__name__)
+class WorkerTokenAuthorization(TypedDict, total=False):
+ """
+ Result of authorizing an OIDC worker token beyond signature verification.
+
+ Returned by a ``[edge] jwt_verifier`` callable to answer "may this token act
+ as an edge worker?". ``authorized`` must be ``True`` for the request to
+ proceed; a falsy result (or a raised exception) rejects it.
+ """
+
+ authorized: bool
+
+
+def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+ """Authorize any token that passed signature, issuer and audience verification."""
+ return {"authorized": True}
+
+
+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 configured audience, or None to accept only tokens without an ``aud`` claim."""
+ return conf.get("edge", "jwt_audience", fallback="") or None
+
+
+def _jwt_issuer() -> str | None:
+ """Return the expected issuer, or None to skip issuer verification when left empty."""
+ return conf.get("edge", "jwt_issuer", 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 _jwt_verifier() -> Callable[[dict], WorkerTokenAuthorization | None]:
Review Comment:
Nit: this resolves the import path on every request, while everything around it is cached.
`conf.getimport` is a config read plus an `import_string` per call, and `_check_worker_authorization` runs on every worker request — job polling and heartbeats included. `_oidc_enabled()` and `jwt_validator()` are both `@cache`d; this is the one piece of OIDC config re-read each time.
It also leaves a gap in the fail-closed check. `_oidc_validator` only asks whether the option is non-empty, so a typo'd path passes construction and then fails per-request — `test_empty_issuer_with_verifier_is_allowed` configures `my_company.edge_auth.verify_worker_token`, which doesn't exist, and the test passes. The operator sees a generic 403 with a fresh error id on every request rather than one clear error when the validator is built. (It does fail *closed* — `getimport` raises `AirflowConfigException` and the catch-all in `jwt_token_authorization` turns it into a 403 — so this is about diagnosability and cost, not safety.)
Resolving the callable inside `_oidc_validator` and storing it next to the validator fixes both: one import at construction time, and a bad path surfaces immediately with `getimport`'s own message naming the key and section.
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,25 +31,146 @@
InvalidSignatureError,
)
-from airflow.api_fastapi.auth.tokens import JWTValidator
-from airflow.providers.common.compat.sdk import conf
+from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
log = logging.getLogger(__name__)
+class WorkerTokenAuthorization(TypedDict, total=False):
+ """
+ Result of authorizing an OIDC worker token beyond signature verification.
+
+ Returned by a ``[edge] jwt_verifier`` callable to answer "may this token act
+ as an edge worker?". ``authorized`` must be ``True`` for the request to
+ proceed; a falsy result (or a raised exception) rejects it.
+ """
+
+ authorized: bool
+
+
+def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+ """Authorize any token that passed signature, issuer and audience verification."""
+ return {"authorized": True}
+
+
+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 configured audience, or None to accept only tokens without an ``aud`` claim."""
+ return conf.get("edge", "jwt_audience", fallback="") or None
+
+
+def _jwt_issuer() -> str | None:
+ """Return the expected issuer, or None to skip issuer verification when left empty."""
+ return conf.get("edge", "jwt_issuer", 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 _jwt_verifier() -> Callable[[dict], WorkerTokenAuthorization | None]:
+ """Return the configured worker-authorization callable, or the permissive default."""
+ return conf.getimport("edge", "jwt_verifier", fallback=None) or _default_jwt_verifier
+
+
+def _jwt_verifier_configured() -> bool:
+ """Return whether an explicit ``[edge] jwt_verifier`` is set."""
+ return bool(conf.get("edge", "jwt_verifier", fallback=""))
+
+
@cache
-def jwt_validator() -> JWTValidator:
+def _oidc_enabled() -> bool:
+ """
+ Return whether OIDC verification is enabled, decided once and cached.
+
+ The validator is also cached, so the mode must be read from a single place;
+ otherwise a request-time re-read could disagree with the cached validator and
+ skip the ``method``-claim check for a shared-secret token.
+ """
+ return bool(_trusted_jwks_url())
+
+
+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.
+
+ Rejects the configuration when issuer verification is skipped (empty
+ ``jwt_issuer``) without a ``jwt_verifier``: that combination would accept any
+ token signed by a key in the JWKS. The validator is built lazily on the first
+ request, so this surfaces as a rejected request (403) rather than a startup
+ failure.
+ """
+ if not _jwt_issuer() and not _jwt_verifier_configured():
+ raise AirflowConfigException(
+ "[edge] jwt_verifier must be set when trusted_jwks_url is configured "
+ "without jwt_issuer, otherwise any token signed by the JWKS is accepted."
+ )
+ return JWTValidator(
+ jwks=JWKS(url=jwks_url),
+ issuer=_jwt_issuer(),
+ audience=_jwt_audience(),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=_jwt_leeway(),
+ )
+
+
+@cache
+def jwt_validator() -> JWTValidator:
+ if _oidc_enabled():
+ return _oidc_validator(_trusted_jwks_url())
+ return _shared_secret_validator()
+
+
def jwt_validate(authorization: str) -> dict:
return jwt_validator().validated_claims(authorization)
Review Comment:
Nit, but worth doing before this meets a real IdP: OIDC mode makes this sync `validated_claims()` do network I/O, and every other JWKS caller in the tree uses the async variant.
`validated_claims()` is `async_to_sync(avalidated_claims)`. On the shared-secret path that's free — `_get_validation_key` returns the secret without awaiting anything. With `jwks=` it now awaits `httpx.AsyncClient.get` against the provider, on the `JWKS` object held by the `@cache`d validator, from whatever event loop `async_to_sync` creates. Starlette runs sync dependencies through `anyio.to_thread`, which asgiref doesn't recognise as its own `sync_to_async` context, so it builds a fresh loop per call rather than reusing one. A single long-lived `httpx.AsyncClient` used across different event loops is the usual way to get "Event loop is closed" / "attached to a different loop".
It won't show up in testing, and it won't show up on day one either. `_should_fetch_jwks()` is true on first use and then roughly hourly, so the first fetch succeeds on whichever loop got there first and the problem only appears at a later refresh. And it fails quietly: `_fetch_remote_jwks` catches `Exception`, logs, and returns `None`, so `fetch_jwks` leaves `_jwks` at the stale keyset. Workers keep authenticating on cached keys until the IdP rotates, and then every worker starts getting 403s from `get_key(kid)` raising `KeyError`, with the actual cause sitting hours earlier in the log as "Failed to fetch remote JWKS".
The rest of the codebase avoids this by awaiting:
- `airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py:160`
- `airflow-core/src/airflow/api_fastapi/execution_api/security.py:126`
- `airflow-core/src/airflow/api_fastapi/execution_api/app.py:149`
- `airflow-core/src/airflow/utils/serve_logs/log_server.py:67`
`jwt_token_authorization_rest` is only ever used as a FastAPI `Depends(...)` (`routes/worker.py`, `routes/jobs.py`, `routes/logs.py`), and `jwt_token_authorization` has no caller outside this module — so making both `async def` and awaiting `avalidated_claims` here stays contained to this file.
##########
providers/edge3/tests/unit/edge3/worker_api/test_auth.py:
##########
@@ -150,3 +155,220 @@ def test_strips_edge_worker_v1_prefix_and_falls_back_to_full_path(
jwt_token_authorization_rest(request, authorization="some-token")
mock_jwt_token_authorization.assert_called_once_with(expected_method, "some-token")
+
+
+class TestOidcConfig:
+ """Config readers translate raw ``[edge]`` options into validator inputs."""
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, ["RS256"], id="default-rs256"),
+ pytest.param("RS512", ["RS512"], id="single-algorithm"),
+ pytest.param("RS256, RS512", ["RS256", "RS512"], id="comma-list-is-split-and-stripped"),
+ ],
+ )
+ def test_oidc_algorithms_parsing(self, configured, expected):
+ """Unset falls back to RS256; a configured value is split on commas and stripped."""
+ overrides = {} if configured is None else {("edge", "jwt_algorithm"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_algorithms() == expected
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, None, id="empty-is-none-accepts-only-aud-less-tokens"),
+ pytest.param("api", "api", id="configured-value-passed-through"),
+ ],
+ )
+ def test_oidc_audience_parsing(self, configured, expected):
+ """An empty audience becomes ``None`` (accept only aud-less tokens); a set value is forwarded."""
+ overrides = {} if configured is None else {("edge", "jwt_audience"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_audience() == expected
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, None, id="empty-is-none-skips-issuer-check"),
+ pytest.param(
+ "https://idp.example.com", "https://idp.example.com", id="configured-passed-through"
+ ),
+ ],
+ )
+ def test_oidc_issuer_parsing(self, configured, expected):
+ """An empty issuer becomes ``None`` (skip issuer check); a set value is forwarded verbatim."""
+ overrides = {} if configured is None else {("edge", "jwt_issuer"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_issuer() == expected
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, 30, id="default-30"),
+ pytest.param("90", 90, id="configured-value-parsed-as-int"),
+ ],
+ )
+ def test_oidc_leeway_parsing(self, configured, expected):
+ """Unset falls back to 30 seconds; a configured value is read as an integer."""
+ overrides = {} if configured is None else {("edge", "jwt_leeway"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_leeway() == expected
+
+
+class TestJwtValidatorSelection:
+ """``jwt_validator`` picks shared-secret vs OIDC based on ``trusted_jwks_url``."""
+
+ @conf_vars({("api_auth", "jwt_secret"): "secret"})
+ def test_uses_shared_secret_validator_when_oidc_jwks_url_unset(self):
+ """Default path is unchanged: a shared-secret validator with no JWKS is built."""
+ validator = auth.jwt_validator()
+
+ assert validator.secret_key == "secret"
+
+ assert validator.jwks is None
+
+ @conf_vars(
+ {
+ ("edge", "trusted_jwks_url"): OIDC_JWKS_URL,
+ ("edge", "jwt_issuer"): "https://idp.example.com",
+ ("edge", "jwt_algorithm"): "RS512",
+ ("edge", "jwt_leeway"): "90",
+ }
+ )
+ def test_uses_oidc_validator_when_jwks_url_set(self):
+ """Setting ``trusted_jwks_url`` builds a JWKS-backed validator wired from ``[edge]`` config."""
+ validator = auth.jwt_validator()
+
+ assert validator.jwks is not None
+
+ assert validator.algorithm == ["RS512"]
+
+ assert validator.issuer == "https://idp.example.com"
+
+ assert validator.audience is None
+
+ assert validator.leeway == 90
+
+ @conf_vars({("edge", "trusted_jwks_url"): OIDC_JWKS_URL})
+ def test_empty_issuer_without_verifier_fails_closed(self):
+ """Skipping issuer verification without a ``jwt_verifier`` is a hard config error."""
+ with pytest.raises(AirflowConfigException, match="jwt_verifier"):
+ auth.jwt_validator()
+
+ @conf_vars(
+ {
+ ("edge", "trusted_jwks_url"): OIDC_JWKS_URL,
+ ("edge", "jwt_verifier"): "my_company.edge_auth.verify_worker_token",
+ }
+ )
+ def test_empty_issuer_with_verifier_is_allowed(self):
+ """A configured ``jwt_verifier`` permits skipping issuer verification."""
+ validator = auth.jwt_validator()
+
+ assert validator.jwks is not None
+
+ assert validator.issuer is None
+
+
+class TestMethodClaimCheck:
+ """``_check_method_claim`` enforces the signed ``method`` only for shared-secret tokens."""
+
+ @conf_vars({("api_auth", "jwt_secret"): "secret"})
+ def test_shared_secret_rejects_mismatched_method(self):
+ """A shared-secret token minted for another endpoint is forbidden (403)."""
+ with pytest.raises(HTTPException) as exc_info:
+ auth._check_method_claim("worker/register", {"method": "worker/other"})
+
+ assert exc_info.value.status_code == 403
+
+ @conf_vars({("api_auth", "jwt_secret"): "secret"})
+ def test_shared_secret_accepts_matching_method(self):
+ """A shared-secret token whose ``method`` matches the request passes without raising."""
+ auth._check_method_claim("worker/register", {"method": "worker/register"})
+
+ @conf_vars({("edge", "trusted_jwks_url"): OIDC_JWKS_URL})
+ def test_oidc_skips_method_claim(self):
+ """OIDC tokens carry no ``method`` claim, so the check is skipped rather than 403."""
+ auth._check_method_claim("worker/register", {})
+
+
+def _signed_token(audience: str = "") -> str:
+ """Mint a real HS512 token, omitting ``aud`` when passed empty."""
+ generator = JWTGenerator(secret_key=JWT_SECRET, valid_for=300, audience=audience)
+ return generator.generate()
+
+
+class TestOidcAudienceVerification:
+ """
+ Pin the real audience semantics against a signed token.
+
+ ``jwt_audience`` empty means ``audience=None``, which PyJWT accepts only for
+ tokens that carry no ``aud`` claim and rejects for tokens that do. A configured
+ value requires a matching ``aud``.
+ """
+
+ def test_empty_audience_accepts_aud_less_token(self):
+ with conf_vars({("edge", "trusted_jwks_url"): OIDC_JWKS_URL}):
+ validator = JWTValidator(secret_key=JWT_SECRET, audience=auth._jwt_audience(), leeway=5)
Review Comment:
Nit on what this class covers: it pins PyJWT's audience handling rather than the OIDC validator this PR builds.
The validator here is hand-constructed with `secret_key=JWT_SECRET` — an HS512 shared secret — so `_oidc_validator()` never runs, and the only PR code in the path is `auth._jwt_audience()`, which `TestOidcConfig.test_oidc_audience_parsing` already covers. The `conf_vars({("edge", "trusted_jwks_url"): OIDC_JWKS_URL})` wrapper is inert: nothing inside the `with` block reads it, since `_jwt_audience()` doesn't depend on the mode. That's the part I'd change regardless — it reads as though OIDC mode is under test, which invites the next person to trust it for more than it covers.
Taken together with `TestWorkerAuthorization` mocking `_jwt_verifier` (so `conf.getimport` never runs), nothing in the suite drives a token through the JWKS path: key resolution by `kid`, the `algorithm` restriction, and issuer binding on a real signed token are all uncovered.
There's a working recipe in the tree at `airflow-core/tests/unit/api_fastapi/auth/test_tokens.py:205-208` — `JWKS.from_private_key((key, "kid1"))` builds an in-memory keyset, so a test can mint a token with `JWTGenerator(private_key=...)` and drive `jwt_token_authorization` end to end with no network. That would let these three audience cases assert against the real `_oidc_validator()` instead of a stand-in.
Being straight about the limits: `from_private_key` sets `url=os.devnull`, so such a test would *not* catch the event-loop issue on the other thread — it closes the semantics gap, not that one.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3919019091
##########
providers/edge3/provider.yaml:
##########
@@ -106,6 +106,47 @@ config:
type: string
example: https://airflow.hosting.org/edge_worker/v1/rpcapi
default: ~
+ oidc_jwks_url:
Review Comment:
This is probably worth being consistent in the namign of the key names (but keep in the edge section) with `trusted_jwks_url` and `jwt_issuer` etc that we have for the api_auth section.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5464857468
Tested on our internal environment by vendoring edge3 and re-building. Works nicely and makes our setup super simple since workers just order therir access via central IDP.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4014286282
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,25 +31,146 @@
InvalidSignatureError,
)
-from airflow.api_fastapi.auth.tokens import JWTValidator
-from airflow.providers.common.compat.sdk import conf
+from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
log = logging.getLogger(__name__)
+class WorkerTokenAuthorization(TypedDict, total=False):
+ """
+ Result of authorizing an OIDC worker token beyond signature verification.
+
+ Returned by a ``[edge] jwt_verifier`` callable to answer "may this token act
+ as an edge worker?". ``authorized`` must be ``True`` for the request to
+ proceed; a falsy result (or a raised exception) rejects it.
+ """
+
+ authorized: bool
+
+
+def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+ """Authorize any token that passed signature, issuer and audience verification."""
+ return {"authorized": True}
+
+
+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 configured audience, or None to accept only tokens without an ``aud`` claim."""
+ return conf.get("edge", "jwt_audience", fallback="") or None
+
+
+def _jwt_issuer() -> str | None:
+ """Return the expected issuer, or None to skip issuer verification when left empty."""
+ return conf.get("edge", "jwt_issuer", 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 _jwt_verifier() -> Callable[[dict], WorkerTokenAuthorization | None]:
+ """Return the configured worker-authorization callable, or the permissive default."""
+ return conf.getimport("edge", "jwt_verifier", fallback=None) or _default_jwt_verifier
+
+
+def _jwt_verifier_configured() -> bool:
+ """Return whether an explicit ``[edge] jwt_verifier`` is set."""
+ return bool(conf.get("edge", "jwt_verifier", fallback=""))
+
+
@cache
-def jwt_validator() -> JWTValidator:
+def _oidc_enabled() -> bool:
+ """
+ Return whether OIDC verification is enabled, decided once and cached.
+
+ The validator is also cached, so the mode must be read from a single place;
+ otherwise a request-time re-read could disagree with the cached validator and
+ skip the ``method``-claim check for a shared-secret token.
+ """
+ return bool(_trusted_jwks_url())
+
+
+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.
+
+ Rejects the configuration when issuer verification is skipped (empty
+ ``jwt_issuer``) without a ``jwt_verifier``: that combination would accept any
+ token signed by a key in the JWKS. The validator is built lazily on the first
+ request, so this surfaces as a rejected request (403) rather than a startup
+ failure.
+ """
+ if not _jwt_issuer() and not _jwt_verifier_configured():
+ raise AirflowConfigException(
+ "[edge] jwt_verifier must be set when trusted_jwks_url is configured "
+ "without jwt_issuer, otherwise any token signed by the JWKS is accepted."
+ )
+ return JWTValidator(
+ jwks=JWKS(url=jwks_url),
+ issuer=_jwt_issuer(),
+ audience=_jwt_audience(),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=_jwt_leeway(),
+ )
+
+
+@cache
+def jwt_validator() -> JWTValidator:
+ if _oidc_enabled():
+ return _oidc_validator(_trusted_jwks_url())
+ return _shared_secret_validator()
+
+
def jwt_validate(authorization: str) -> dict:
return jwt_validator().validated_claims(authorization)
+def _check_worker_authorization(payload: dict) -> None:
Review Comment:
Nit, and explicitly *not* a request to validate anything: consider passing `method` through to the verifier now, unused, so the signature doesn't have to change later.
You future-proofed the *output* of this hook on purpose — returning a mapping rather than a bool so `teams` and friends can be added without breaking callers. The same argument applies to the input, and right now it's one-way: the verifier receives `payload` only, so it can never see which endpoint the token is being presented to. `jwt_token_authorization` has `method` in hand two lines up.
Why that's worth more than it looks: `method` is the request path after `/edge_worker/v1/`, and the worker routes are `@worker_router.post("/{worker_name}")` — so the path carries the worker name. An operator who wants their verifier to ask "does this token's `sub` match the worker it's addressing?" currently can't. That's the scoping the shared-secret path gets from its `method` claim and OIDC mode gives up. Handing the verifier the path doesn't restore it — whether to enforce anything stays the operator's call, in their own callable — it just makes it possible.
The timing is the actual argument. `jwt_verifier` is new in 4.4.0 and unreleased, so nothing implements `Callable[[dict], ...]` against it yet. Once deployments do, widening it is a breaking change for every one of them. Something like:
```python
def _default_jwt_verifier(claims: dict, method: str) -> WorkerTokenAuthorization:
return {"authorized": True}
```
with `_check_worker_authorization(method, payload)` forwarding it, costs nothing today and is awkward to retrofit. Worth a line in `jwt_verifier`'s `provider.yaml` description too — it currently says the callable "receives the validated claims (``dict``)".
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3980125261
##########
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:
Addressed this bit differently. My IDP does not provide iss or other idenfiable info so in my vendored impl i did custom fuction so I didn't make `jwt_issuer` strictly required — it can
be empty *if* you set a `jwt_verifier`. Some IdPs are per-deployment and binding on
`sub`/`scope` in the verifier is more useful than on `iss`. I can add other fixed checks but would keep jwt_verifier. When iss jwt_issuer is not defined jwt_verifier is required so some kind of authorzataion must ber provided
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3956292570
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,21 +31,64 @@
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 _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()),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=conf.getint("api_auth", "jwt_leeway", fallback=30),
Review Comment:
Should this be api_auth or edge section?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4016321320
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,25 +31,146 @@
InvalidSignatureError,
)
-from airflow.api_fastapi.auth.tokens import JWTValidator
-from airflow.providers.common.compat.sdk import conf
+from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
log = logging.getLogger(__name__)
+class WorkerTokenAuthorization(TypedDict, total=False):
+ """
+ Result of authorizing an OIDC worker token beyond signature verification.
+
+ Returned by a ``[edge] jwt_verifier`` callable to answer "may this token act
+ as an edge worker?". ``authorized`` must be ``True`` for the request to
+ proceed; a falsy result (or a raised exception) rejects it.
+ """
+
+ authorized: bool
+
+
+def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+ """Authorize any token that passed signature, issuer and audience verification."""
+ return {"authorized": True}
+
+
+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 configured audience, or None to accept only tokens without an ``aud`` claim."""
+ return conf.get("edge", "jwt_audience", fallback="") or None
+
+
+def _jwt_issuer() -> str | None:
+ """Return the expected issuer, or None to skip issuer verification when left empty."""
+ return conf.get("edge", "jwt_issuer", 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 _jwt_verifier() -> Callable[[dict], WorkerTokenAuthorization | None]:
+ """Return the configured worker-authorization callable, or the permissive default."""
+ return conf.getimport("edge", "jwt_verifier", fallback=None) or _default_jwt_verifier
+
+
+def _jwt_verifier_configured() -> bool:
+ """Return whether an explicit ``[edge] jwt_verifier`` is set."""
+ return bool(conf.get("edge", "jwt_verifier", fallback=""))
+
+
@cache
-def jwt_validator() -> JWTValidator:
+def _oidc_enabled() -> bool:
+ """
+ Return whether OIDC verification is enabled, decided once and cached.
+
+ The validator is also cached, so the mode must be read from a single place;
+ otherwise a request-time re-read could disagree with the cached validator and
+ skip the ``method``-claim check for a shared-secret token.
+ """
+ return bool(_trusted_jwks_url())
+
+
+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.
+
+ Rejects the configuration when issuer verification is skipped (empty
+ ``jwt_issuer``) without a ``jwt_verifier``: that combination would accept any
+ token signed by a key in the JWKS. The validator is built lazily on the first
+ request, so this surfaces as a rejected request (403) rather than a startup
+ failure.
+ """
+ if not _jwt_issuer() and not _jwt_verifier_configured():
+ raise AirflowConfigException(
+ "[edge] jwt_verifier must be set when trusted_jwks_url is configured "
+ "without jwt_issuer, otherwise any token signed by the JWKS is accepted."
+ )
+ return JWTValidator(
+ jwks=JWKS(url=jwks_url),
+ issuer=_jwt_issuer(),
+ audience=_jwt_audience(),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=_jwt_leeway(),
+ )
+
+
+@cache
+def jwt_validator() -> JWTValidator:
+ if _oidc_enabled():
+ return _oidc_validator(_trusted_jwks_url())
+ return _shared_secret_validator()
+
+
def jwt_validate(authorization: str) -> dict:
return jwt_validator().validated_claims(authorization)
Review Comment:
Validator is now invoked async.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5681304816
Added it as new commit so changes are clearly visible
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4016205637
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,25 +31,146 @@
InvalidSignatureError,
)
-from airflow.api_fastapi.auth.tokens import JWTValidator
-from airflow.providers.common.compat.sdk import conf
+from airflow.api_fastapi.auth.tokens import JWKS, JWTValidator
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
log = logging.getLogger(__name__)
+class WorkerTokenAuthorization(TypedDict, total=False):
+ """
+ Result of authorizing an OIDC worker token beyond signature verification.
+
+ Returned by a ``[edge] jwt_verifier`` callable to answer "may this token act
+ as an edge worker?". ``authorized`` must be ``True`` for the request to
+ proceed; a falsy result (or a raised exception) rejects it.
+ """
+
+ authorized: bool
+
+
+def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+ """Authorize any token that passed signature, issuer and audience verification."""
+ return {"authorized": True}
+
+
+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 configured audience, or None to accept only tokens without an ``aud`` claim."""
+ return conf.get("edge", "jwt_audience", fallback="") or None
+
+
+def _jwt_issuer() -> str | None:
+ """Return the expected issuer, or None to skip issuer verification when left empty."""
+ return conf.get("edge", "jwt_issuer", 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 _jwt_verifier() -> Callable[[dict], WorkerTokenAuthorization | None]:
Review Comment:
Adjusted by adding _OidcAuthenticator to hold everything needed for verification
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb merged PR #72262:
URL: https://github.com/apache/airflow/pull/72262
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4092628075
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -52,7 +53,15 @@ class WorkerTokenAuthorization(TypedDict, total=False):
authorized: bool
-def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+@dataclass(frozen=True)
+class WorkerTokenContext:
+ """Request context passed to a ``[edge] jwt_verifier`` alongside the token claims."""
Review Comment:
Alright, that sounds sensible.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3957626991
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -30,21 +31,64 @@
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 _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()),
+ algorithm=_jwt_algorithms(),
+ required_claims=frozenset({"iat", "exp"}),
+ leeway=conf.getint("api_auth", "jwt_leeway", fallback=30),
Review Comment:
Thank you for feedback. I adjusted to use edge. It was kinda tricky becase on one side I want to have on place to define this. but on other side token for worker might have much different lifetime then normal token so using edge makes more sense.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r3980300551
##########
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:
Fixed as you suggested — cast gone, widened `JWTValidator.audience` to
`str | Sequence[str] | None` (it already forwarded `None` to `jwt.decode`). Docs
now match: empty audience only accepts tokens without `aud`, a set value requires a
match. Added signed-token tests for both.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4091874443
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -52,7 +53,15 @@ class WorkerTokenAuthorization(TypedDict, total=False):
authorized: bool
-def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+@dataclass(frozen=True)
+class WorkerTokenContext:
+ """Request context passed to a ``[edge] jwt_verifier`` alongside the token claims."""
Review Comment:
He did say just method, but given this:
> the worker routes are @worker_router.post("/{worker_name}") — so the path carries the worker name. An operator who wants their verifier to ask "does this token's sub match the worker it's addressing?" currently can't.
I'd say passing the whole Request object is better, and means we don't need a new type either.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5813461164
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5620684117
Adjusted implementation to ensure that at least some assertions are required for checking weather token is allowed to be user when oidc is enabled.
jwt_verifier concept is inspured by [core].hostname_callable so should fit overall style
jwt_verifier is on purpose built to return dict back. This way future requirements can be added. For exmaple core Airflow hasfunction to get_teams (i think in executor impelemtations). So We can use this return as place to extend and return as from jwt_verifier {authorized: true teams [x, y]}. Much cleaner to have compatible extension if we return dict then fixed values.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4016199013
##########
providers/edge3/tests/unit/edge3/worker_api/test_auth.py:
##########
@@ -150,3 +155,220 @@ def test_strips_edge_worker_v1_prefix_and_falls_back_to_full_path(
jwt_token_authorization_rest(request, authorization="some-token")
mock_jwt_token_authorization.assert_called_once_with(expected_method, "some-token")
+
+
+class TestOidcConfig:
+ """Config readers translate raw ``[edge]`` options into validator inputs."""
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, ["RS256"], id="default-rs256"),
+ pytest.param("RS512", ["RS512"], id="single-algorithm"),
+ pytest.param("RS256, RS512", ["RS256", "RS512"], id="comma-list-is-split-and-stripped"),
+ ],
+ )
+ def test_oidc_algorithms_parsing(self, configured, expected):
+ """Unset falls back to RS256; a configured value is split on commas and stripped."""
+ overrides = {} if configured is None else {("edge", "jwt_algorithm"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_algorithms() == expected
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, None, id="empty-is-none-accepts-only-aud-less-tokens"),
+ pytest.param("api", "api", id="configured-value-passed-through"),
+ ],
+ )
+ def test_oidc_audience_parsing(self, configured, expected):
+ """An empty audience becomes ``None`` (accept only aud-less tokens); a set value is forwarded."""
+ overrides = {} if configured is None else {("edge", "jwt_audience"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_audience() == expected
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, None, id="empty-is-none-skips-issuer-check"),
+ pytest.param(
+ "https://idp.example.com", "https://idp.example.com", id="configured-passed-through"
+ ),
+ ],
+ )
+ def test_oidc_issuer_parsing(self, configured, expected):
+ """An empty issuer becomes ``None`` (skip issuer check); a set value is forwarded verbatim."""
+ overrides = {} if configured is None else {("edge", "jwt_issuer"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_issuer() == expected
+
+ @pytest.mark.parametrize(
+ ("configured", "expected"),
+ [
+ pytest.param(None, 30, id="default-30"),
+ pytest.param("90", 90, id="configured-value-parsed-as-int"),
+ ],
+ )
+ def test_oidc_leeway_parsing(self, configured, expected):
+ """Unset falls back to 30 seconds; a configured value is read as an integer."""
+ overrides = {} if configured is None else {("edge", "jwt_leeway"): configured}
+ with conf_vars(overrides):
+ assert auth._jwt_leeway() == expected
+
+
+class TestJwtValidatorSelection:
+ """``jwt_validator`` picks shared-secret vs OIDC based on ``trusted_jwks_url``."""
+
+ @conf_vars({("api_auth", "jwt_secret"): "secret"})
+ def test_uses_shared_secret_validator_when_oidc_jwks_url_unset(self):
+ """Default path is unchanged: a shared-secret validator with no JWKS is built."""
+ validator = auth.jwt_validator()
+
+ assert validator.secret_key == "secret"
+
+ assert validator.jwks is None
+
+ @conf_vars(
+ {
+ ("edge", "trusted_jwks_url"): OIDC_JWKS_URL,
+ ("edge", "jwt_issuer"): "https://idp.example.com",
+ ("edge", "jwt_algorithm"): "RS512",
+ ("edge", "jwt_leeway"): "90",
+ }
+ )
+ def test_uses_oidc_validator_when_jwks_url_set(self):
+ """Setting ``trusted_jwks_url`` builds a JWKS-backed validator wired from ``[edge]`` config."""
+ validator = auth.jwt_validator()
+
+ assert validator.jwks is not None
+
+ assert validator.algorithm == ["RS512"]
+
+ assert validator.issuer == "https://idp.example.com"
+
+ assert validator.audience is None
+
+ assert validator.leeway == 90
+
+ @conf_vars({("edge", "trusted_jwks_url"): OIDC_JWKS_URL})
+ def test_empty_issuer_without_verifier_fails_closed(self):
+ """Skipping issuer verification without a ``jwt_verifier`` is a hard config error."""
+ with pytest.raises(AirflowConfigException, match="jwt_verifier"):
+ auth.jwt_validator()
+
+ @conf_vars(
+ {
+ ("edge", "trusted_jwks_url"): OIDC_JWKS_URL,
+ ("edge", "jwt_verifier"): "my_company.edge_auth.verify_worker_token",
+ }
+ )
+ def test_empty_issuer_with_verifier_is_allowed(self):
+ """A configured ``jwt_verifier`` permits skipping issuer verification."""
+ validator = auth.jwt_validator()
+
+ assert validator.jwks is not None
+
+ assert validator.issuer is None
+
+
+class TestMethodClaimCheck:
+ """``_check_method_claim`` enforces the signed ``method`` only for shared-secret tokens."""
+
+ @conf_vars({("api_auth", "jwt_secret"): "secret"})
+ def test_shared_secret_rejects_mismatched_method(self):
+ """A shared-secret token minted for another endpoint is forbidden (403)."""
+ with pytest.raises(HTTPException) as exc_info:
+ auth._check_method_claim("worker/register", {"method": "worker/other"})
+
+ assert exc_info.value.status_code == 403
+
+ @conf_vars({("api_auth", "jwt_secret"): "secret"})
+ def test_shared_secret_accepts_matching_method(self):
+ """A shared-secret token whose ``method`` matches the request passes without raising."""
+ auth._check_method_claim("worker/register", {"method": "worker/register"})
+
+ @conf_vars({("edge", "trusted_jwks_url"): OIDC_JWKS_URL})
+ def test_oidc_skips_method_claim(self):
+ """OIDC tokens carry no ``method`` claim, so the check is skipped rather than 403."""
+ auth._check_method_claim("worker/register", {})
+
+
+def _signed_token(audience: str = "") -> str:
+ """Mint a real HS512 token, omitting ``aud`` when passed empty."""
+ generator = JWTGenerator(secret_key=JWT_SECRET, valid_for=300, audience=audience)
+ return generator.generate()
+
+
+class TestOidcAudienceVerification:
+ """
+ Pin the real audience semantics against a signed token.
+
+ ``jwt_audience`` empty means ``audience=None``, which PyJWT accepts only for
+ tokens that carry no ``aud`` claim and rejects for tokens that do. A configured
+ value requires a matching ``aud``.
+ """
+
+ def test_empty_audience_accepts_aud_less_token(self):
+ with conf_vars({("edge", "trusted_jwks_url"): OIDC_JWKS_URL}):
+ validator = JWTValidator(secret_key=JWT_SECRET, audience=auth._jwt_audience(), leeway=5)
Review Comment:
Adjusted tests for this case
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "robertpofuk (via GitHub)" <gi...@apache.org>.
robertpofuk commented on code in PR #72262:
URL: https://github.com/apache/airflow/pull/72262#discussion_r4092231332
##########
providers/edge3/src/airflow/providers/edge3/worker_api/auth.py:
##########
@@ -52,7 +53,15 @@ class WorkerTokenAuthorization(TypedDict, total=False):
authorized: bool
-def _default_jwt_verifier(claims: dict) -> WorkerTokenAuthorization:
+@dataclass(frozen=True)
+class WorkerTokenContext:
+ """Request context passed to a ``[edge] jwt_verifier`` alongside the token claims."""
Review Comment:
Good point on the worker-addressing use case. I kept WorkerTokenContext minimal for the first cut — expose the least now, extend when there's a concrete requirement — and I'd rather grow the context with request field than pass the raw Request as param. Request isn't the only thing we'll want to hand the verifier (e.g. whether teams are enabled), so it'd end up as an attribute of the context anyway — and passing it whole leaks Starlette into the soon-to-be-public verifier contract.
To be clear on scope: this function does not hand off JWT verification and it never should because we would lower the security barrier. Signature/issuer/audience validation stays in core (JWTValidator against trusted_jwks_url, with jwt_issuer/jwt_audience); jwt_verifier only answers the narrower "may this validated identity act as a worker?" question. method here is the existing concept for the API path after /edge_worker/v1/ (same one _check_method_claim uses), documented on the field.
I would still keep claims as first params because i think most of use cases will be simple token validation and this will keep most of implementations clean and simple.
Regarding worker name i would for now stick with methd. method is pre-existing vocabulary in this module and I can gladely in future extend context with more attributes that would improve usubility. In this PR i would keep it to minimum API surface towards verifier.
On that note, Happy to go either way — both solve the immediate need.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OIDC worker token verification to Edge3 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #72262:
URL: https://github.com/apache/airflow/pull/72262#issuecomment-5812511356
I've done a final rebase to re-run tests against latest main then we'll merge.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org