ashb commented on code in PR #67503:
URL: https://github.com/apache/airflow/pull/67503#discussion_r3373321935
##########
task-sdk/src/airflow/sdk/log.py:
##########
@@ -271,19 +271,37 @@ def mask_secret(secret: JsonValue, name: str | None =
None) -> None:
they're masked in both the task subprocess AND supervisor's log output.
Works safely in both sync and async contexts.
"""
- from contextlib import suppress
-
from airflow.sdk._shared.secrets_masker import _secrets_masker
_secrets_masker().add_mask(secret, name)
- with suppress(Exception):
- # Try to tell supervisor (only if in task execution context)
- from airflow.sdk.execution_time import task_runner
- from airflow.sdk.execution_time.comms import MaskSecret
+ # Mirror the mask to the supervisor so it also masks the value in any logs
+ # forwarded through it. When this fails we *must* emit a warning rather
+ # than silently swallowing the failure: when ``sending_to_supervisor=True``
+ # the local task process skips its own ``mask_logs`` processor (it relies
+ # on the supervisor doing the masking instead), so a silent IPC failure
+ # would leave the secret unmasked in supervisor-level logs.
+ from airflow.sdk.execution_time import task_runner
+
+ comms = getattr(task_runner, "SUPERVISOR_COMMS", None)
+ if comms is None:
+ # Not in a task-execution context — there is no supervisor to notify.
Review Comment:
Nit: don't use the word "context" here, as we already have a context meaning
in the SDK that this is dangerously close to overlapping with the thing Kaxil
mentioned which was client or in process die etc.)
cc @kaxil
##########
task-sdk/src/airflow/sdk/log.py:
##########
@@ -249,19 +249,37 @@ def mask_secret(secret: JsonValue, name: str | None =
None) -> None:
they're masked in both the task subprocess AND supervisor's log output.
Works safely in both sync and async contexts.
"""
- from contextlib import suppress
-
from airflow.sdk._shared.secrets_masker import _secrets_masker
_secrets_masker().add_mask(secret, name)
- with suppress(Exception):
- # Try to tell supervisor (only if in task execution context)
- from airflow.sdk.execution_time import task_runner
- from airflow.sdk.execution_time.comms import MaskSecret
+ # Mirror the mask to the supervisor so it also masks the value in any logs
+ # forwarded through it. When this fails we *must* emit a warning rather
+ # than silently swallowing the failure: when ``sending_to_supervisor=True``
+ # the local task process skips its own ``mask_logs`` processor (it relies
+ # on the supervisor doing the masking instead), so a silent IPC failure
+ # would leave the secret unmasked in supervisor-level logs.
+ from airflow.sdk.execution_time import task_runner
+
+ comms = getattr(task_runner, "SUPERVISOR_COMMS", None)
+ if comms is None:
+ # Not in a task-execution context — there is no supervisor to notify.
+ return
+
+ from airflow.sdk.execution_time.comms import MaskSecret
- if comms := getattr(task_runner, "SUPERVISOR_COMMS", None):
- comms.send(MaskSecret(value=secret, name=name))
+ try:
+ comms.send(MaskSecret(value=secret, name=name))
+ except Exception:
+ # Dedicated logger so operators can grep for this signal — without a
+ # warning they'd have no way to find out the supervisor never received
+ # the registration.
+ structlog.get_logger("airflow.logging.mask_secret").warning(
+ "supervisor_mask_secret_failed",
+ secret_name=name,
+ note="Could not register secret with supervisor; secret may not be
masked in supervisor-level logs.",
+ exc_info=True,
Review Comment:
An interesting thought. Is there any chance that this exception being
included essentially _guarantees_ the sensitive data does get logged? Can you
force this case to happen and show what gets recorded to the task logs please?
--
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]
