You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "rahul-madaan (via GitHub)" <gi...@apache.org> on 2026/09/08 13:58:42 UTC
[PR] Add OAuth2 client credentials auth for OpenLineage HTTP transport [airflow]
rahul-madaan opened a new pull request, #72711:
URL: https://github.com/apache/airflow/pull/72711
Adds OAuth 2.0 client credentials authentication for the OpenLineage HTTP transport.
The OpenLineage Python client ships only static `api_key` authentication (plus an API-key-to-JWT exchange) for HTTP transports, so backends that issue short-lived OAuth 2.0 access tokens could not be used from Airflow without a custom `TokenProvider` on the Python path.
This PR adds:
- `OAuth2ClientCredentialsTokenProvider`, usable as `auth.type` by import path (the client's documented extension mechanism) with `http` and `async_http` transports, including those nested in `composite`. Tokens are cached under a lock and re-requested `tokenRefreshBuffer` seconds before expiry (capped at half the token lifetime). Options accept camelCase and snake_case, so `OPENLINEAGE__TRANSPORT__AUTH__*` environment variables work as well.
- `airflow_connection_oauth2_client_credentials` auth type, resolved like `airflow_connection_api_key` (#66342): client ID from the connection login, client secret from the password, token endpoint from the host.
- Spark transport injection forwards the current access token as `api_key` (as already happens for the client's `jwt` provider) and logs the lifetime caveat; a token error logs a warning instead of failing the Spark task.
Design notes:
- The provider class lives in the Airflow provider because the connection-backed auth type and the Spark injection need it here. A generic version can be upstreamed to openlineage-python later and the connection alias pointed at it.
- No short `oauth2_client_credentials` alias is introduced, so a future native client auth type of that name would not be shadowed by Airflow.
- `time.monotonic` is mocked with `mock.patch` in tests because `time_machine` does not patch it.
- `requests` is imported directly; it is a hard dependency of both `apache-airflow` and `openlineage-python`.
Tested end to end against an OAuth-protected OpenLineage backend (Keycloak client credentials, 600 s tokens): `http`, `async_http` and `composite` transports, connection-backed configuration, a real Dag run, and a soak run confirming the token was refreshed before expiry while the original token was rejected once expired.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Fable 5.1)
Generated-by: Claude Code (Fable 5.1) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01QBfMWG5AG33JMqGGQgAV6v
--
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 OAuth2 client credentials auth for OpenLineage HTTP transport [airflow]
Posted by "rahul-madaan (via GitHub)" <gi...@apache.org>.
rahul-madaan closed pull request #72711: Add OAuth2 client credentials auth for OpenLineage HTTP transport
URL: https://github.com/apache/airflow/pull/72711
--
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 OAuth2 client credentials auth for OpenLineage HTTP transport [airflow]
Posted by "rahul-madaan (via GitHub)" <gi...@apache.org>.
rahul-madaan commented on PR #72711:
URL: https://github.com/apache/airflow/pull/72711#issuecomment-5614091312
closed in favour of https://github.com/OpenLineage/OpenLineage/pull/4935
--
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 OAuth2 client credentials auth for OpenLineage HTTP transport [airflow]
Posted by "rahul-madaan (via GitHub)" <gi...@apache.org>.
rahul-madaan commented on PR #72711:
URL: https://github.com/apache/airflow/pull/72711#issuecomment-5590359432
@kacpermuda @JDarDagran
could you please review this PR? I have tested everything locally.
--
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 OAuth2 client credentials auth for OpenLineage HTTP transport [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on PR #72711:
URL: https://github.com/apache/airflow/pull/72711#issuecomment-5599780530
Hey, it looks like the token provider itself should be implemented in client first, then we can release the client, bump the required client version in OL provider and then implement the part that retrieves oauth details from airflow connection. So if you don't mind, let's start with implementing oauth token provider in OL client
--
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 OAuth2 client credentials auth for OpenLineage HTTP transport [airflow]
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #72711:
URL: https://github.com/apache/airflow/pull/72711#discussion_r3958767188
##########
providers/openlineage/src/airflow/providers/openlineage/token_provider.py:
##########
@@ -32,6 +41,127 @@
"""Raised when OpenLineage config cannot be resolved from an Airflow connection."""
+class OpenLineageOAuth2ConfigError(AirflowException):
+ """Raised when OpenLineage OAuth2 client credentials auth is misconfigured."""
+
+
+class OpenLineageOAuth2TokenError(AirflowException):
+ """Raised when an OAuth2 access token for OpenLineage HTTP transport cannot be obtained."""
+
+
+def _get_config_value(config: dict[str, Any], *keys: str) -> Any:
+ """Return the first non-empty value found under the given keys (camelCase first, then snake_case)."""
+ for key in keys:
+ value = config.get(key)
+ if value is not None and value != "":
+ return value
+ return None
+
+
+def _get_required_config_value(config: dict[str, Any], *keys: str) -> str:
+ value = _get_config_value(config, *keys)
+ if not value:
+ raise OpenLineageOAuth2ConfigError(
+ f"OpenLineage OAuth2 client credentials auth requires a non-empty `{keys[0]}`."
+ )
+ return str(value)
+
+
+class OAuth2ClientCredentialsTokenProvider(TokenProvider):
+ """
+ OpenLineage HTTP transport ``TokenProvider`` using the OAuth 2.0 client credentials grant (RFC 6749, 4.4).
+
+ Access tokens requested from ``tokenEndpoint`` with ``clientId`` and ``clientSecret`` are cached and
+ requested again ``tokenRefreshBuffer`` seconds before they expire. Options are accepted in camelCase and
+ snake_case, as OpenLineage client environment variables are read in snake_case.
+ """
+
+ DEFAULT_TOKEN_REFRESH_BUFFER = 120.0
+ DEFAULT_TOKEN_LIFETIME = 300.0 # used when the token endpoint does not return `expires_in`
+ CLIENT_AUTH_METHODS = ("client_secret_basic", "client_secret_post")
+
+ def __init__(self, config: dict[str, Any]) -> None:
+ super().__init__(config)
+ self.token_endpoint = _get_required_config_value(config, "tokenEndpoint", "token_endpoint")
+ self.client_id = _get_required_config_value(config, "clientId", "client_id")
+ self.client_secret = _get_required_config_value(config, "clientSecret", "client_secret")
+ self.scope = _get_config_value(config, "scope")
+ self.client_auth_method = str(
+ _get_config_value(config, "clientAuthMethod", "client_auth_method") or self.CLIENT_AUTH_METHODS[0]
+ ).lower()
+ if self.client_auth_method not in self.CLIENT_AUTH_METHODS:
+ raise OpenLineageOAuth2ConfigError(
+ f"OpenLineage OAuth2 auth option `clientAuthMethod` must be one of {self.CLIENT_AUTH_METHODS}, "
+ f"got `{self.client_auth_method}`."
+ )
+ token_refresh_buffer = _get_config_value(config, "tokenRefreshBuffer", "token_refresh_buffer")
+ self.token_refresh_buffer = (
+ self.DEFAULT_TOKEN_REFRESH_BUFFER if token_refresh_buffer is None else float(token_refresh_buffer)
+ )
+ self._lock = threading.Lock()
+ self._access_token: str | None = None
+ self._refresh_at = 0.0
+
+ def get_bearer(self) -> str | None:
+ with self._lock:
+ if self._access_token is None or time.monotonic() >= self._refresh_at:
+ self._request_token()
+ return f"Bearer {self._access_token}"
+
+ def _request_token(self) -> None:
+ data = {"grant_type": "client_credentials"}
+ if self.scope:
+ data["scope"] = self.scope
+ basic_auth = None
+ if self.client_auth_method == "client_secret_post":
+ data["client_id"] = self.client_id
+ data["client_secret"] = self.client_secret
+ else:
+ basic_auth = (self.client_id, self.client_secret)
+
+ log.debug(
+ "Requesting OAuth2 access token from `%s` for client `%s`.", self.token_endpoint, self.client_id
Review Comment:
## CodeQL / Clear-text logging of sensitive information
This expression logs [sensitive data (password)](1) as clear text.
[Show more details](https://github.com/apache/airflow/security/code-scanning/644)
##########
providers/openlineage/src/airflow/providers/openlineage/token_provider.py:
##########
@@ -32,6 +41,127 @@
"""Raised when OpenLineage config cannot be resolved from an Airflow connection."""
+class OpenLineageOAuth2ConfigError(AirflowException):
+ """Raised when OpenLineage OAuth2 client credentials auth is misconfigured."""
+
+
+class OpenLineageOAuth2TokenError(AirflowException):
+ """Raised when an OAuth2 access token for OpenLineage HTTP transport cannot be obtained."""
+
+
+def _get_config_value(config: dict[str, Any], *keys: str) -> Any:
+ """Return the first non-empty value found under the given keys (camelCase first, then snake_case)."""
+ for key in keys:
+ value = config.get(key)
+ if value is not None and value != "":
+ return value
+ return None
+
+
+def _get_required_config_value(config: dict[str, Any], *keys: str) -> str:
+ value = _get_config_value(config, *keys)
+ if not value:
+ raise OpenLineageOAuth2ConfigError(
+ f"OpenLineage OAuth2 client credentials auth requires a non-empty `{keys[0]}`."
+ )
+ return str(value)
+
+
+class OAuth2ClientCredentialsTokenProvider(TokenProvider):
+ """
+ OpenLineage HTTP transport ``TokenProvider`` using the OAuth 2.0 client credentials grant (RFC 6749, 4.4).
+
+ Access tokens requested from ``tokenEndpoint`` with ``clientId`` and ``clientSecret`` are cached and
+ requested again ``tokenRefreshBuffer`` seconds before they expire. Options are accepted in camelCase and
+ snake_case, as OpenLineage client environment variables are read in snake_case.
+ """
+
+ DEFAULT_TOKEN_REFRESH_BUFFER = 120.0
+ DEFAULT_TOKEN_LIFETIME = 300.0 # used when the token endpoint does not return `expires_in`
+ CLIENT_AUTH_METHODS = ("client_secret_basic", "client_secret_post")
+
+ def __init__(self, config: dict[str, Any]) -> None:
+ super().__init__(config)
+ self.token_endpoint = _get_required_config_value(config, "tokenEndpoint", "token_endpoint")
+ self.client_id = _get_required_config_value(config, "clientId", "client_id")
+ self.client_secret = _get_required_config_value(config, "clientSecret", "client_secret")
+ self.scope = _get_config_value(config, "scope")
+ self.client_auth_method = str(
+ _get_config_value(config, "clientAuthMethod", "client_auth_method") or self.CLIENT_AUTH_METHODS[0]
+ ).lower()
+ if self.client_auth_method not in self.CLIENT_AUTH_METHODS:
+ raise OpenLineageOAuth2ConfigError(
+ f"OpenLineage OAuth2 auth option `clientAuthMethod` must be one of {self.CLIENT_AUTH_METHODS}, "
+ f"got `{self.client_auth_method}`."
+ )
+ token_refresh_buffer = _get_config_value(config, "tokenRefreshBuffer", "token_refresh_buffer")
+ self.token_refresh_buffer = (
+ self.DEFAULT_TOKEN_REFRESH_BUFFER if token_refresh_buffer is None else float(token_refresh_buffer)
+ )
+ self._lock = threading.Lock()
+ self._access_token: str | None = None
+ self._refresh_at = 0.0
+
+ def get_bearer(self) -> str | None:
+ with self._lock:
+ if self._access_token is None or time.monotonic() >= self._refresh_at:
+ self._request_token()
+ return f"Bearer {self._access_token}"
+
+ def _request_token(self) -> None:
+ data = {"grant_type": "client_credentials"}
+ if self.scope:
+ data["scope"] = self.scope
+ basic_auth = None
+ if self.client_auth_method == "client_secret_post":
+ data["client_id"] = self.client_id
+ data["client_secret"] = self.client_secret
+ else:
+ basic_auth = (self.client_id, self.client_secret)
+
+ log.debug(
+ "Requesting OAuth2 access token from `%s` for client `%s`.", self.token_endpoint, self.client_id
Review Comment:
## CodeQL / Clear-text logging of sensitive information
This expression logs [sensitive data (password)](1) as clear text.
[Show more details](https://github.com/apache/airflow/security/code-scanning/643)
##########
providers/openlineage/src/airflow/providers/openlineage/token_provider.py:
##########
@@ -32,6 +41,127 @@
"""Raised when OpenLineage config cannot be resolved from an Airflow connection."""
+class OpenLineageOAuth2ConfigError(AirflowException):
+ """Raised when OpenLineage OAuth2 client credentials auth is misconfigured."""
+
+
+class OpenLineageOAuth2TokenError(AirflowException):
+ """Raised when an OAuth2 access token for OpenLineage HTTP transport cannot be obtained."""
+
+
+def _get_config_value(config: dict[str, Any], *keys: str) -> Any:
+ """Return the first non-empty value found under the given keys (camelCase first, then snake_case)."""
+ for key in keys:
+ value = config.get(key)
+ if value is not None and value != "":
+ return value
+ return None
+
+
+def _get_required_config_value(config: dict[str, Any], *keys: str) -> str:
+ value = _get_config_value(config, *keys)
+ if not value:
+ raise OpenLineageOAuth2ConfigError(
+ f"OpenLineage OAuth2 client credentials auth requires a non-empty `{keys[0]}`."
+ )
+ return str(value)
+
+
+class OAuth2ClientCredentialsTokenProvider(TokenProvider):
+ """
+ OpenLineage HTTP transport ``TokenProvider`` using the OAuth 2.0 client credentials grant (RFC 6749, 4.4).
+
+ Access tokens requested from ``tokenEndpoint`` with ``clientId`` and ``clientSecret`` are cached and
+ requested again ``tokenRefreshBuffer`` seconds before they expire. Options are accepted in camelCase and
+ snake_case, as OpenLineage client environment variables are read in snake_case.
+ """
+
+ DEFAULT_TOKEN_REFRESH_BUFFER = 120.0
+ DEFAULT_TOKEN_LIFETIME = 300.0 # used when the token endpoint does not return `expires_in`
+ CLIENT_AUTH_METHODS = ("client_secret_basic", "client_secret_post")
+
+ def __init__(self, config: dict[str, Any]) -> None:
+ super().__init__(config)
+ self.token_endpoint = _get_required_config_value(config, "tokenEndpoint", "token_endpoint")
+ self.client_id = _get_required_config_value(config, "clientId", "client_id")
+ self.client_secret = _get_required_config_value(config, "clientSecret", "client_secret")
+ self.scope = _get_config_value(config, "scope")
+ self.client_auth_method = str(
+ _get_config_value(config, "clientAuthMethod", "client_auth_method") or self.CLIENT_AUTH_METHODS[0]
+ ).lower()
+ if self.client_auth_method not in self.CLIENT_AUTH_METHODS:
+ raise OpenLineageOAuth2ConfigError(
+ f"OpenLineage OAuth2 auth option `clientAuthMethod` must be one of {self.CLIENT_AUTH_METHODS}, "
+ f"got `{self.client_auth_method}`."
+ )
+ token_refresh_buffer = _get_config_value(config, "tokenRefreshBuffer", "token_refresh_buffer")
+ self.token_refresh_buffer = (
+ self.DEFAULT_TOKEN_REFRESH_BUFFER if token_refresh_buffer is None else float(token_refresh_buffer)
+ )
+ self._lock = threading.Lock()
+ self._access_token: str | None = None
+ self._refresh_at = 0.0
+
+ def get_bearer(self) -> str | None:
+ with self._lock:
+ if self._access_token is None or time.monotonic() >= self._refresh_at:
+ self._request_token()
+ return f"Bearer {self._access_token}"
+
+ def _request_token(self) -> None:
+ data = {"grant_type": "client_credentials"}
+ if self.scope:
+ data["scope"] = self.scope
+ basic_auth = None
+ if self.client_auth_method == "client_secret_post":
+ data["client_id"] = self.client_id
+ data["client_secret"] = self.client_secret
+ else:
+ basic_auth = (self.client_id, self.client_secret)
+
+ log.debug(
+ "Requesting OAuth2 access token from `%s` for client `%s`.", self.token_endpoint, self.client_id
+ )
+ try:
+ response = requests.post(self.token_endpoint, data=data, auth=basic_auth, timeout=10)
+ response.raise_for_status()
+ except requests.RequestException as e:
+ raise OpenLineageOAuth2TokenError(
+ f"OAuth2 token request to `{self.token_endpoint}` failed: {e}"
+ ) from e
+ try:
+ payload = response.json()
+ access_token = payload["access_token"]
+ except (ValueError, KeyError, TypeError):
+ raise OpenLineageOAuth2TokenError(
+ f"OAuth2 token endpoint `{self.token_endpoint}` did not return an `access_token`."
+ ) from None
+ lifetime = self._get_token_lifetime(payload)
+ # Refresh early, but never so early that every event would trigger a new token request.
+ self._refresh_at = time.monotonic() + lifetime - min(self.token_refresh_buffer, lifetime / 2)
+ self._access_token = str(access_token)
+ log.debug(
+ "Obtained OAuth2 access token for client `%s`, valid for %s seconds.", self.client_id, lifetime
Review Comment:
## CodeQL / Clear-text logging of sensitive information
This expression logs [sensitive data (password)](1) as clear text.
[Show more details](https://github.com/apache/airflow/security/code-scanning/645)
--
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 OAuth2 client credentials auth for OpenLineage HTTP transport [airflow]
Posted by "rahul-madaan (via GitHub)" <gi...@apache.org>.
rahul-madaan commented on code in PR #72711:
URL: https://github.com/apache/airflow/pull/72711#discussion_r3966812944
##########
providers/openlineage/docs/spark.rst:
##########
@@ -181,6 +181,9 @@ as Airflow without manual configuration.
.. caution::
Currently, only HTTP transport is supported for automatic transport injection (with api_key authentication, if configured).
+ If the transport uses :ref:`OAuth 2.0 client credentials authentication <configuration_oauth2:openlineage>`, the current
+ access token is injected as ``api_key`` authentication, because Spark OpenLineage integration cannot refresh it.
+ Spark applications running longer than the token lifetime will fail to emit OpenLineage events.
Review Comment:
Flagging this for review since OAuth tokens might have a limited lifetime and Spark would silently fail to send OL events if we use OAuth.
--
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