From adde1b63478de86cc50e53635c98e5d10b4e6268 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Mon, 7 Sep 2026 18:17:22 +0200 Subject: [PATCH 01/13] Mask CLI secrets in the audit log by parsed argument, not argv position _build_metrics recorded sys.argv into the audit log and masked the variables value by position, assuming the key and value were the final two arguments. Any supported trailing option displaces them: 'airflow variables set KEY VALUE --description TEXT' ends in '--description TEXT', so the value was written to Log.extra in clear. --description is the instance found, not the class -- the same is true of --serialize-json and of any option added later. Take the key and value from the parsed Namespace instead, and redact that value wherever it appears in the recorded command. Position no longer matters. --conn-extra was not masked at all. It is the field connection secrets live in, and it accepts nested JSON. It is now redacted alongside --conn-json, and both are walked recursively -- --conn-json previously masked only top-level keys, so a secret one level down was recorded in clear. A value that cannot be parsed as JSON is redacted whole, because these options carry connection material and a value that cannot be inspected is not safe to record. Both argparse spellings are handled. --conn-json, --conn-extra and --conn-uri were matched as standalone tokens only, so the equally valid --conn-json=VALUE form skipped redaction entirely and recorded the secret. --conn-uri had that gap before this change as well. Note for operators: this prevents new disclosures only. Existing Log.extra rows and backups retain values already written and should be redacted or expired, and any credentials exposed that way rotated. The seven new cases fail against unpatched sources. --- airflow-core/src/airflow/utils/cli.py | 103 ++++++++++---- .../tests/unit/utils/test_cli_util.py | 129 +++++++++++++++++- 2 files changed, 206 insertions(+), 26 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index e3491bf91f1c2..e3ddcad7b7a81 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -137,6 +137,58 @@ def wrapper(*args, **kwargs): return action_logging +MASKED_VALUE = "*" * 8 + + +def _redact_structure(obj): + """Redact values whose key looks sensitive, at any depth.""" + from airflow._shared.secrets_masker import _secrets_masker + + if isinstance(obj, dict): + return { + k: MASKED_VALUE if k and _secrets_masker().should_hide_value_for_key(k) else _redact_structure(v) + for k, v in obj.items() + } + if isinstance(obj, list): + return [_redact_structure(item) for item in obj] + return obj + + +def _find_option_value(command: list[str], option: str) -> tuple[int | None, str]: + """ + Locate an option's value in a recorded command, in either accepted spelling. + + argparse accepts both ``--option VALUE`` and ``--option=VALUE``. Returns the index of + the element holding the value and the prefix that must be preserved when rewriting it + (empty for the separated form, ``--option=`` for the joined one). + """ + for idx, argument in enumerate(command): + if argument == option and idx + 1 < len(command): + return idx + 1, "" + if argument.startswith(f"{option}="): + return idx, f"{option}=" + return None, "" + + +def _redact_json_argument(value: str) -> str: + """ + Redact sensitive entries in a JSON command-line argument. + + Nested structures are walked, because a secret one level down is still a secret: + ``--conn-extra '{"nested": {"password": "..."}}'`` records the password otherwise. + + A value that is not JSON we can parse is redacted whole. These options carry + connection material, so a value that cannot be inspected is not safe to record. + """ + import json + + try: + parsed = json.loads(value) + except (json.JSONDecodeError, TypeError, ValueError): + return MASKED_VALUE + return json.dumps(_redact_structure(parsed)) + + def _build_metrics(func_name, namespace): """ Build metrics dict from function args. @@ -158,46 +210,51 @@ def _build_metrics(func_name, namespace): sub_command = full_command[1] if len(full_command) > 1 else None # For cases when value under sub_commands_to_check_for_sensitive_key have sensitive info if sub_command in sub_commands_to_check_for_sensitive_key: - key = full_command[-2] if len(full_command) > 3 else None - if key and _secrets_masker().should_hide_value_for_key(key): - # Mask the sensitive value since key contain sensitive keyword - full_command[-1] = "*" * 8 + # Identify the value from the parsed command rather than from its position in + # argv. The key and value are not reliably the last two arguments: any supported + # trailing option displaces them (``variables set KEY VALUE --description TEXT`` + # ends in ``--description TEXT``), which left the value recorded in clear. + key = getattr(namespace, "key", None) + value = getattr(namespace, "value", None) + if key and value and _secrets_masker().should_hide_value_for_key(key): + full_command = [MASKED_VALUE if arg == value else arg for arg in full_command] elif sub_command in sub_commands_to_check_for_sensitive_fields: for idx, command in enumerate(full_command): if command in sensitive_fields: # For cases when password is passed as "--password xyz" (with space between key and value) - full_command[idx + 1] = "*" * 8 + full_command[idx + 1] = MASKED_VALUE else: # For cases when password is passed as "--password=xyz" (with '=' between key and value) for sensitive_field in sensitive_fields: if command.startswith(f"{sensitive_field}="): - full_command[idx] = f"{sensitive_field}={'*' * 8}" - - # handle conn-json and conn-uri separately as it requires different handling - if "--conn-json" in full_command: - import json - - json_index = full_command.index("--conn-json") + 1 - conn_json = json.loads(full_command[json_index]) - for k in conn_json: - if k and _secrets_masker().should_hide_value_for_key(k): - conn_json[k] = "*" * 8 - full_command[json_index] = json.dumps(conn_json) - - if "--conn-uri" in full_command: + full_command[idx] = f"{sensitive_field}={MASKED_VALUE}" + + # handle conn-json, conn-extra and conn-uri separately as they require different handling + for json_option in ("--conn-json", "--conn-extra"): + for idx, argument in enumerate(full_command): + # argparse accepts both "--conn-json VALUE" and "--conn-json=VALUE", so the + # value is matched in either spelling rather than by looking one element past + # a standalone token. + if argument == json_option and idx + 1 < len(full_command): + full_command[idx + 1] = _redact_json_argument(full_command[idx + 1]) + elif argument.startswith(f"{json_option}="): + redacted = _redact_json_argument(argument[len(json_option) + 1 :]) + full_command[idx] = f"{json_option}={redacted}" + + uri_index, uri_prefix = _find_option_value(full_command, "--conn-uri") + if uri_index is not None: from urllib.parse import urlparse, urlunparse - uri_index = full_command.index("--conn-uri") + 1 - conn_uri = full_command[uri_index] + conn_uri = full_command[uri_index][len(uri_prefix) :] parsed_uri = urlparse(conn_uri) netloc = parsed_uri.netloc if parsed_uri.password: - password = "*" * 8 + password = MASKED_VALUE netloc = f"{parsed_uri.username}:{password}@{parsed_uri.hostname}" if parsed_uri.port: netloc += f":{parsed_uri.port}" - full_command[uri_index] = urlunparse( + full_command[uri_index] = uri_prefix + urlunparse( ( parsed_uri.scheme, netloc, diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 3539f8b3e52ef..bd5e82981d2c4 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -160,6 +160,95 @@ def test_get_dags(self): ], True, ), + ( + [ + "airflow", + "connections", + "add", + "tp005-demo", + "--conn-type", + "generic", + "--conn-extra", + '{"nested": {"password": "TP005_CONN_DUMMY"}, "region": "eu-west-1"}', + ], + [ + "airflow", + "connections", + "add", + "tp005-demo", + "--conn-type", + "generic", + "--conn-extra", + '{"nested": {"password": "********"}, "region": "eu-west-1"}', + ], + True, + ), + ( + [ + "airflow", + "connections", + "add", + "nested_conn", + "--conn-json", + '{"conn_type": "generic", "extra": {"auth": {"password": "deep-secret"}}}', + ], + [ + "airflow", + "connections", + "add", + "nested_conn", + "--conn-json", + '{"conn_type": "generic", "extra": {"auth": {"password": "********"}}}', + ], + True, + ), + # argparse accepts "--option=VALUE" too; the audit masker must cover both + # spellings or a supported invocation records the secret in clear. + ( + [ + "airflow", + "connections", + "add", + "eq_json_conn", + '--conn-json={"conn_type": "generic", "password": "eq-secret", ' + '"extra": {"auth": {"api_key": "deep-eq-secret"}}}', + ], + [ + "airflow", + "connections", + "add", + "eq_json_conn", + '--conn-json={"conn_type": "generic", "password": "********", ' + '"extra": {"auth": {"api_key": "********"}}}', + ], + True, + ), + ( + [ + "airflow", + "connections", + "add", + "eq_extra_conn", + "--conn-type", + "generic", + '--conn-extra={"nested": {"password": "eq-extra-secret"}}', + ], + [ + "airflow", + "connections", + "add", + "eq_extra_conn", + "--conn-type", + "generic", + '--conn-extra={"nested": {"password": "********"}}', + ], + True, + ), + ( + "airflow connections add eq_uri_conn --conn-uri=postgresql://user:my-password@localhost:5432/db", + "airflow connections add eq_uri_conn --conn-uri=postgresql://user:********@localhost:5432/db", + False, + ), ( "airflow scheduler -p", "airflow scheduler -p", @@ -261,27 +350,61 @@ def test_setup_locations_none_pid_path(self): assert pid == default_pid_path @pytest.mark.parametrize( - ("given_command", "expected_masked_command"), + ("given_command", "expected_masked_command", "var_key", "var_value"), [ ( "airflow variables set --description 'needed for dag 4' client_secret_234 7fh4375f5gy353wdf", "airflow variables set --description 'needed for dag 4' client_secret_234 ********", + "client_secret_234", + "7fh4375f5gy353wdf", ), ( "airflow variables set cust_secret_234 7fh4375f5gy353wdf", "airflow variables set cust_secret_234 ********", + "cust_secret_234", + "7fh4375f5gy353wdf", + ), + # A trailing option displaces the value out of the last position. The masker + # used to key off argv positions and recorded the value in clear here. + ( + "airflow variables set client_secret_234 7fh4375f5gy353wdf --description validation", + "airflow variables set client_secret_234 ******** --description validation", + "client_secret_234", + "7fh4375f5gy353wdf", + ), + ( + "airflow variables set client_secret_234 7fh4375f5gy353wdf --serialize-json", + "airflow variables set client_secret_234 ******** --serialize-json", + "client_secret_234", + "7fh4375f5gy353wdf", + ), + # A non-sensitive key must still be recorded as given. + ( + "airflow variables set retries 5 --description validation", + "airflow variables set retries 5 --description validation", + "retries", + "5", ), ], ) def test_cli_set_variable_supplied_sensitive_value_is_masked( - self, given_command, expected_masked_command, session + self, given_command, expected_masked_command, var_key, var_value, session ): args = given_command.split() expected_command = expected_masked_command.split() + # `variables set` parses the key and value into the namespace; the masker reads + # them from there rather than guessing their position in argv. exec_date = timezone.utcnow() - namespace = Namespace(dag_id="foo", task_id="bar", subcommand="test", execution_date=exec_date) + namespace = Namespace( + dag_id="foo", + task_id="bar", + subcommand="test", + execution_date=exec_date, + key=var_key, + value=var_value, + ) with ( mock.patch.object(sys, "argv", args), mock.patch("airflow.utils.session.create_session") as mock_create_session, From 2028c738d4f6e48cf229d7729cb24109021199c8 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Wed, 23 Sep 2026 00:02:51 +0200 Subject: [PATCH 02/13] Mask long-form password options on every CLI subcommand in the audit log --password and --conn-password name a password on every subcommand, including those contributed by providers (keycloak-auth-manager ... --password), but were only masked for users and connections. Mask the long spellings everywhere; keep -p limited to users and connections, as elsewhere it means --port or --exclude-providers. Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 18 ++++++++++++++---- airflow-core/tests/unit/utils/test_cli_util.py | 15 +++++++++++++++ 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index e3ddcad7b7a81..58eeaadadb430 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -206,6 +206,10 @@ def _build_metrics(func_name, namespace): sub_commands_to_check_for_sensitive_fields = {"users", "connections"} sub_commands_to_check_for_sensitive_key = {"variables"} sensitive_fields = {"-p", "--password", "--conn-password"} + # The long spellings name a password on every subcommand, including ones contributed by + # providers (``keycloak-auth-manager ... --password``), so they are masked everywhere. ``-p`` + # is only a password on the subcommands above; elsewhere it is e.g. ``--port``. + always_sensitive_fields = {"--password", "--conn-password"} full_command = list(sys.argv) sub_command = full_command[1] if len(full_command) > 1 else None # For cases when value under sub_commands_to_check_for_sensitive_key have sensitive info @@ -218,14 +222,20 @@ def _build_metrics(func_name, namespace): value = getattr(namespace, "value", None) if key and value and _secrets_masker().should_hide_value_for_key(key): full_command = [MASKED_VALUE if arg == value else arg for arg in full_command] - elif sub_command in sub_commands_to_check_for_sensitive_fields: + else: + fields = ( + sensitive_fields + if sub_command in sub_commands_to_check_for_sensitive_fields + else always_sensitive_fields + ) for idx, command in enumerate(full_command): - if command in sensitive_fields: + if command in fields: # For cases when password is passed as "--password xyz" (with space between key and value) - full_command[idx + 1] = MASKED_VALUE + if idx + 1 < len(full_command): + full_command[idx + 1] = MASKED_VALUE else: # For cases when password is passed as "--password=xyz" (with '=' between key and value) - for sensitive_field in sensitive_fields: + for sensitive_field in fields: if command.startswith(f"{sensitive_field}="): full_command[idx] = f"{sensitive_field}={MASKED_VALUE}" diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index bd5e82981d2c4..aeca3a1f893d6 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -259,6 +259,21 @@ def test_get_dags(self): "airflow celery flower -p 8888", False, ), + ( + "airflow keycloak-auth-manager create-scopes --username admin --password s3cr3t", + "airflow keycloak-auth-manager create-scopes --username admin --password ********", + False, + ), + ( + "airflow keycloak-auth-manager create-scopes --username admin --password=s3cr3t", + "airflow keycloak-auth-manager create-scopes --username admin --password=********", + False, + ), + ( + "airflow api-server -p 8080", + "airflow api-server -p 8080", + False, + ), ], ) def test_cli_create_user_supplied_password_is_masked( From 0885b4aec2634d87045efde842da5b6ab45589df Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Fri, 16 Oct 2026 07:04:40 +0100 Subject: [PATCH 03/13] Mask CLI secrets by parsed value in every argument spelling Mask secrets supplied through any supported argument spelling in the recorded command line: abbreviated and attached options, repeated options, JSON (incl. conf) and URI values, and celery flower auth. Generated-by: Claude Opus 5 following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions --- airflow-core/src/airflow/utils/cli.py | 206 +++++++++---- .../tests/unit/utils/test_cli_util.py | 287 ++++++++++++++++++ 2 files changed, 442 insertions(+), 51 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 58eeaadadb430..aa6c65824bf6d 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -139,35 +139,100 @@ def wrapper(*args, **kwargs): MASKED_VALUE = "*" * 8 +# Options whose value is a secret on its own, keyed by the destination argparse stores the +# parsed value under. Masking is driven by what is present in the parsed namespace, so ``-p`` +# is only treated as a password on commands where it parses into ``password`` (``api-server -p`` +# parses into ``port`` and is left alone). +_SECRET_OPTIONS: dict[str, tuple[str, ...]] = { + "password": ("-p", "--password"), + "conn_password": ("--conn-password",), + "basic_auth": ("-A", "--basic-auth"), +} +# Options whose value is JSON that may hold secrets under sensitive keys. +_JSON_OPTIONS: dict[str, tuple[str, ...]] = { + "conn_json": ("--conn-json",), + "conn_extra": ("--conn-extra",), + "conf": ("-c", "--conf"), + "dag_run_conf": ("--dag-run-conf",), +} +# Options whose value is a URI that may hold secrets in its credentials or query. +_URI_OPTIONS: dict[str, tuple[str, ...]] = { + "conn_uri": ("--conn-uri",), + "broker_api": ("-a", "--broker-api"), +} + + +def _should_hide_key(key) -> bool: + from airflow._shared.secrets_masker import _secrets_masker + + return bool(key) and isinstance(key, str) and _secrets_masker().should_hide_value_for_key(key) + def _redact_structure(obj): - """Redact values whose key looks sensitive, at any depth.""" - from airflow._shared.secrets_masker import _secrets_masker + """ + Redact values whose key looks sensitive, at any depth. + + String values that themselves hold a JSON object or list are parsed and redacted too: + ``--conn-json`` accepts ``extra`` as a JSON-encoded string as well as an object. + """ + import json if isinstance(obj, dict): - return { - k: MASKED_VALUE if k and _secrets_masker().should_hide_value_for_key(k) else _redact_structure(v) - for k, v in obj.items() - } + return {k: MASKED_VALUE if _should_hide_key(k) else _redact_structure(v) for k, v in obj.items()} if isinstance(obj, list): return [_redact_structure(item) for item in obj] + if isinstance(obj, str) and obj.lstrip()[:1] in ("{", "["): + try: + parsed = json.loads(obj) + except ValueError: + return obj + redacted = _redact_structure(parsed) + return obj if redacted == parsed else json.dumps(redacted) return obj -def _find_option_value(command: list[str], option: str) -> tuple[int | None, str]: +def _find_option_values( + command: list[str], option_strings: tuple[str, ...], *, allow_abbrev: bool +) -> list[tuple[int, str]]: """ - Locate an option's value in a recorded command, in either accepted spelling. + Locate every value given for an option in a recorded command, in any accepted spelling. - argparse accepts both ``--option VALUE`` and ``--option=VALUE``. Returns the index of - the element holding the value and the prefix that must be preserved when rewriting it - (empty for the separated form, ``--option=`` for the joined one). + argparse accepts ``--option VALUE``, ``--option=VALUE``, unambiguous abbreviations of + long options (``--opt VALUE``), and for short options also ``-o VALUE``, ``-o=VALUE`` and + ``-oVALUE``. Returns ``(index, prefix)`` pairs: the index of the element holding the + value and the prefix to keep when rewriting it (empty for the separated forms). + + Every occurrence is returned: argparse keeps the last one, but all of them were typed. + Abbreviations and short options are only matched with ``allow_abbrev``, i.e. when the + parsed namespace confirms the command takes this option. """ + long_options = [o for o in option_strings if o.startswith("--")] + short_options = [o for o in option_strings if not o.startswith("--")] if allow_abbrev else [] + positions: list[tuple[int, str]] = [] for idx, argument in enumerate(command): - if argument == option and idx + 1 < len(command): - return idx + 1, "" - if argument.startswith(f"{option}="): - return idx, f"{option}=" - return None, "" + if argument.startswith("--"): + name, eq, _ = argument.partition("=") + matches = ( + any(len(name) > 2 and option.startswith(name) for option in long_options) + if allow_abbrev + else name in long_options + ) + if not matches: + continue + if eq: + positions.append((idx, f"{name}=")) + elif idx + 1 < len(command): + positions.append((idx + 1, "")) + else: + for option in short_options: + if argument == option: + if idx + 1 < len(command): + positions.append((idx + 1, "")) + elif argument.startswith(f"{option}="): + positions.append((idx, f"{option}=")) + elif argument.startswith(option): + positions.append((idx, option)) + return positions def _redact_json_argument(value: str) -> str: @@ -189,6 +254,77 @@ def _redact_json_argument(value: str) -> str: return json.dumps(_redact_structure(parsed)) +def _redact_uri_argument(value: str) -> str: + """ + Redact the secrets a URI command-line argument can carry. + + That is the password in the credentials, query parameters with a sensitive name + (``?api_key=...``) and sensitive entries of the URL-encoded ``__extra__`` JSON that + connection URIs accept. + """ + from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse + + try: + parsed_uri = urlparse(value) + has_password = parsed_uri.password is not None + except ValueError: + return MASKED_VALUE + netloc = parsed_uri.netloc + if has_password: + userinfo, _, host = netloc.rpartition("@") + netloc = f"{userinfo.partition(':')[0]}:{MASKED_VALUE}@{host}" + + query = parsed_uri.query + if query: + params = parse_qsl(query, keep_blank_values=True) + redacted_params = [] + for key, item in params: + if key == "__extra__": + redacted_params.append((key, _redact_json_argument(item))) + elif _should_hide_key(key): + redacted_params.append((key, MASKED_VALUE)) + else: + redacted_params.append((key, item)) + if redacted_params != params: + query = urlencode(redacted_params, safe="*") + + return urlunparse( + (parsed_uri.scheme, netloc, parsed_uri.path, parsed_uri.params, query, parsed_uri.fragment) + ) + + +def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: + """ + Mask secrets supplied through any supported argument spelling in the recorded command line. + + The values are taken from the parsed namespace, so the masking does not depend on how + the option was spelled on the command line. + """ + parsed = vars(namespace) if isinstance(namespace, Namespace) else {} + full_command = list(full_command) + + for dest, option_strings in _SECRET_OPTIONS.items(): + value = parsed.get(dest) + if not isinstance(value, str) or not value: + continue + for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=True): + full_command[idx] = prefix + MASKED_VALUE + full_command = [MASKED_VALUE if argument == value else argument for argument in full_command] + + for options, redact in ((_JSON_OPTIONS, _redact_json_argument), (_URI_OPTIONS, _redact_uri_argument)): + for dest, option_strings in options.items(): + value = parsed.get(dest) + in_namespace = isinstance(value, str) and bool(value) + for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=in_namespace): + full_command[idx] = prefix + redact(full_command[idx][len(prefix) :]) + if in_namespace: + full_command = [ + redact(argument) if argument == value else argument for argument in full_command + ] + + return full_command + + def _build_metrics(func_name, namespace): """ Build metrics dict from function args. @@ -239,41 +375,9 @@ def _build_metrics(func_name, namespace): if command.startswith(f"{sensitive_field}="): full_command[idx] = f"{sensitive_field}={MASKED_VALUE}" - # handle conn-json, conn-extra and conn-uri separately as they require different handling - for json_option in ("--conn-json", "--conn-extra"): - for idx, argument in enumerate(full_command): - # argparse accepts both "--conn-json VALUE" and "--conn-json=VALUE", so the - # value is matched in either spelling rather than by looking one element past - # a standalone token. - if argument == json_option and idx + 1 < len(full_command): - full_command[idx + 1] = _redact_json_argument(full_command[idx + 1]) - elif argument.startswith(f"{json_option}="): - redacted = _redact_json_argument(argument[len(json_option) + 1 :]) - full_command[idx] = f"{json_option}={redacted}" - - uri_index, uri_prefix = _find_option_value(full_command, "--conn-uri") - if uri_index is not None: - from urllib.parse import urlparse, urlunparse - - conn_uri = full_command[uri_index][len(uri_prefix) :] - parsed_uri = urlparse(conn_uri) - netloc = parsed_uri.netloc - if parsed_uri.password: - password = MASKED_VALUE - netloc = f"{parsed_uri.username}:{password}@{parsed_uri.hostname}" - if parsed_uri.port: - netloc += f":{parsed_uri.port}" - - full_command[uri_index] = uri_prefix + urlunparse( - ( - parsed_uri.scheme, - netloc, - parsed_uri.path, - parsed_uri.params, - parsed_uri.query, - parsed_uri.fragment, - ) - ) + # Options carrying secrets (on their own, or inside JSON and URIs) are masked by their + # parsed value, whatever spelling was used for them on the command line. + full_command = _mask_parsed_arguments(full_command, namespace) metrics = { "sub_command": func_name, diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index aeca3a1f893d6..0bfc4343485be 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -47,6 +47,7 @@ # - ``cli_action_loggers.on_post_execution`` pytestmark = pytest.mark.db_test repo_root = Path(airflow.__file__).parents[1] +MASKED = "*" * 8 class TestCliUtil: @@ -306,6 +307,292 @@ def test_cli_create_user_supplied_password_is_masked( command = ast.literal_eval(command) assert command == expected_command + @pytest.mark.parametrize( + ("given_command", "parsed", "expected_command"), + [ + pytest.param( + ["airflow", "users", "create", "-u", "a", "-r", "Admin", "-pS3CRET"], + {"username": "a", "role": "Admin", "password": "S3CRET"}, + ["airflow", "users", "create", "-u", "a", "-r", "Admin", f"-p{MASKED}"], + id="users-short-option-attached", + ), + pytest.param( + ["airflow", "users", "create", "-u", "a", "--pass", "S3CRET"], + {"username": "a", "password": "S3CRET"}, + ["airflow", "users", "create", "-u", "a", "--pass", MASKED], + id="users-abbreviated", + ), + pytest.param( + ["airflow", "users", "reset-password", "-u", "a", "--passw=S3CRET"], + {"username": "a", "password": "S3CRET"}, + ["airflow", "users", "reset-password", "-u", "a", f"--passw={MASKED}"], + id="users-abbreviated-joined", + ), + pytest.param( + ["airflow", "users", "create", "-u", "a", "--p", "S3CRET"], + {"username": "a", "password": "S3CRET"}, + ["airflow", "users", "create", "-u", "a", "--p", MASKED], + id="users-shortest-abbreviation", + ), + pytest.param( + [ + "airflow", + "keycloak-auth-manager", + "create-scopes", + "--username", + "admin", + "--pass", + "S3CRET", + ], + {"username": "admin", "password": "S3CRET"}, + [ + "airflow", + "keycloak-auth-manager", + "create-scopes", + "--username", + "admin", + "--pass", + MASKED, + ], + id="keycloak-abbreviated", + ), + pytest.param( + ["airflow", "keycloak-auth-manager", "create-all", "--username", "admin", "--p=S3CRET"], + {"username": "admin", "password": "S3CRET"}, + ["airflow", "keycloak-auth-manager", "create-all", "--username", "admin", f"--p={MASKED}"], + id="keycloak-abbreviated-joined", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-type", "generic", "--conn-pass", "S3CRET"], + {"conn_id": "c1", "conn_type": "generic", "conn_password": "S3CRET"}, + ["airflow", "connections", "add", "c1", "--conn-type", "generic", "--conn-pass", MASKED], + id="conn-password-abbreviated", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-type", "generic", "--conn-pa=S3CRET"], + {"conn_id": "c1", "conn_type": "generic", "conn_password": "S3CRET"}, + ["airflow", "connections", "add", "c1", "--conn-type", "generic", f"--conn-pa={MASKED}"], + id="conn-password-abbreviated-joined", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-ext", '{"password": "S3CRET"}'], + {"conn_id": "c1", "conn_extra": '{"password": "S3CRET"}'}, + ["airflow", "connections", "add", "c1", "--conn-ext", f'{{"password": "{MASKED}"}}'], + id="conn-extra-abbreviated", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-j", + '{"conn_type": "generic", "password": "S3CRET"}', + ], + {"conn_id": "c1", "conn_json": '{"conn_type": "generic", "password": "S3CRET"}'}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-j", + f'{{"conn_type": "generic", "password": "{MASKED}"}}', + ], + id="conn-json-abbreviated", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-u", "postgres://u:S3CRET@h/db"], + {"conn_id": "c1", "conn_uri": "postgres://u:S3CRET@h/db"}, + ["airflow", "connections", "add", "c1", "--conn-u", f"postgres://u:{MASKED}@h/db"], + id="conn-uri-abbreviated", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-json", + '{"extra": "{\\"password\\": \\"S3CRET\\"}"}', + ], + {"conn_id": "c1", "conn_json": '{"extra": "{\\"password\\": \\"S3CRET\\"}"}'}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-json", + f'{{"extra": "{{\\"password\\": \\"{MASKED}\\"}}"}}', + ], + id="conn-json-extra-as-json-string", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "http://h/?api_key=S3CRET®ion=eu"], + {"conn_id": "c1", "conn_uri": "http://h/?api_key=S3CRET®ion=eu"}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + f"http://h/?api_key={MASKED}®ion=eu", + ], + id="conn-uri-query-parameter", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + "http://h/?__extra__=%7B%22token%22%3A%22S3CRET%22%7D", + ], + {"conn_id": "c1", "conn_uri": "http://h/?__extra__=%7B%22token%22%3A%22S3CRET%22%7D"}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + f"http://h/?__extra__=%7B%22token%22%3A+%22{MASKED}%22%7D", + ], + id="conn-uri-encoded-extra", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "aws://AKIA:S3CRET@"], + {"conn_id": "c1", "conn_uri": "aws://AKIA:S3CRET@"}, + ["airflow", "connections", "add", "c1", "--conn-uri", f"aws://AKIA:{MASKED}@"], + id="conn-uri-without-host", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + "postgres://u:FIRST@h/db", + "--conn-uri", + "postgres://u:S3CRET@h/db", + ], + {"conn_id": "c1", "conn_uri": "postgres://u:S3CRET@h/db"}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + f"postgres://u:{MASKED}@h/db", + "--conn-uri", + f"postgres://u:{MASKED}@h/db", + ], + id="conn-uri-repeated", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-extra", + '{"password": "FIRST"}', + "--conn-extra", + '{"password": "S3CRET"}', + ], + {"conn_id": "c1", "conn_extra": '{"password": "S3CRET"}'}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-extra", + f'{{"password": "{MASKED}"}}', + "--conn-extra", + f'{{"password": "{MASKED}"}}', + ], + id="conn-extra-repeated", + ), + pytest.param( + ["airflow", "celery", "flower", "--basic-auth", "u:S3CRET"], + {"basic_auth": "u:S3CRET", "port": 5555}, + ["airflow", "celery", "flower", "--basic-auth", MASKED], + id="flower-basic-auth", + ), + pytest.param( + ["airflow", "celery", "flower", "-A", "u:S3CRET"], + {"basic_auth": "u:S3CRET", "port": 5555}, + ["airflow", "celery", "flower", "-A", MASKED], + id="flower-basic-auth-short", + ), + pytest.param( + ["airflow", "celery", "flower", "--broker-api", "http://u:S3CRET@rabbit:15672/api/"], + {"broker_api": "http://u:S3CRET@rabbit:15672/api/", "port": 5555}, + ["airflow", "celery", "flower", "--broker-api", f"http://u:{MASKED}@rabbit:15672/api/"], + id="flower-broker-api", + ), + pytest.param( + ["airflow", "dags", "trigger", "d1", "--conf", '{"password": "S3CRET", "x": 1}'], + {"dag_id": "d1", "conf": '{"password": "S3CRET", "x": 1}'}, + ["airflow", "dags", "trigger", "d1", "--conf", f'{{"password": "{MASKED}", "x": 1}}'], + id="dags-trigger-conf", + ), + pytest.param( + ["airflow", "dags", "trigger", "d1", '-c{"api_key": "S3CRET"}'], + {"dag_id": "d1", "conf": '{"api_key": "S3CRET"}'}, + ["airflow", "dags", "trigger", "d1", f'-c{{"api_key": "{MASKED}"}}'], + id="dags-trigger-conf-short-attached", + ), + pytest.param( + ["airflow", "backfill", "create", "--dag-id", "d1", "--dag-run-conf", '{"token": "S3CRET"}'], + {"dag_id": "d1", "dag_run_conf": '{"token": "S3CRET"}'}, + [ + "airflow", + "backfill", + "create", + "--dag-id", + "d1", + "--dag-run-conf", + f'{{"token": "{MASKED}"}}', + ], + id="backfill-dag-run-conf", + ), + # ``-p`` that does not parse into a password is recorded as given. + pytest.param( + ["airflow", "api-server", "-p", "8080"], + {"port": 8080}, + ["airflow", "api-server", "-p", "8080"], + id="api-server-port", + ), + pytest.param( + ["airflow", "roles", "list", "-p"], + {"permission": True}, + ["airflow", "roles", "list", "-p"], + id="roles-list-permission", + ), + pytest.param( + ["airflow", "celery", "flower", "-p", "5555"], + {"port": 5555, "basic_auth": None, "broker_api": None}, + ["airflow", "celery", "flower", "-p", "5555"], + id="flower-port", + ), + pytest.param( + ["airflow", "dags", "trigger", "d1", "--conf", '{"retries": 3}'], + {"dag_id": "d1", "conf": '{"retries": 3}'}, + ["airflow", "dags", "trigger", "d1", "--conf", '{"retries": 3}'], + id="dags-trigger-conf-without-secrets", + ), + ], + ) + def test_secrets_masked_in_any_argument_spelling(self, given_command, parsed, expected_command): + """Mask secrets supplied through any supported argument spelling in the recorded command line.""" + with mock.patch.object(sys, "argv", given_command): + metrics = cli._build_metrics(given_command[1], Namespace(**parsed)) + + command = ast.literal_eval(metrics["full_command"]) + assert command == expected_command + assert not any("S3CRET" in argument or "FIRST" in argument for argument in command) + def test_action_log_records_team_name(self, session): namespace = Namespace(team_name="payments") with ( From d073b53020da3711f933975c80fbd8494d823da6 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Fri, 16 Oct 2026 07:19:52 +0100 Subject: [PATCH 04/13] Mask secrets after grouped short flags and in task-params / env-vars Mask a parsed secret wherever it appears in a recorded token, which covers grouped short flags, and add tasks test -t/--task-params and --env-vars to the JSON options. Generated-by: Claude Opus 5 following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions --- airflow-core/src/airflow/utils/cli.py | 10 ++++++++-- airflow-core/tests/unit/utils/test_cli_util.py | 18 ++++++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index aa6c65824bf6d..57d156d24097d 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -154,6 +154,8 @@ def wrapper(*args, **kwargs): "conn_extra": ("--conn-extra",), "conf": ("-c", "--conf"), "dag_run_conf": ("--dag-run-conf",), + "task_params": ("-t", "--task-params"), + "env_vars": ("--env-vars",), } # Options whose value is a URI that may hold secrets in its credentials or query. _URI_OPTIONS: dict[str, tuple[str, ...]] = { @@ -309,12 +311,16 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: continue for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=True): full_command[idx] = prefix + MASKED_VALUE - full_command = [MASKED_VALUE if argument == value else argument for argument in full_command] + # The value can also sit inside a token the option lookup does not recognise, e.g. after + # grouped short flags (``-vpVALUE``), so mask it wherever it appears. + full_command = [argument.replace(value, MASKED_VALUE) for argument in full_command] for options, redact in ((_JSON_OPTIONS, _redact_json_argument), (_URI_OPTIONS, _redact_uri_argument)): for dest, option_strings in options.items(): value = parsed.get(dest) - in_namespace = isinstance(value, str) and bool(value) + # Some of these options parse their JSON on the way in (``--env-vars``), so a dict or + # list in the namespace also means the command takes the option. + in_namespace = isinstance(value, (str, dict, list)) and bool(value) for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=in_namespace): full_command[idx] = prefix + redact(full_command[idx][len(prefix) :]) if in_namespace: diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 0bfc4343485be..4dfaa770ffdf5 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -557,6 +557,24 @@ def test_cli_create_user_supplied_password_is_masked( ], id="backfill-dag-run-conf", ), + pytest.param( + ["airflow", "users", "create", "-u", "a", "-vpS3CRET"], + {"username": "a", "verbose": True, "password": "S3CRET"}, + ["airflow", "users", "create", "-u", "a", "-vp********"], + id="grouped-short-flags", + ), + pytest.param( + ["airflow", "tasks", "test", "d1", "t1", "-t", '{"password": "S3CRET"}'], + {"dag_id": "d1", "task_id": "t1", "task_params": '{"password": "S3CRET"}'}, + ["airflow", "tasks", "test", "d1", "t1", "-t", '{"password": "********"}'], + id="tasks-test-task-params", + ), + pytest.param( + ["airflow", "tasks", "test", "d1", "t1", "--env-v", '{"API_TOKEN": "S3CRET"}'], + {"dag_id": "d1", "task_id": "t1", "env_vars": {"API_TOKEN": "S3CRET"}}, + ["airflow", "tasks", "test", "d1", "t1", "--env-v", '{"API_TOKEN": "********"}'], + id="tasks-test-env-vars-abbreviated", + ), # ``-p`` that does not parse into a password is recorded as given. pytest.param( ["airflow", "api-server", "-p", "8080"], From bdd57a3484ac05eeaf20f442277c51549c34f173 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Fri, 16 Oct 2026 07:26:53 +0100 Subject: [PATCH 05/13] Limit in-token masking to whole values and short-option clusters Mask a parsed value only where it is a whole token or ends a single-dash short-option cluster, for plain, JSON and URI options alike, so a short value is not masked wherever its text happens to appear in the recorded command line. Generated-by: Claude Opus 5 following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions --- airflow-core/src/airflow/utils/cli.py | 27 +++++++++---- .../tests/unit/utils/test_cli_util.py | 40 +++++++++++++++++++ 2 files changed, 60 insertions(+), 7 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 57d156d24097d..1f7616447fdad 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -295,6 +295,21 @@ def _redact_uri_argument(value: str) -> str: ) +def _mask_value_token(argument: str, value: str, redact) -> str: + """ + Redact a parsed option value where it forms a whole token or ends a short-option cluster. + + A cluster of grouped short flags (``-vpVALUE``) carries the value of its last option, so a + single-dash token ending with the value is rewritten too. Other tokens are left alone, so + a short value is not masked wherever its text happens to appear. + """ + if argument == value: + return redact(value) + if argument.startswith("-") and not argument.startswith("--") and argument.endswith(value): + return argument[: -len(value)] + redact(value) + return argument + + def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: """ Mask secrets supplied through any supported argument spelling in the recorded command line. @@ -311,9 +326,9 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: continue for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=True): full_command[idx] = prefix + MASKED_VALUE - # The value can also sit inside a token the option lookup does not recognise, e.g. after - # grouped short flags (``-vpVALUE``), so mask it wherever it appears. - full_command = [argument.replace(value, MASKED_VALUE) for argument in full_command] + full_command = [ + _mask_value_token(argument, value, lambda _: MASKED_VALUE) for argument in full_command + ] for options, redact in ((_JSON_OPTIONS, _redact_json_argument), (_URI_OPTIONS, _redact_uri_argument)): for dest, option_strings in options.items(): @@ -323,10 +338,8 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: in_namespace = isinstance(value, (str, dict, list)) and bool(value) for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=in_namespace): full_command[idx] = prefix + redact(full_command[idx][len(prefix) :]) - if in_namespace: - full_command = [ - redact(argument) if argument == value else argument for argument in full_command - ] + if in_namespace and isinstance(value, str): + full_command = [_mask_value_token(argument, value, redact) for argument in full_command] return full_command diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 4dfaa770ffdf5..69af2d4de2dec 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -563,6 +563,28 @@ def test_cli_create_user_supplied_password_is_masked( ["airflow", "users", "create", "-u", "a", "-vp********"], id="grouped-short-flags", ), + pytest.param( + ["airflow", "tasks", "test", "d1", "t1", "-mt", '{"password": "S3CRET"}'], + { + "dag_id": "d1", + "task_id": "t1", + "post_mortem": True, + "task_params": '{"password": "S3CRET"}', + }, + ["airflow", "tasks", "test", "d1", "t1", "-mt", '{"password": "********"}'], + id="grouped-short-flags-json-separate", + ), + pytest.param( + ["airflow", "tasks", "test", "d1", "t1", '-mt{"password": "S3CRET"}'], + { + "dag_id": "d1", + "task_id": "t1", + "post_mortem": True, + "task_params": '{"password": "S3CRET"}', + }, + ["airflow", "tasks", "test", "d1", "t1", '-mt{"password": "********"}'], + id="grouped-short-flags-json-attached", + ), pytest.param( ["airflow", "tasks", "test", "d1", "t1", "-t", '{"password": "S3CRET"}'], {"dag_id": "d1", "task_id": "t1", "task_params": '{"password": "S3CRET"}'}, @@ -611,6 +633,24 @@ def test_secrets_masked_in_any_argument_spelling(self, given_command, parsed, ex assert command == expected_command assert not any("S3CRET" in argument or "FIRST" in argument for argument in command) + def test_short_secret_is_not_masked_inside_other_arguments(self): + """A short secret only masks the option value, not every token containing its text.""" + command = ["airflow", "users", "create", "-u", "alice", "-r", "Admin", "-p", "a"] + with mock.patch.object(sys, "argv", command): + metrics = cli._build_metrics("users", Namespace(username="alice", role="Admin", password="a")) + + assert ast.literal_eval(metrics["full_command"]) == [ + "airflow", + "users", + "create", + "-u", + "alice", + "-r", + "Admin", + "-p", + "********", + ] + def test_action_log_records_team_name(self, session): namespace = Namespace(team_name="payments") with ( From 150c845f319171277c56755a4ce3ac2fc0f9498b Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 17 Oct 2026 01:44:14 +0200 Subject: [PATCH 06/13] Mask clustered, repeated-empty and double-scheme URI secrets in CLI audit log Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 93 ++++++++++++++----- .../tests/unit/utils/test_cli_util.py | 76 +++++++++++++++ 2 files changed, 147 insertions(+), 22 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 1f7616447fdad..45c47ef9f207c 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -194,7 +194,11 @@ def _redact_structure(obj): def _find_option_values( - command: list[str], option_strings: tuple[str, ...], *, allow_abbrev: bool + command: list[str], + option_strings: tuple[str, ...], + *, + allow_abbrev: bool, + other_values: frozenset[str] = frozenset(), ) -> list[tuple[int, str]]: """ Locate every value given for an option in a recorded command, in any accepted spelling. @@ -207,6 +211,11 @@ def _find_option_values( Every occurrence is returned: argparse keeps the last one, but all of them were typed. Abbreviations and short options are only matched with ``allow_abbrev``, i.e. when the parsed namespace confirms the command takes this option. + + A short option can also follow grouped flags (``-vpVALUE``, ``-vp VALUE``). The cluster is + walked until the option is found, unless the rest of the cluster is one of + ``other_values``, i.e. the parsed value of a preceding short option (``-uapple`` for a + username ``apple``). """ long_options = [o for o in option_strings if o.startswith("--")] short_options = [o for o in option_strings if not o.startswith("--")] if allow_abbrev else [] @@ -225,18 +234,38 @@ def _find_option_values( positions.append((idx, f"{name}=")) elif idx + 1 < len(command): positions.append((idx + 1, "")) - else: - for option in short_options: - if argument == option: - if idx + 1 < len(command): - positions.append((idx + 1, "")) - elif argument.startswith(f"{option}="): - positions.append((idx, f"{option}=")) - elif argument.startswith(option): - positions.append((idx, option)) + elif argument.startswith("-") and short_options: + prefix = _find_short_option_prefix(argument, short_options, other_values) + if prefix is None: + continue + if prefix: + positions.append((idx, prefix)) + elif idx + 1 < len(command): + positions.append((idx + 1, "")) return positions +def _find_short_option_prefix( + argument: str, short_options: list[str], other_values: frozenset[str] +) -> str | None: + """ + Return the part of a short-option token that precedes the option value. + + ``None`` means the token does not carry the option; an empty string means the value is + the next command element. + """ + for pos in range(1, len(argument)): + char = argument[pos] + if not char.isalnum() or (pos > 1 and argument[pos:] in other_values): + return None + if f"-{char}" in short_options: + rest = argument[pos + 1 :] + if not rest: + return "" + return argument[: pos + 1] + ("=" if rest.startswith("=") else "") + return None + + def _redact_json_argument(value: str) -> str: """ Redact sensitive entries in a JSON command-line argument. @@ -266,6 +295,10 @@ def _redact_uri_argument(value: str) -> str: """ from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse + if value.count("://") > 1: + # ``Connection`` accepts a protocol in the host: ``http://https://user:pass@host``. + scheme, separator, rest = value.partition("://") + return f"{scheme}{separator}{_redact_uri_argument(rest)}" try: parsed_uri = urlparse(value) has_password = parsed_uri.password is not None @@ -286,7 +319,8 @@ def _redact_uri_argument(value: str) -> str: elif _should_hide_key(key): redacted_params.append((key, MASKED_VALUE)) else: - redacted_params.append((key, item)) + # ``Connection`` parses every query value as JSON into ``extra``. + redacted_params.append((key, _redact_structure(item))) if redacted_params != params: query = urlencode(redacted_params, safe="*") @@ -310,6 +344,10 @@ def _mask_value_token(argument: str, value: str, redact) -> str: return argument +def _get_other_string_values(parsed: dict, dest: str) -> frozenset[str]: + return frozenset(v for k, v in parsed.items() if k != dest and isinstance(v, str) and v) + + def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: """ Mask secrets supplied through any supported argument spelling in the recorded command line. @@ -320,25 +358,36 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: parsed = vars(namespace) if isinstance(namespace, Namespace) else {} full_command = list(full_command) + # argparse stores a default for every option of the command, so presence in the namespace, + # not the truthiness of the last value, tells whether the command takes the option: a final + # empty value (``--password ''``) must not unmask the earlier occurrences. for dest, option_strings in _SECRET_OPTIONS.items(): - value = parsed.get(dest) - if not isinstance(value, str) or not value: + if dest not in parsed: continue - for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=True): + value = parsed[dest] + for idx, prefix in _find_option_values( + full_command, + option_strings, + allow_abbrev=True, + other_values=_get_other_string_values(parsed, dest), + ): full_command[idx] = prefix + MASKED_VALUE - full_command = [ - _mask_value_token(argument, value, lambda _: MASKED_VALUE) for argument in full_command - ] + if isinstance(value, str) and value: + full_command = [ + _mask_value_token(argument, value, lambda _: MASKED_VALUE) for argument in full_command + ] for options, redact in ((_JSON_OPTIONS, _redact_json_argument), (_URI_OPTIONS, _redact_uri_argument)): for dest, option_strings in options.items(): value = parsed.get(dest) - # Some of these options parse their JSON on the way in (``--env-vars``), so a dict or - # list in the namespace also means the command takes the option. - in_namespace = isinstance(value, (str, dict, list)) and bool(value) - for idx, prefix in _find_option_values(full_command, option_strings, allow_abbrev=in_namespace): + for idx, prefix in _find_option_values( + full_command, + option_strings, + allow_abbrev=dest in parsed, + other_values=_get_other_string_values(parsed, dest), + ): full_command[idx] = prefix + redact(full_command[idx][len(prefix) :]) - if in_namespace and isinstance(value, str): + if isinstance(value, str) and value: full_command = [_mask_value_token(argument, value, redact) for argument in full_command] return full_command diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 69af2d4de2dec..25c0a1aaa1d42 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -597,6 +597,82 @@ def test_cli_create_user_supplied_password_is_masked( ["airflow", "tasks", "test", "d1", "t1", "--env-v", '{"API_TOKEN": "********"}'], id="tasks-test-env-vars-abbreviated", ), + pytest.param( + ["airflow", "users", "reset-password", "-u", "a", "-vpFIRST", "-vpS3CRET"], + {"username": "a", "verbose": True, "password": "S3CRET"}, + ["airflow", "users", "reset-password", "-u", "a", f"-vp{MASKED}", f"-vp{MASKED}"], + id="grouped-short-flags-repeated", + ), + pytest.param( + ["airflow", "users", "reset-password", "-u", "a", "-vp", "FIRST", "-p", "S3CRET"], + {"username": "a", "verbose": True, "password": "S3CRET"}, + ["airflow", "users", "reset-password", "-u", "a", "-vp", MASKED, "-p", MASKED], + id="grouped-short-flags-separate-value-repeated", + ), + pytest.param( + ["airflow", "users", "create", "-uapple", "-pS3CRET"], + {"username": "apple", "password": "S3CRET"}, + ["airflow", "users", "create", "-uapple", f"-p{MASKED}"], + id="short-option-value-containing-secret-option-letter", + ), + pytest.param( + ["airflow", "users", "create", "-u", "a", "--pass", "FIRST", "--password", ""], + {"username": "a", "password": ""}, + ["airflow", "users", "create", "-u", "a", "--pass", MASKED, "--password", MASKED], + id="password-repeated-last-empty", + ), + pytest.param( + [ + "airflow", + "tasks", + "test", + "d1", + "t1", + "--env-v", + '{"API_TOKEN": "FIRST"}', + "--env-v", + "{}", + ], + {"dag_id": "d1", "task_id": "t1", "env_vars": {}}, + [ + "airflow", + "tasks", + "test", + "d1", + "t1", + "--env-v", + f'{{"API_TOKEN": "{MASKED}"}}', + "--env-v", + "{}", + ], + id="env-vars-repeated-last-empty", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "http://https://u:S3CRET@h:443/p"], + {"conn_id": "c1", "conn_uri": "http://https://u:S3CRET@h:443/p"}, + ["airflow", "connections", "add", "c1", "--conn-uri", f"http://https://u:{MASKED}@h:443/p"], + id="conn-uri-protocol-in-host", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + "http://h/?nested=%7B%22password%22%3A%22S3CRET%22%7D", + ], + {"conn_id": "c1", "conn_uri": "http://h/?nested=%7B%22password%22%3A%22S3CRET%22%7D"}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + f"http://h/?nested=%7B%22password%22%3A+%22{MASKED}%22%7D", + ], + id="conn-uri-json-query-parameter", + ), # ``-p`` that does not parse into a password is recorded as given. pytest.param( ["airflow", "api-server", "-p", "8080"], From 54c15b2099d3416d2a5d3f48d36c583f26017ae5 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 17 Oct 2026 01:57:13 +0200 Subject: [PATCH 07/13] Only treat a URI as double-scheme when the host carries a protocol Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 7 ++-- .../tests/unit/utils/test_cli_util.py | 33 +++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 45c47ef9f207c..444491b856c5d 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -295,9 +295,10 @@ def _redact_uri_argument(value: str) -> str: """ from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse - if value.count("://") > 1: - # ``Connection`` accepts a protocol in the host: ``http://https://user:pass@host``. - scheme, separator, rest = value.partition("://") + scheme, separator, rest = value.partition("://") + # ``Connection`` accepts a protocol in the host: ``http://https://user:pass@host``. A ``://`` + # elsewhere (``?next=http://...``) does not make the URI double-scheme. + if separator and re.match(r"[A-Za-z][A-Za-z0-9+.-]*://", rest): return f"{scheme}{separator}{_redact_uri_argument(rest)}" try: parsed_uri = urlparse(value) diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 25c0a1aaa1d42..69f25481e8ac2 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -653,6 +653,39 @@ def test_cli_create_user_supplied_password_is_masked( ["airflow", "connections", "add", "c1", "--conn-uri", f"http://https://u:{MASKED}@h:443/p"], id="conn-uri-protocol-in-host", ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "postgres://u:S3CRET@h/db?x=http://y"], + {"conn_id": "c1", "conn_uri": "postgres://u:S3CRET@h/db?x=http://y"}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + f"postgres://u:{MASKED}@h/db?x=http://y", + ], + id="conn-uri-url-in-query", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + "http://https://u:S3CRET@h/p?next=http://z", + ], + {"conn_id": "c1", "conn_uri": "http://https://u:S3CRET@h/p?next=http://z"}, + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + f"http://https://u:{MASKED}@h/p?next=http://z", + ], + id="conn-uri-protocol-in-host-and-url-in-query", + ), pytest.param( [ "airflow", From 59b9570c1c9ca8a52ea934e3b4bc93acaa1ac2ed Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 17 Oct 2026 02:01:17 +0200 Subject: [PATCH 08/13] Match Connection's double-scheme parsing when masking URIs Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 14 ++++++++++++++ airflow-core/tests/unit/utils/test_cli_util.py | 18 ++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 444491b856c5d..1006f3412bd98 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -285,6 +285,15 @@ def _redact_json_argument(value: str) -> str: return json.dumps(_redact_structure(parsed)) +_URL_BYTES_TO_REMOVE = str.maketrans("", "", "\t\r\n") +_C0_CONTROL_OR_SPACE = "".join(chr(i) for i in range(0x21)) + + +def _normalize_url_like_urlsplit(value: str) -> str: + """Drop the characters ``urlsplit`` ignores, so the inner URI is seen as ``Connection`` sees it.""" + return value.translate(_URL_BYTES_TO_REMOVE).lstrip(_C0_CONTROL_OR_SPACE) + + def _redact_uri_argument(value: str) -> str: """ Redact the secrets a URI command-line argument can carry. @@ -295,7 +304,9 @@ def _redact_uri_argument(value: str) -> str: """ from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse + value = _normalize_url_like_urlsplit(value) scheme, separator, rest = value.partition("://") + rest = _normalize_url_like_urlsplit(rest) # ``Connection`` accepts a protocol in the host: ``http://https://user:pass@host``. A ``://`` # elsewhere (``?next=http://...``) does not make the URI double-scheme. if separator and re.match(r"[A-Za-z][A-Za-z0-9+.-]*://", rest): @@ -306,6 +317,9 @@ def _redact_uri_argument(value: str) -> str: except ValueError: return MASKED_VALUE netloc = parsed_uri.netloc + # Fail closed on credentials the parse did not recognise (``u:pa/ss@h`` splits the netloc). + if not has_password and re.search(r":[^@]*@", netloc + parsed_uri.path): + return MASKED_VALUE if has_password: userinfo, _, host = netloc.rpartition("@") netloc = f"{userinfo.partition(':')[0]}:{MASKED_VALUE}@{host}" diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 69f25481e8ac2..c2add6c2481b3 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -686,6 +686,24 @@ def test_cli_create_user_supplied_password_is_masked( ], id="conn-uri-protocol-in-host-and-url-in-query", ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "http:// https://u:S3CRET@h"], + {"conn_id": "c1", "conn_uri": "http:// https://u:S3CRET@h"}, + ["airflow", "connections", "add", "c1", "--conn-uri", f"http://https://u:{MASKED}@h"], + id="conn-uri-protocol-in-host-after-space", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "http://\thttps://u:S3CRET@h"], + {"conn_id": "c1", "conn_uri": "http://\thttps://u:S3CRET@h"}, + ["airflow", "connections", "add", "c1", "--conn-uri", f"http://https://u:{MASKED}@h"], + id="conn-uri-protocol-in-host-after-tab", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "postgres://u:S3CRET/x@h/db"], + {"conn_id": "c1", "conn_uri": "postgres://u:S3CRET/x@h/db"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-unparsed-credentials", + ), pytest.param( [ "airflow", From 5b0470a45325e0b967afcfaeb18fe48566eb16b9 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 17 Oct 2026 02:28:10 +0200 Subject: [PATCH 09/13] Walk short-option clusters by option arity and fail closed on any unparsed credentials Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 60 ++++++++++++------- .../tests/unit/utils/test_cli_util.py | 24 ++++++++ 2 files changed, 64 insertions(+), 20 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 1006f3412bd98..50c22b4d23bd0 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -162,6 +162,18 @@ def wrapper(*args, **kwargs): "conn_uri": ("--conn-uri",), "broker_api": ("-a", "--broker-api"), } +# Short options of the commands that take one of the short options above, as +# ``(boolean flags, options taking a value)``, so a cluster such as ``-vpVALUE`` is walked the +# way argparse reads it. Keep in sync with the command definitions in ``cli_config`` and the +# FAB and Celery providers. Commands missing here are masked conservatively. +_SHORT_OPTION_ARITY: dict[tuple[str, str], tuple[str, str]] = { + ("users", "create"): ("v", "eflpru"), + ("users", "reset-password"): ("v", "epu"), + ("celery", "flower"): ("Dv", "AHaclpu"), + ("dags", "trigger"): ("v", "clor"), + ("dags", "test"): ("v", "Bcf"), + ("tasks", "test"): ("mv", "Bt"), +} def _should_hide_key(key) -> bool: @@ -198,7 +210,7 @@ def _find_option_values( option_strings: tuple[str, ...], *, allow_abbrev: bool, - other_values: frozenset[str] = frozenset(), + short_option_arity: tuple[str, str] | None = None, ) -> list[tuple[int, str]]: """ Locate every value given for an option in a recorded command, in any accepted spelling. @@ -212,10 +224,11 @@ def _find_option_values( Abbreviations and short options are only matched with ``allow_abbrev``, i.e. when the parsed namespace confirms the command takes this option. - A short option can also follow grouped flags (``-vpVALUE``, ``-vp VALUE``). The cluster is - walked until the option is found, unless the rest of the cluster is one of - ``other_values``, i.e. the parsed value of a preceding short option (``-uapple`` for a - username ``apple``). + A short option can also follow grouped flags (``-vpVALUE``, ``-vp VALUE``). With the + command's ``short_option_arity`` the cluster is walked as argparse does: boolean flags are + skipped and any other option ends the cluster by consuming its rest (``-uapple`` is a + username). Without it, every character before the option is taken as a flag, which may + mask more than the value but never less. """ long_options = [o for o in option_strings if o.startswith("--")] short_options = [o for o in option_strings if not o.startswith("--")] if allow_abbrev else [] @@ -235,7 +248,7 @@ def _find_option_values( elif idx + 1 < len(command): positions.append((idx + 1, "")) elif argument.startswith("-") and short_options: - prefix = _find_short_option_prefix(argument, short_options, other_values) + prefix = _find_short_option_prefix(argument, short_options, short_option_arity) if prefix is None: continue if prefix: @@ -246,7 +259,7 @@ def _find_option_values( def _find_short_option_prefix( - argument: str, short_options: list[str], other_values: frozenset[str] + argument: str, short_options: list[str], short_option_arity: tuple[str, str] | None ) -> str | None: """ Return the part of a short-option token that precedes the option value. @@ -256,13 +269,14 @@ def _find_short_option_prefix( """ for pos in range(1, len(argument)): char = argument[pos] - if not char.isalnum() or (pos > 1 and argument[pos:] in other_values): - return None if f"-{char}" in short_options: rest = argument[pos + 1 :] if not rest: return "" return argument[: pos + 1] + ("=" if rest.startswith("=") else "") + is_flag = char in short_option_arity[0] if short_option_arity else char.isalnum() + if not is_flag: + return None return None @@ -317,16 +331,23 @@ def _redact_uri_argument(value: str) -> str: except ValueError: return MASKED_VALUE netloc = parsed_uri.netloc - # Fail closed on credentials the parse did not recognise (``u:pa/ss@h`` splits the netloc). - if not has_password and re.search(r":[^@]*@", netloc + parsed_uri.path): + query = parsed_uri.query + params = parse_qsl(query, keep_blank_values=True) if query else [] + # Fail closed on ``user:pass@`` the parse did not take as credentials: one ``u:pa/ss@h`` splits + # across netloc and path, or one sits after the authority. ``__extra__`` is redacted as JSON. + unparsed_parts = [ + parsed_uri.path if has_password else netloc + parsed_uri.path, + parsed_uri.params, + parsed_uri.fragment, + *(f"{key}={item}" for key, item in params if key != "__extra__"), + ] + if any(re.search(r":[^@]*@", part) for part in unparsed_parts): return MASKED_VALUE if has_password: userinfo, _, host = netloc.rpartition("@") netloc = f"{userinfo.partition(':')[0]}:{MASKED_VALUE}@{host}" - query = parsed_uri.query - if query: - params = parse_qsl(query, keep_blank_values=True) + if params: redacted_params = [] for key, item in params: if key == "__extra__": @@ -359,10 +380,6 @@ def _mask_value_token(argument: str, value: str, redact) -> str: return argument -def _get_other_string_values(parsed: dict, dest: str) -> frozenset[str]: - return frozenset(v for k, v in parsed.items() if k != dest and isinstance(v, str) and v) - - def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: """ Mask secrets supplied through any supported argument spelling in the recorded command line. @@ -372,6 +389,9 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: """ parsed = vars(namespace) if isinstance(namespace, Namespace) else {} full_command = list(full_command) + short_option_arity = ( + _SHORT_OPTION_ARITY.get((full_command[1], full_command[2])) if len(full_command) > 2 else None + ) # argparse stores a default for every option of the command, so presence in the namespace, # not the truthiness of the last value, tells whether the command takes the option: a final @@ -384,7 +404,7 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: full_command, option_strings, allow_abbrev=True, - other_values=_get_other_string_values(parsed, dest), + short_option_arity=short_option_arity, ): full_command[idx] = prefix + MASKED_VALUE if isinstance(value, str) and value: @@ -399,7 +419,7 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: full_command, option_strings, allow_abbrev=dest in parsed, - other_values=_get_other_string_values(parsed, dest), + short_option_arity=short_option_arity, ): full_command[idx] = prefix + redact(full_command[idx][len(prefix) :]) if isinstance(value, str) and value: diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index c2add6c2481b3..d92923302db6e 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -704,6 +704,30 @@ def test_cli_create_user_supplied_password_is_masked( ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], id="conn-uri-unparsed-credentials", ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "http://u:pw@h/db#u:S3CRET@h2"], + {"conn_id": "c1", "conn_uri": "http://u:pw@h/db#u:S3CRET@h2"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-credentials-in-fragment", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "http://h/db?target=u:S3CRET@h2"], + {"conn_id": "c1", "conn_uri": "http://h/db?target=u:S3CRET@h2"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-credentials-in-query-value", + ), + pytest.param( + ["airflow", "users", "reset-password", "-u", "p", "-vp", "FIRST", "-p", "S3CRET"], + {"username": "p", "verbose": True, "password": "S3CRET"}, + ["airflow", "users", "reset-password", "-u", "p", "-vp", MASKED, "-p", MASKED], + id="grouped-short-flags-username-equals-cluster-tail", + ), + pytest.param( + ["airflow", "users", "reset-password", "-u", "p", "-vpFIRST", "-p", "S3CRET"], + {"username": "p", "verbose": True, "password": "S3CRET"}, + ["airflow", "users", "reset-password", "-u", "p", f"-vp{MASKED}", "-p", MASKED], + id="grouped-short-flags-attached-username-p", + ), pytest.param( [ "airflow", From 6b35f4a01b24e2586bc546c7f79ab8dedbda7679 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 17 Oct 2026 02:37:34 +0200 Subject: [PATCH 10/13] Mask the whole URI when credentials cannot be located reliably Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 18 ++++++++---------- airflow-core/tests/unit/utils/test_cli_util.py | 12 ++++++++++++ 2 files changed, 20 insertions(+), 10 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 50c22b4d23bd0..71f15900ddc5b 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -331,18 +331,16 @@ def _redact_uri_argument(value: str) -> str: except ValueError: return MASKED_VALUE netloc = parsed_uri.netloc + # A ``:`` followed later by ``@`` that the parse did not take as the credentials may still be + # one (``u:pa?ss@h`` splits it into the query), so the whole value is masked. This errs on the + # side of masking: ``user@host:port`` is kept, anything shaped like ``user:pass@`` is not. + after_authority = urlunparse( + ("", "", parsed_uri.path, parsed_uri.params, parsed_uri.query, parsed_uri.fragment) + ) + if re.search(r":.*@", after_authority if has_password else rest if separator else value): + return MASKED_VALUE query = parsed_uri.query params = parse_qsl(query, keep_blank_values=True) if query else [] - # Fail closed on ``user:pass@`` the parse did not take as credentials: one ``u:pa/ss@h`` splits - # across netloc and path, or one sits after the authority. ``__extra__`` is redacted as JSON. - unparsed_parts = [ - parsed_uri.path if has_password else netloc + parsed_uri.path, - parsed_uri.params, - parsed_uri.fragment, - *(f"{key}={item}" for key, item in params if key != "__extra__"), - ] - if any(re.search(r":[^@]*@", part) for part in unparsed_parts): - return MASKED_VALUE if has_password: userinfo, _, host = netloc.rpartition("@") netloc = f"{userinfo.partition(':')[0]}:{MASKED_VALUE}@{host}" diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index d92923302db6e..79f7bf10904a5 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -716,6 +716,18 @@ def test_cli_create_user_supplied_password_is_masked( ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], id="conn-uri-credentials-in-query-value", ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "postgres://u:S3CRET?tail@h/db"], + {"conn_id": "c1", "conn_uri": "postgres://u:S3CRET?tail@h/db"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-password-split-by-question-mark", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "postgres://u:S3CRET#tail@h/db"], + {"conn_id": "c1", "conn_uri": "postgres://u:S3CRET#tail@h/db"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-password-split-by-hash", + ), pytest.param( ["airflow", "users", "reset-password", "-u", "p", "-vp", "FIRST", "-p", "S3CRET"], {"username": "p", "verbose": True, "password": "S3CRET"}, From 2da94c6aeb93266fbd8eef281429edf8823383ea Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 17 Oct 2026 02:40:40 +0200 Subject: [PATCH 11/13] Detect percent-encoded credentials when masking URIs Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 6 +++-- .../tests/unit/utils/test_cli_util.py | 22 +++++++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 71f15900ddc5b..c5bf05532ff0d 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -316,7 +316,7 @@ def _redact_uri_argument(value: str) -> str: (``?api_key=...``) and sensitive entries of the URL-encoded ``__extra__`` JSON that connection URIs accept. """ - from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse + from urllib.parse import parse_qsl, unquote, urlencode, urlparse, urlunparse value = _normalize_url_like_urlsplit(value) scheme, separator, rest = value.partition("://") @@ -337,7 +337,9 @@ def _redact_uri_argument(value: str) -> str: after_authority = urlunparse( ("", "", parsed_uri.path, parsed_uri.params, parsed_uri.query, parsed_uri.fragment) ) - if re.search(r":.*@", after_authority if has_password else rest if separator else value): + unlocated = after_authority if has_password else rest if separator else value + # ``Connection`` decodes query values, so ``u%3Apass%40host`` is a credential too. + if any(re.search(r":.*@", text) for text in (unlocated, unquote(unlocated))): return MASKED_VALUE query = parsed_uri.query params = parse_qsl(query, keep_blank_values=True) if query else [] diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 79f7bf10904a5..2239c13c9a2a3 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -728,6 +728,28 @@ def test_cli_create_user_supplied_password_is_masked( ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], id="conn-uri-password-split-by-hash", ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "postgres://h/db?x=u%3AS3CRET%40h2"], + {"conn_id": "c1", "conn_uri": "postgres://h/db?x=u%3AS3CRET%40h2"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-percent-encoded-credentials-in-query", + ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + "http://h/?__extra__=%7B%22token%22%3A%22S3CRET%22%2C%22email%22%3A%22a%40b.c%22%7D", + ], + { + "conn_id": "c1", + "conn_uri": "http://h/?__extra__=%7B%22token%22%3A%22S3CRET%22%2C%22email%22%3A%22a%40b.c%22%7D", + }, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-encoded-extra-with-email", + ), pytest.param( ["airflow", "users", "reset-password", "-u", "p", "-vp", "FIRST", "-p", "S3CRET"], {"username": "p", "verbose": True, "password": "S3CRET"}, From f4c4ec64da839e9fecab6f720f3988dda612a79e Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 17 Oct 2026 02:44:57 +0200 Subject: [PATCH 12/13] Match credentials across newlines when masking URIs Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 2 +- .../tests/unit/utils/test_cli_util.py | 19 +++++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index c5bf05532ff0d..87d76059a8aaa 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -339,7 +339,7 @@ def _redact_uri_argument(value: str) -> str: ) unlocated = after_authority if has_password else rest if separator else value # ``Connection`` decodes query values, so ``u%3Apass%40host`` is a credential too. - if any(re.search(r":.*@", text) for text in (unlocated, unquote(unlocated))): + if any(re.search(r":.*@", text, re.DOTALL) for text in (unlocated, unquote(unlocated))): return MASKED_VALUE query = parsed_uri.query params = parse_qsl(query, keep_blank_values=True) if query else [] diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 2239c13c9a2a3..1802128c78882 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -734,6 +734,25 @@ def test_cli_create_user_supplied_password_is_masked( ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], id="conn-uri-percent-encoded-credentials-in-query", ), + pytest.param( + [ + "airflow", + "connections", + "add", + "c1", + "--conn-uri", + "http://h/db?target=u%3AS3CRET%0A%40h2", + ], + {"conn_id": "c1", "conn_uri": "http://h/db?target=u%3AS3CRET%0A%40h2"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-percent-encoded-newline-in-credentials", + ), + pytest.param( + ["airflow", "connections", "add", "c1", "--conn-uri", "http://h/db?target=u:S3CRET\r\n@h2"], + {"conn_id": "c1", "conn_uri": "http://h/db?target=u:S3CRET\r\n@h2"}, + ["airflow", "connections", "add", "c1", "--conn-uri", MASKED], + id="conn-uri-raw-newline-in-credentials", + ), pytest.param( [ "airflow", From ce5d9f1d9fa0154f8ef5dd6d0a5aa9bf69839cc0 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Mon, 19 Oct 2026 17:10:35 +0200 Subject: [PATCH 13/13] Keep CLI commands running when audit-log masking fails The command line is masked for the audit log before the command runs, so an unexpected error in the masking aborted every CLI command. A failure now records the command fully masked instead: the command still runs, and nothing a secret could be in reaches the log. The short-option table is also checked against the real parsers, so a command definition change cannot silently change what gets masked. Generated-by: Claude Opus 5 --- airflow-core/src/airflow/utils/cli.py | 41 +++++++++----- .../tests/unit/utils/test_cli_util.py | 53 +++++++++++++++++++ 2 files changed, 80 insertions(+), 14 deletions(-) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index 87d76059a8aaa..5725bc58387a2 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -428,18 +428,8 @@ def _mask_parsed_arguments(full_command: list[str], namespace) -> list[str]: return full_command -def _build_metrics(func_name, namespace): - """ - Build metrics dict from function args. - - It assumes that function arguments is from airflow.bin.cli module's function - and has Namespace instance where it optionally contains "dag_id", "task_id", - "logical_date", and a team name. - - :param func_name: name of function - :param namespace: Namespace instance from argparse - :return: dict with metrics - """ +def _mask_full_command(full_command: list[str], namespace) -> list[str]: + """Mask the secrets in the command line recorded in the audit log.""" from airflow._shared.secrets_masker import _secrets_masker sub_commands_to_check_for_sensitive_fields = {"users", "connections"} @@ -449,7 +439,7 @@ def _build_metrics(func_name, namespace): # providers (``keycloak-auth-manager ... --password``), so they are masked everywhere. ``-p`` # is only a password on the subcommands above; elsewhere it is e.g. ``--port``. always_sensitive_fields = {"--password", "--conn-password"} - full_command = list(sys.argv) + full_command = list(full_command) sub_command = full_command[1] if len(full_command) > 1 else None # For cases when value under sub_commands_to_check_for_sensitive_key have sensitive info if sub_command in sub_commands_to_check_for_sensitive_key: @@ -480,7 +470,30 @@ def _build_metrics(func_name, namespace): # Options carrying secrets (on their own, or inside JSON and URIs) are masked by their # parsed value, whatever spelling was used for them on the command line. - full_command = _mask_parsed_arguments(full_command, namespace) + return _mask_parsed_arguments(full_command, namespace) + + +def _build_metrics(func_name, namespace): + """ + Build metrics dict from function args. + + It assumes that function arguments is from airflow.bin.cli module's function + and has Namespace instance where it optionally contains "dag_id", "task_id", + "logical_date", and a team name. + + :param func_name: name of function + :param namespace: Namespace instance from argparse + :return: dict with metrics + """ + try: + full_command = _mask_full_command(sys.argv, namespace) + except Exception as e: + # This runs before the command itself, so a masking bug must not abort the command, and + # must not record anything a secret could be in; the error text could quote it too. + logger.warning( + "Could not mask the command line for the audit log (%s); recording it masked", type(e).__name__ + ) + full_command = [*sys.argv[:1], MASKED_VALUE] metrics = { "sub_command": func_name, diff --git a/airflow-core/tests/unit/utils/test_cli_util.py b/airflow-core/tests/unit/utils/test_cli_util.py index 1802128c78882..1626910fbca3b 100644 --- a/airflow-core/tests/unit/utils/test_cli_util.py +++ b/airflow-core/tests/unit/utils/test_cli_util.py @@ -17,6 +17,7 @@ # under the License. from __future__ import annotations +import argparse import ast import json import os @@ -855,6 +856,53 @@ def test_short_secret_is_not_masked_inside_other_arguments(self): "********", ] + @mock.patch.object(cli, "_mask_parsed_arguments", side_effect=RuntimeError("cannot mask S3CRET")) + @mock.patch.object(cli_action_loggers, "on_pre_execution", autospec=True) + def test_masking_failure_does_not_abort_the_command_or_leak_the_secret( + self, mock_on_pre_execution, _mock_mask + ): + command = ["airflow", "users", "create", "-u", "alice", "--password", "S3CRET"] + with mock.patch.object(sys, "argv", command): + assert returns_ran(Namespace(username="alice", password="S3CRET")) == "ran" + + recorded = mock_on_pre_execution.call_args.kwargs["full_command"] + assert ast.literal_eval(recorded) == ["airflow", "********"] + assert "S3CRET" not in recorded + + def test_short_option_table_matches_the_parsers(self): + """The short-option table must follow the command definitions, or masking drifts silently.""" + from airflow.cli import cli_parser + + def get_subparsers(parser): + for action in parser._actions: + if isinstance(action, argparse._SubParsersAction): + return action.choices + return {} + + masked_dests = {*cli._SECRET_OPTIONS, *cli._JSON_OPTIONS, *cli._URI_OPTIONS} + groups = get_subparsers(cli_parser.get_parser()) + actual = {} + for group, group_parser in groups.items(): + for command, command_parser in get_subparsers(group_parser).items(): + short = { + option[1]: action + for action in command_parser._actions + for option in action.option_strings + if len(option) == 2 and option != "-h" + } + if any(action.dest in masked_dests for action in short.values()): + flags = "".join(sorted(o for o, a in short.items() if a.nargs == 0)) + values = "".join(sorted(o for o, a in short.items() if a.nargs != 0)) + actual[(group, command)] = (flags, values) + + # Provider command groups (``users``, ``celery``) only exist when the provider is installed. + expected = { + key: ("".join(sorted(flags)), "".join(sorted(values))) + for key, (flags, values) in cli._SHORT_OPTION_ARITY.items() + if key[0] in groups + } + assert expected == actual + def test_action_log_records_team_name(self, session): namespace = Namespace(team_name="payments") with ( @@ -1013,6 +1061,11 @@ def success_func(_): pass +@cli.action_cli(check_db=False) +def returns_ran(_): + return "ran" + + def test__search_for_dags_file(): dags_folder = settings.DAGS_FOLDER assert _search_for_dag_file("") is None