potiuk commented on code in PR #60953:
URL: https://github.com/apache/airflow/pull/60953#discussion_r3678041149
##########
providers/hashicorp/src/airflow/providers/hashicorp/_internal_client/vault_client.py:
##########
@@ -16,8 +16,14 @@
# under the License.
from __future__ import annotations
+import fcntl
Review Comment:
`import fcntl` is top-level, so this module no longer imports at all on
Windows — not just when caching is enabled. The docs added in this PR note
Windows is unsupported for the feature, but this makes it unsupported for the
whole provider.
Please move the import inside the caching code paths, or guard it.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
providers/hashicorp/src/airflow/providers/hashicorp/_internal_client/vault_client.py:
##########
@@ -200,6 +218,102 @@ def __init__(
self.jwt_role = jwt_role
self.jwt_token = jwt_token
self.jwt_token_path = jwt_token_path
+ self.cache_approle_token = cache_approle_token
+ self._cached_token_expiry: float | None = None
+
+ def _get_cache_file_path(self) -> Path:
+ """
+ Get the path to the token cache file.
+
+ Uses a deterministic path based on the vault URL and role_id to ensure
+ the same cache is used across different processes.
+
+ :return: Path to the cache file
+ """
+ cache_dir = Path(tempfile.gettempdir()) / CACHE_DIR_NAME
+ try:
+ cache_dir.mkdir(mode=CACHE_DIR_PERMISSIONS, exist_ok=True)
Review Comment:
`mkdir(mode=..., exist_ok=True)` **does not set the mode on an existing
directory** — the `0o700` only applies when this call actually creates it.
If `/tmp/airflow_vault_cache` already exists, including pre-created by
another local user with permissive modes, it is reused as-is and we write a
live Vault token into it. Worth an explicit `os.chmod` plus an ownership check,
or a location that isn't world-writable.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
providers/hashicorp/src/airflow/providers/hashicorp/_internal_client/vault_client.py:
##########
@@ -200,6 +218,102 @@ def __init__(
self.jwt_role = jwt_role
self.jwt_token = jwt_token
self.jwt_token_path = jwt_token_path
+ self.cache_approle_token = cache_approle_token
+ self._cached_token_expiry: float | None = None
+
+ def _get_cache_file_path(self) -> Path:
+ """
+ Get the path to the token cache file.
+
+ Uses a deterministic path based on the vault URL and role_id to ensure
+ the same cache is used across different processes.
+
+ :return: Path to the cache file
+ """
+ cache_dir = Path(tempfile.gettempdir()) / CACHE_DIR_NAME
+ try:
+ cache_dir.mkdir(mode=CACHE_DIR_PERMISSIONS, exist_ok=True)
+ except OSError as e:
+ self.log.warning("Failed to create cache directory %s: %s",
cache_dir, e)
+ # Fallback to a cache file in temp directory root if we can't
create the subdirectory
+ return Path(tempfile.gettempdir()) /
f"vault_token_{self.role_id}.json"
+
+ # Create a unique filename based on vault URL and role_id
+ cache_key = f"{self.url}_{self.role_id}".encode()
+ cache_hash = hashlib.sha256(cache_key).hexdigest()[:16]
+ return cache_dir / f"vault_token_{cache_hash}.json"
+
+ def _read_cached_token(self) -> tuple[str | None, float | None]:
+ """
+ Read cached token from file.
+
+ :return: Tuple of (token, expiry_time) or (None, None) if not
available or invalid
+ """
+ if not self.cache_approle_token:
+ return None, None
+
+ cache_file = self._get_cache_file_path()
+ if not cache_file.exists():
+ return None, None
+
+ try:
+ # Use file locking to prevent race conditions
+ with open(cache_file) as f:
+ fcntl.flock(f.fileno(), fcntl.LOCK_SH) # Shared lock for
reading
+ try:
+ data = json.load(f)
+ token = data.get("token")
+ expiry = data.get("expiry")
+ if token and expiry:
+ return token, expiry
+ finally:
+ fcntl.flock(f.fileno(), fcntl.LOCK_UN) # Release lock
+ except OSError as e:
+ self.log.warning("Failed to read cached token file: %s", e)
+ except (json.JSONDecodeError, ValueError) as e:
+ self.log.debug("Cached token file is invalid or corrupted: %s", e)
+
+ return None, None
+
+ def _write_cached_token(self, token: str, expiry: float) -> None:
+ """
+ Write token to cache file.
+
+ :param token: The Vault token to cache
+ :param expiry: Unix timestamp when the token expires
+ """
+ if not self.cache_approle_token:
+ return
+
+ cache_file = self._get_cache_file_path()
+ try:
+ # Write with exclusive lock
+ with open(cache_file, "w") as f:
Review Comment:
`open(cache_file, "w")` truncates the file on open, and `flock` is only
acquired on the next line. A concurrent reader holding the shared lock in
`_read_cached_token` can therefore observe an empty or partial file.
The docs added in this PR state that "File locking ensures safe concurrent
access by multiple processes on the same node" — as written that guarantee
doesn't hold. Opening `r+`/`a+`, locking, then truncating would.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
providers/hashicorp/src/airflow/providers/hashicorp/_internal_client/vault_client.py:
##########
@@ -200,6 +218,102 @@ def __init__(
self.jwt_role = jwt_role
self.jwt_token = jwt_token
self.jwt_token_path = jwt_token_path
+ self.cache_approle_token = cache_approle_token
+ self._cached_token_expiry: float | None = None
+
+ def _get_cache_file_path(self) -> Path:
+ """
+ Get the path to the token cache file.
+
+ Uses a deterministic path based on the vault URL and role_id to ensure
+ the same cache is used across different processes.
+
+ :return: Path to the cache file
+ """
+ cache_dir = Path(tempfile.gettempdir()) / CACHE_DIR_NAME
+ try:
+ cache_dir.mkdir(mode=CACHE_DIR_PERMISSIONS, exist_ok=True)
+ except OSError as e:
+ self.log.warning("Failed to create cache directory %s: %s",
cache_dir, e)
+ # Fallback to a cache file in temp directory root if we can't
create the subdirectory
+ return Path(tempfile.gettempdir()) /
f"vault_token_{self.role_id}.json"
+
+ # Create a unique filename based on vault URL and role_id
+ cache_key = f"{self.url}_{self.role_id}".encode()
+ cache_hash = hashlib.sha256(cache_key).hexdigest()[:16]
+ return cache_dir / f"vault_token_{cache_hash}.json"
+
+ def _read_cached_token(self) -> tuple[str | None, float | None]:
+ """
+ Read cached token from file.
+
+ :return: Tuple of (token, expiry_time) or (None, None) if not
available or invalid
+ """
+ if not self.cache_approle_token:
+ return None, None
+
+ cache_file = self._get_cache_file_path()
+ if not cache_file.exists():
+ return None, None
+
+ try:
+ # Use file locking to prevent race conditions
+ with open(cache_file) as f:
+ fcntl.flock(f.fileno(), fcntl.LOCK_SH) # Shared lock for
reading
+ try:
+ data = json.load(f)
+ token = data.get("token")
+ expiry = data.get("expiry")
+ if token and expiry:
+ return token, expiry
+ finally:
+ fcntl.flock(f.fileno(), fcntl.LOCK_UN) # Release lock
+ except OSError as e:
+ self.log.warning("Failed to read cached token file: %s", e)
+ except (json.JSONDecodeError, ValueError) as e:
+ self.log.debug("Cached token file is invalid or corrupted: %s", e)
+
+ return None, None
+
+ def _write_cached_token(self, token: str, expiry: float) -> None:
+ """
+ Write token to cache file.
+
+ :param token: The Vault token to cache
+ :param expiry: Unix timestamp when the token expires
+ """
+ if not self.cache_approle_token:
+ return
+
+ cache_file = self._get_cache_file_path()
+ try:
+ # Write with exclusive lock
+ with open(cache_file, "w") as f:
+ fcntl.flock(f.fileno(), fcntl.LOCK_EX) # Exclusive lock for
writing
+ try:
+ json.dump({"token": token, "expiry": expiry}, f)
+ # Set restrictive permissions (only owner can read/write)
+ os.chmod(cache_file, CACHE_FILE_PERMISSIONS)
Review Comment:
The token is written before the permissions are tightened: `open(..., "w")`
→ `json.dump(...)` → `os.chmod(..., 0o600)`. Between the dump and this chmod
the file exists with umask-default permissions (commonly `0644`) and already
contains the token.
Creating it with `os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY,
0o600)`, or writing to a private temp file and `os.replace`-ing it into place,
closes that window.
---
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
