microsoft--magentic-ui/tests/test_log_scrub.py

"""``_scrub_for_log`` must recursively redact sensitive keys before logging.

The backend logger formats outgoing WebSocket frames as dicts; some carry
per-slot RFB passwords (``browser_address`` messages) or other secrets.
``_scrub_for_log`` walks the structure and replaces values of known-sensitive
keys with ``***REDACTED***`` regardless of nesting depth.
"""

from __future__ import annotations

from typing import Any

from magentic_ui.backend.web.managers.connection import _scrub_for_log


def test_top_level_password_redacted() -> None:
    result = _scrub_for_log({"password": "secret-token"})
    assert result == {"password": "***REDACTED***"}


def test_nested_password_redacted() -> None:
    msg = {"type": "message", "data": {"metadata": {"password": "secret-token"}}}
    result = _scrub_for_log(msg)
    assert result == {
        "type": "message",
        "data": {"metadata": {"password": "***REDACTED***"}},
    }


def test_deeply_nested_password_redacted() -> None:
    msg = {"a": {"b": {"c": {"d": {"password": "secret-token"}}}}}
    result = _scrub_for_log(msg)
    assert result == {"a": {"b": {"c": {"d": {"password": "***REDACTED***"}}}}}


def test_password_inside_list_of_dicts_redacted() -> None:
    msg = {
        "items": [
            {"password": "a"},
            {"password": "b"},
            {"other": "c"},
        ]
    }
    result = _scrub_for_log(msg)
    assert result == {
        "items": [
            {"password": "***REDACTED***"},
            {"password": "***REDACTED***"},
            {"other": "c"},
        ]
    }


def test_all_sensitive_keys_redacted() -> None:
    msg = {
        "password": "p",
        "token": "t",
        "api_key": "k",
        "secret": "s",
        "normal": "n",
    }
    result = _scrub_for_log(msg)
    assert result == {
        "password": "***REDACTED***",
        "token": "***REDACTED***",
        "api_key": "***REDACTED***",
        "secret": "***REDACTED***",
        "normal": "n",
    }


def test_case_insensitive_match() -> None:
    msg = {"PASSWORD": "x", "Token": "y", "API_KEY": "z"}
    result = _scrub_for_log(msg)
    assert result == {
        "PASSWORD": "***REDACTED***",
        "Token": "***REDACTED***",
        "API_KEY": "***REDACTED***",
    }


def test_non_sensitive_keys_unchanged() -> None:
    msg = {"username": "u", "port": 9999, "url": "http://x"}
    result = _scrub_for_log(msg)
    assert result == msg


def test_primitives_pass_through_unchanged() -> None:
    assert _scrub_for_log("hello") == "hello"
    assert _scrub_for_log(42) == 42
    assert _scrub_for_log(True) is True
    assert _scrub_for_log(None) is None
    assert _scrub_for_log([1, 2, 3]) == [1, 2, 3]


def test_empty_structures_pass_through() -> None:
    assert _scrub_for_log({}) == {}
    assert _scrub_for_log([]) == []


def test_real_browser_address_message_redacted() -> None:
    """Repro of the exact message shape the bug produced in logs."""
    msg: dict[str, Any] = {
        "type": "message",
        "data": {
            "source": "web_surfer",
            "content": [{"type": "text", "text": "Browser ready at noVNC port 59253"}],
            "metadata": {
                "source": "web_surfer",
                "type": "browser_address",
                "novnc_port": "59253",
                "playwright_port": "51305",
                "password": "secret-token",
            },
        },
    }
    result = _scrub_for_log(msg)
    metadata = result["data"]["metadata"]  # type: ignore[index]
    assert metadata["password"] == "***REDACTED***"
    assert metadata["novnc_port"] == "59253"  # unchanged
    assert metadata["type"] == "browser_address"  # unchanged


def test_original_input_not_mutated() -> None:
    msg = {"password": "secret-token"}
    _scrub_for_log(msg)
    assert msg == {"password": "secret-token"}  # original untouched