diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index f2031f7c93fe5..165c26c8b7297 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -138,76 +138,363 @@ def wrapper(*args, **kwargs): return action_logging -def _build_metrics(func_name, namespace): +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",), + "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, ...]] = { + "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: + 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): """ - Build metrics dict from function args. + Redact values whose key looks sensitive, at any depth. - 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. + 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 _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_values( + command: list[str], + option_strings: tuple[str, ...], + *, + allow_abbrev: bool, + 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. + + 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. + + 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 [] + positions: list[tuple[int, str]] = [] + for idx, argument in enumerate(command): + 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, "")) + elif argument.startswith("-") and short_options: + prefix = _find_short_option_prefix(argument, short_options, short_option_arity) + 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], short_option_arity: tuple[str, str] | None +) -> str | None: + """ + Return the part of a short-option token that precedes the option value. - :param func_name: name of function - :param namespace: Namespace instance from argparse - :return: dict with metrics + ``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 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 + + +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)) + + +_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. + + 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, unquote, 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): + return f"{scheme}{separator}{_redact_uri_argument(rest)}" + try: + parsed_uri = urlparse(value) + has_password = parsed_uri.password is not None + 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) + ) + 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, 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 [] + if has_password: + userinfo, _, host = netloc.rpartition("@") + netloc = f"{userinfo.partition(':')[0]}:{MASKED_VALUE}@{host}" + + if params: + 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: + # ``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="*") + + return urlunparse( + (parsed_uri.scheme, netloc, parsed_uri.path, parsed_uri.params, query, parsed_uri.fragment) + ) + + +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. + + 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) + 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 + # empty value (``--password ''``) must not unmask the earlier occurrences. + for dest, option_strings in _SECRET_OPTIONS.items(): + if dest not in parsed: + continue + value = parsed[dest] + for idx, prefix in _find_option_values( + full_command, + option_strings, + allow_abbrev=True, + short_option_arity=short_option_arity, + ): + full_command[idx] = prefix + MASKED_VALUE + 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) + for idx, prefix in _find_option_values( + full_command, + option_strings, + allow_abbrev=dest in parsed, + short_option_arity=short_option_arity, + ): + full_command[idx] = prefix + redact(full_command[idx][len(prefix) :]) + if isinstance(value, str) and value: + full_command = [_mask_value_token(argument, value, redact) for argument in full_command] + + return full_command + + +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"} sub_commands_to_check_for_sensitive_key = {"variables"} sensitive_fields = {"-p", "--password", "--conn-password"} - full_command = list(sys.argv) + # 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(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: - 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 - elif sub_command in sub_commands_to_check_for_sensitive_fields: + # 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] + 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] = "*" * 8 + 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}={'*' * 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: - from urllib.parse import urlparse, urlunparse - - uri_index = full_command.index("--conn-uri") + 1 - conn_uri = full_command[uri_index] - parsed_uri = urlparse(conn_uri) - netloc = parsed_uri.netloc - if parsed_uri.password: - password = "*" * 8 - netloc = f"{parsed_uri.username}:{password}@{parsed_uri.hostname}" - if parsed_uri.port: - netloc += f":{parsed_uri.port}" - - full_command[uri_index] = urlunparse( - ( - parsed_uri.scheme, - netloc, - parsed_uri.path, - parsed_uri.params, - parsed_uri.query, - parsed_uri.fragment, - ) + full_command[idx] = f"{sensitive_field}={MASKED_VALUE}" + + # 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. + 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 82dce4b668f7e..98cbad6fbcf3d 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 @@ -56,6 +57,7 @@ # - ``cli_action_loggers.on_post_execution`` pytestmark = pytest.mark.db_test repo_root = Path(airflow.__file__).parents[1] +MASKED = "*" * 8 _DEFERRED_BUNDLE_NAME = "deferred-path-bundle" _DEFERRED_BUNDLE_DAG_ID = "deferred_path_bundle_dag" @@ -228,6 +230,95 @@ def test_explicit_bundle_name_is_initialized(self, mock_manager_cls, tmp_path, g ], 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", @@ -238,6 +329,21 @@ def test_explicit_bundle_name_is_initialized(self, mock_manager_cls, tmp_path, g "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( @@ -270,6 +376,601 @@ 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", + ), + 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", "-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"}'}, + ["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", + ), + 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", "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", "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", "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", "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", "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/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", + "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"}, + ["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", + "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"], + {"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_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", + "********", + ] + + @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 ( @@ -329,27 +1030,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, @@ -394,6 +1129,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