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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
