From 71ef2b1879606bf1be8cac3c9920a0ef6da0bdd2 Mon Sep 17 00:00:00 2001 From: Maximilien Cuony Date: Fri, 14 Aug 2026 17:16:14 +0200 Subject: [PATCH] [mockuss] Sanitize configuration logging --- monitoring/mock_uss/app.py | 95 ++++++++++++++++- monitoring/mock_uss/app_test.py | 176 ++++++++++++++++++++++++++++++++ monitoring/monitorlib/auth.py | 6 +- 3 files changed, 275 insertions(+), 2 deletions(-) create mode 100644 monitoring/mock_uss/app_test.py diff --git a/monitoring/mock_uss/app.py b/monitoring/mock_uss/app.py index 8e7f32379c..d09d034002 100644 --- a/monitoring/mock_uss/app.py +++ b/monitoring/mock_uss/app.py @@ -1,5 +1,6 @@ import inspect import os +import re from collections.abc import Callable from typing import Any @@ -132,10 +133,102 @@ def require_config_value(config_key: str) -> None: enabled_services.add(SERVICE_FLIGHT_PLANNING) from monitoring.mock_uss.flight_planning import routes as flight_planning_routes # noqa F401 + +_SECRET_NAME_RE = re.compile( + r"API|AUTH|TOKEN|KEY|SECRET|PASS|SIGNATURE|HTTP_COOKIE", + flags=re.I, +) +_SAFE_KEYS = ["MOCK_USS_PUBLIC_KEY", "MOCK_USS_TOKEN_AUDIENCE"] + +# Extract the userinfo part of an URI +_USERINFO_RE = re.compile(r"(?<=://)[^/@\s]*:[^/@\s]*@") + +_ARITY = { + "NoAuth": 2, + "InvalidTokenSignatureAuth": 1, + "DummyOAuth": 2, + "ServiceAccount": 2, + "ServiceAccountImpersonation": 2, + "SignedRequest": 6, + "UsernamePassword": 4, + "ClientIdClientSecret": 4, + "Keycloak": 3, + "FlightPassport": 4, +} + +_SECRETS = { + "UsernamePassword": (2, "password"), + "ClientIdClientSecret": (2, "client_secret"), + "Keycloak": (2, "client_secret"), + "FlightPassport": (2, "client_secret"), +} + + +def sanitize_secrets(key, value): + """Ensure value is free of sensitive values: + + * Authentication information is removed from URLs + * Key names are used to detect secrets and obfuscate the value + * MOCK_USS_AUTH_SPEC value secrets are removed + * Unrecognized elements are escaped as a safe fallback""" + + if key != "MOCK_USS_AUTH_SPEC": # Non-auth spec case + if ( + key not in _SAFE_KEYS + and _SECRET_NAME_RE.search( # Name contains key/secret/etc... + key, + ) + ): + return "***" + + if not isinstance(value, str): # Value may be non-string + return value + + return _USERINFO_RE.sub("***@", value) + + from monitoring.monitorlib.auth import SPEC_RE # Loaded here due to circular import + + m = SPEC_RE.match(value) # Try to parse an AuthSpec + if m is None: + return "***" + + name, param_string = m.group(1), m.group(2) + params = [p.strip() for p in param_string.split(",")] + + if ( + name not in _ARITY or len(params) > _ARITY[name] + ): # Unknown adapter, we escape everything + hidden = [ + p.split("=", 1)[0].strip() + "=***" if "=" in p else "***" for p in params + ] + return "{}({})".format(name, ", ".join(hidden)) + + pos_index, kwarg = _SECRETS.get(name, (None, None)) + out = [] + pos = 0 + for p in params: # For each parameter + if "=" in p: # Named parameter + k, v = p.split("=", 1) + k = k.strip() + out.append( + k + "=***" + if k == kwarg + else k + "=" + _USERINFO_RE.sub("***@", v.strip()) + ) + else: # Positional parameter + out.append("***" if pos == pos_index else _USERINFO_RE.sub("***@", p)) + pos += 1 + + return "{}({})".format(name, ", ".join(out)) + + msg = ( "################################################################################\n" + "################################ Configuration ################################\n" - + "\n".join(f"## {key}: {webapp.config[key]}" for key in webapp.config) + + "\n".join( + f"## {key}: {sanitize_secrets(key, webapp.config[key])}" + for key in webapp.config + ) + "\n" + "################################################################################" ) diff --git a/monitoring/mock_uss/app_test.py b/monitoring/mock_uss/app_test.py new file mode 100644 index 0000000000..5af0356f0e --- /dev/null +++ b/monitoring/mock_uss/app_test.py @@ -0,0 +1,176 @@ +import pytest + +from monitoring.mock_uss.app import sanitize_secrets + +K = "MOCK_USS_AUTH_SPEC" + + +@pytest.mark.parametrize( + "key", + [ + "API_URL", + "AUTH_SPEC", + "TOKEN_ENDPOINT", + "KEY_PATH", + "SECRET_KEY", + "PASSWORD", + "PASS", + "SIGNATURE_STYLE", + "HTTP_COOKIE", + "secret_key", + "api_key", + "DB_PASSWORD", + "USS_AUTH_TOKEN", + "MY_SECRET", + "MY_API", + "X_AUTH_SPEC", + ], +) +def test_sensitive_key_hidden(key): + assert sanitize_secrets(key, "randomvalue") == "***" + + +@pytest.mark.parametrize("key", ["PORT", "DEBUG", "USS_QUALIFIER_URL"]) +def test_insensitive_key_untouched(key): + assert sanitize_secrets(key, "randomvalue") == "randomvalue" + + +@pytest.mark.parametrize( + "value,expected", + [ + ("NoAuth(sub=uss1)", "NoAuth(sub=uss1)"), + ( + "InvalidTokenSignatureAuth(uss_unsigned)", + "InvalidTokenSignatureAuth(uss_unsigned)", + ), + ( + "DummyOAuth(http://oauth.authority.localutm:8085/token,uss2)", + "DummyOAuth(http://oauth.authority.localutm:8085/token, uss2)", + ), + ( + "ServiceAccount(http://host/token,/secrets/sa.json)", + "ServiceAccount(http://host/token, /secrets/sa.json)", + ), + ( + "ServiceAccountImpersonation(http://host/token,sa@example.com)", + "ServiceAccountImpersonation(http://host/token, sa@example.com)", + ), + ( + "SignedRequest(http://host/token,client,/keys/k.pem,https://host/c.crt)", + "SignedRequest(http://host/token, client, /keys/k.pem, https://host/c.crt)", + ), + ], +) +def test_no_secret_untouched(value, expected): + assert sanitize_secrets(K, value) == expected + + +def test_username_password_positional(): + assert ( + sanitize_secrets(K, "UsernamePassword(http://host/token,alice,hunter2,cli)") + == "UsernamePassword(http://host/token, alice, ***, cli)" + ) + + +def test_username_password_kwarg(): + assert ( + sanitize_secrets( + K, + "UsernamePassword(http://host/token,alice,client_id=cli,password=hunter2)", + ) + == "UsernamePassword(http://host/token, alice, client_id=cli, password=***)" + ) + + +@pytest.mark.parametrize("name", ["ClientIdClientSecret", "Keycloak", "FlightPassport"]) +def test_client_secret_positional(name): + assert ( + sanitize_secrets(K, f"{name}(http://host/token,cli,s3cr3t)") + == f"{name}(http://host/token, cli, ***)" + ) + + +def test_client_secret_kwarg_out_of_order(): + assert ( + sanitize_secrets( + K, + "Keycloak(client_secret=s3cr3t,token_endpoint=http://host/token,client_id=cli)", + ) + == "Keycloak(client_secret=***, token_endpoint=http://host/token, client_id=cli)" + ) + + +def test_client_secret_mixed_with_trailing_positional(): + assert ( + sanitize_secrets( + K, + "ClientIdClientSecret(http://host/token,cli,s3cr3t,send_request_as_data=true)", + ) + == "ClientIdClientSecret(http://host/token, cli, ***, send_request_as_data=true)" + ) + + +def test_unknown_adapter_hides_everything(): + assert ( + sanitize_secrets(K, "MysteryAuth(http://host/token,cli,whatever,foo=bar)") + == "MysteryAuth(***, ***, ***, foo=***)" + ) + + +@pytest.mark.parametrize( + "value", + ["", "not a spec", "DummyOAuth(unclosed", "DummyOAuth(a) trailing"], +) +def test_malformed_hidden(value): + assert sanitize_secrets(K, value) == "***" + + +def test_other_key_untouched(): + assert sanitize_secrets( + "PORT", "UsernamePassword(http://host/token,alice,hunter2,cli)" + ) == ("UsernamePassword(http://host/token,alice,hunter2,cli)") + + +@pytest.mark.parametrize( + "value,expected", + [ + ( + "UsernamePassword(http://host/token,alice,pa,ss,cli)", + "UsernamePassword(***, ***, ***, ***, ***)", + ), + ("DummyOAuth(http://host/token,uss2,extra)", "DummyOAuth(***, ***, ***)"), + ("NoAuth(a,b,c)", "NoAuth(***, ***, ***)"), + ], +) +def test_too_many_params_hidden(value, expected): + assert sanitize_secrets(K, value) == expected + + +@pytest.mark.parametrize( + "value,expected", + [ + ( + "DummyOAuth(https://user:pw@host/token,uss2)", + "DummyOAuth(https://***@host/token, uss2)", + ), + ( + "Keycloak(token_endpoint=https://user:pw@host/token,client_id=cli,client_secret=s3cr3t)", + "Keycloak(token_endpoint=https://***@host/token, client_id=cli, client_secret=***)", + ), + ], +) +def test_url_userinfo_hidden(value, expected): + assert sanitize_secrets(K, value) == expected + + +@pytest.mark.parametrize( + "value,expected", + [ + ("https://user:pw@host/path", "https://***@host/path"), + ("http://user:pw@host:8085/token", "http://***@host:8085/token"), + ("https://host/path", "https://host/path"), + ("sa@example.com", "sa@example.com"), + ], +) +def test_userinfo_hidden_on_unfiltered_key(value, expected): + assert sanitize_secrets("USS_QUALIFIER_URL", value) == expected diff --git a/monitoring/monitorlib/auth.py b/monitoring/monitorlib/auth.py index 14de18748b..f1daa3befa 100644 --- a/monitoring/monitorlib/auth.py +++ b/monitoring/monitorlib/auth.py @@ -606,6 +606,9 @@ def all_subclasses(cls): ) +SPEC_RE = re.compile(r"^\s*([^\s(]+)\s*\(\s*([^)]*)\s*\)\s*$") + + def make_auth_adapter(spec: AuthSpec) -> AuthAdapter: """Make an AuthAdapter according to a string specification. @@ -621,7 +624,8 @@ def make_auth_adapter(spec: AuthSpec) -> AuthAdapter: An instance of the appropriate AuthAdapter subclass according to the provided spec. """ - m = re.match(r"^\s*([^\s(]+)\s*\(\s*([^)]*)\s*\)\s*$", spec) + + m = SPEC_RE.match(spec) if m is None: raise ValueError( "Auth adapter specification did not match the pattern `AdapterName(param, param, ...)`"