[mockuss] Sanitize configuration logging

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"
     + "################################################################################"
 )

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

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, ...)`"