kernalix7--winpodx/tests/test_security.py

# SPDX-License-Identifier: MIT
"""Security-focused tests for winpodx."""

from __future__ import annotations

import logging
from pathlib import Path

import pytest

from winpodx.core.app import _SAFE_NAME_RE, load_app
from winpodx.core.config import Config, PodConfig, RDPConfig
from winpodx.core.rdp import _filter_extra_flags, linux_to_unc

# --- App name injection ---


class TestAppNameValidation:
    @pytest.mark.parametrize(
        "name",
        [
            "../etc/passwd",
            "../../root",
            "app;rm -rf /",
            "app && echo pwned",
            "app|cat /etc/shadow",
            "app\x00null",
            "",
            "app name with spaces",
            "app/subdir",
        ],
    )
    def test_rejects_malicious_names(self, name: str):
        assert not _SAFE_NAME_RE.match(name)

    @pytest.mark.parametrize(
        "name",
        [
            "notepad",
            "word-o365",
            "my_app_2",
            "Excel",
            "vscode",
        ],
    )
    def test_accepts_valid_names(self, name: str):
        assert _SAFE_NAME_RE.match(name)


# --- UNC path injection ---


class TestUNCPathValidation:
    def test_rejects_invalid_windows_chars(self):
        with pytest.raises(ValueError, match="invalid for Windows"):
            linux_to_unc('/home/user/file"name.txt')

    def test_rejects_pipe_char(self):
        with pytest.raises(ValueError, match="invalid for Windows"):
            linux_to_unc("/home/user/file|name.txt")

    def test_rejects_wildcard(self):
        with pytest.raises(ValueError, match="invalid for Windows"):
            linux_to_unc("/home/user/*.txt")

    def test_accepts_normal_path(self):
        result = linux_to_unc(str(Path.home() / "Documents" / "test.docx"))
        assert result.startswith("\\\\tsclient\\home\\")
        assert "test.docx" in result


# --- Extra flags whitelist ---


class TestExtraFlagsWhitelist:
    def test_allows_safe_flags(self):
        result = _filter_extra_flags("/scale:200 /sound:sys:alsa +fonts /dynamic-resolution")
        assert "/scale:200" in result
        assert "/sound:sys:alsa" in result
        assert "+fonts" in result
        assert "/dynamic-resolution" in result

    def test_allows_multitouch(self):
        # #623: +multitouch enables touchscreen / stylus / pen passthrough.
        # Input-only bare toggle, so it's on the allowlist.
        assert _filter_extra_flags("+multitouch") == ["+multitouch"]

    def test_blocks_dangerous_flags(self):
        result = _filter_extra_flags("/app:cmd.exe /cmd:bash /exec:whoami /shell:sh")
        assert len(result) == 0

    def test_blocks_cert_override(self):
        result = _filter_extra_flags("/cert:ignore /cert:tofu")
        assert len(result) == 0

    def test_blocks_auth_manipulation(self):
        result = _filter_extra_flags("/u:hacker /p:password /v:evil.com")
        assert len(result) == 0

    def test_empty_flags(self):
        assert _filter_extra_flags("") == []

    def test_freerdp3_cache_option_forms_pass(self):
        # #380: FreeRDP 3.x folds cache toggles into /cache:<sub>:<val>.
        for flag in (
            "/cache:bitmap:on",
            "/cache:bitmap:off",
            "/cache:glyph:off",
            "/cache:offscreen:on",
            "/cache:codec:rfx",
            "/cache:persist",
            "/cache:bitmap:on,glyph:off,offscreen:on",
        ):
            assert _filter_extra_flags(flag) == [flag], flag

    def test_stale_freerdp2_bare_cache_flags_blocked(self):
        # #380 (reopened): the bare FreeRDP-2 spellings are rejected by xfreerdp
        # 3.x, so they must not pass the allowlist.
        result = _filter_extra_flags(
            "+bitmap-cache -bitmap-cache +offscreen-cache -offscreen-cache "
            "+glyph-cache -glyph-cache"
        )
        assert result == []

    def test_gfx_bare_toggle_allows_disable_workaround(self):
        # #393: `-gfx` (disable the GFX channel → legacy GDI path) is a valid
        # FreeRDP workaround for the XWayland RAIL render glitch and must pass;
        # `/gfx:RFX` value form keeps working; the OPTIONAL sub-option
        # `-gfx-h264` stays blocked (it's not a bare toggle).
        assert _filter_extra_flags("-gfx") == ["-gfx"]
        assert _filter_extra_flags("+gfx") == ["+gfx"]
        assert _filter_extra_flags("/gfx:RFX") == ["/gfx:RFX"]
        assert _filter_extra_flags("-gfx-h264") == []

    def test_cache_option_rejects_file_path_and_injection(self):
        # persist-file:<path> and bad values must not slip through /cache.
        for bad in (
            "/cache:persist-file:/etc/passwd",
            "/cache:bitmap:maybe",
            "/cache:../etc",
            "/cache:bitmap:on;rm -rf",
        ):
            assert _filter_extra_flags(bad) == [], bad


# --- Device-redirection flag hardening (H1) ---


class TestDeviceRedirectionFlags:
    """Per-flag argument-shape validation for /drive, /serial, /parallel, /smartcard, /usb."""

    def test_drive_rejects_host_path_payload(self):
        result = _filter_extra_flags("/drive:etc,/etc")
        assert result == []

    def test_drive_rejects_windows_path_payload(self):
        result = _filter_extra_flags("/drive:share,C:\\Windows")
        assert result == []

    def test_drive_rejects_arbitrary_name_without_allowlist(self):
        result = _filter_extra_flags("/drive:downloads,/home/user/Downloads")
        assert result == []

    def test_drive_rejects_unknown_bare_name(self):
        result = _filter_extra_flags("/drive:root")
        assert result == []

    def test_drive_accepts_known_share_home(self):
        assert _filter_extra_flags("/drive:home") == ["/drive:home"]

    def test_drive_accepts_known_share_media(self):
        assert _filter_extra_flags("/drive:media") == ["/drive:media"]

    def test_drive_rejects_traversal_in_name(self):
        result = _filter_extra_flags("/drive:../../etc")
        assert result == []

    def test_serial_rejects_device_node(self):
        result = _filter_extra_flags("/serial:/dev/ttyUSB0")
        assert result == []

    def test_serial_rejects_named_payload(self):
        result = _filter_extra_flags("/serial:COM1,/dev/ttyS0")
        assert result == []

    def test_parallel_rejects_device_node(self):
        result = _filter_extra_flags("/parallel:lp0,/dev/lp0")
        assert result == []

    def test_smartcard_bare_accepted(self):
        assert _filter_extra_flags("/smartcard") == ["/smartcard"]

    def test_smartcard_payload_rejected(self):
        result = _filter_extra_flags("/smartcard:MyCard,/dev/card")
        assert result == []

    def test_usb_auto_accepted(self):
        assert _filter_extra_flags("/usb:auto") == ["/usb:auto"]

    def test_usb_specific_id_rejected(self):
        result = _filter_extra_flags("/usb:id,dev=1234:5678")
        assert result == []

    def test_usb_device_path_rejected(self):
        result = _filter_extra_flags("/usb:/dev/bus/usb/001/002")
        assert result == []

    def test_multi_flag_injection_drops_only_bad(self):
        result = _filter_extra_flags("/scale:150 /drive:etc,/etc /serial:/dev/tty")
        assert result == ["/scale:150"]

    def test_multi_flag_injection_logs_drops(self, caplog):
        import logging as _logging

        with caplog.at_level(_logging.WARNING, logger="winpodx.core.rdp"):
            _filter_extra_flags("/drive:etc,/etc /serial:/dev/tty")
        messages = " ".join(r.getMessage() for r in caplog.records)
        assert "/drive:etc,/etc" in messages
        assert "/serial:/dev/tty" in messages

    def test_scale_rejects_non_numeric(self):
        result = _filter_extra_flags("/scale:abc")
        assert result == []

    def test_scale_rejects_injection_suffix(self):
        result = _filter_extra_flags("/scale:100;rm")
        assert result == []

    def test_network_rejects_unknown_profile(self):
        assert _filter_extra_flags("/network:ethernet") == []

    def test_network_accepts_documented_profile(self):
        assert _filter_extra_flags("/network:lan") == ["/network:lan"]

    def test_log_level_rejects_scope_wildcard(self):
        assert _filter_extra_flags("/log-level:TRACE:com.foo") == []

    def test_log_level_accepts_plain_level(self):
        assert _filter_extra_flags("/log-level:INFO") == ["/log-level:INFO"]

    def test_valid_flag_bundle_passes_unchanged(self):
        result = _filter_extra_flags("/scale:200 /sound:sys:alsa")
        assert result == ["/scale:200", "/sound:sys:alsa"]


# --- Config validation ---


class TestConfigSecurity:
    def test_invalid_backend_reset(self):
        pod = PodConfig(backend="'; DROP TABLE users; --")
        assert pod.backend == "podman"

    def test_extreme_port_clamped(self):
        rdp = RDPConfig(port=99999)
        assert rdp.port == 65535
        rdp = RDPConfig(port=-1)
        assert rdp.port == 1

    def test_config_file_permissions(self, tmp_path, monkeypatch):
        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        cfg.rdp.password = "supersecret123"
        cfg.save()

        config_path = tmp_path / "winpodx" / "winpodx.toml"
        mode = config_path.stat().st_mode & 0o777
        assert mode == 0o600

    def test_corrupted_config_falls_back(self, tmp_path, monkeypatch):
        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        config_dir = tmp_path / "winpodx"
        config_dir.mkdir()
        (config_dir / "winpodx.toml").write_text("{{invalid toml}}")

        cfg = Config.load()
        assert cfg.rdp.ip == "127.0.0.1"


# --- App TOML injection ---


class TestAppTomlInjection:
    def test_rejects_app_with_no_name(self, tmp_path):
        toml = tmp_path / "app.toml"
        toml.write_text('executable = "notepad.exe"\n')
        assert load_app(tmp_path) is None

    def test_rejects_app_with_traversal_name(self, tmp_path):
        toml = tmp_path / "app.toml"
        toml.write_text('name = "../evil"\nexecutable = "notepad.exe"\n')
        assert load_app(tmp_path) is None

    def test_rejects_app_with_shell_name(self, tmp_path):
        toml = tmp_path / "app.toml"
        toml.write_text('name = "app;rm -rf /"\nexecutable = "notepad.exe"\n')
        assert load_app(tmp_path) is None

    def test_accepts_valid_app(self, tmp_path):
        toml = tmp_path / "app.toml"
        toml.write_text('name = "notepad"\nexecutable = "notepad.exe"\n')
        app = load_app(tmp_path)
        assert app is not None
        assert app.name == "notepad"

    def test_rejects_oversized_name(self, tmp_path):
        toml = tmp_path / "app.toml"
        long_name = "a" * 256
        toml.write_text(f'name = "{long_name}"\nexecutable = "notepad.exe"\n')
        assert load_app(tmp_path) is None

    def test_accepts_max_length_name(self, tmp_path):
        toml = tmp_path / "app.toml"
        name_255 = "a" * 255
        toml.write_text(f'name = "{name_255}"\nexecutable = "notepad.exe"\n')
        app = load_app(tmp_path)
        assert app is not None


# --- TOML writer escaping ---


class TestTomlWriterEscaping:
    def test_escapes_control_characters(self):
        from winpodx.utils.toml_writer import dumps

        data = {"section": {"key": "line1\nline2\ttab"}}
        result = dumps(data)
        assert "\\n" in result
        assert "\\t" in result
        assert "\n" not in result.split("=")[1].split('"')[1]

    def test_escapes_backslash_and_quotes(self):
        from winpodx.utils.toml_writer import dumps

        data = {"section": {"key": 'back\\slash "quoted"'}}
        result = dumps(data)
        assert '\\\\"' in result or '\\"' in result

    def test_escapes_all_control_chars(self):
        from winpodx.utils.toml_writer import dumps

        # Null byte, bell, vertical tab, DEL: all must be escaped
        data = {"section": {"key": "a\x00b\x07c\x0bd\x7fe"}}
        result = dumps(data)
        for bad in ("\x00", "\x07", "\x0b", "\x7f"):
            assert bad not in result
        assert "\\u0000" in result
        assert "\\u007F" in result


class TestYamlEscape:
    def test_yaml_escape_newlines(self, tmp_path, monkeypatch):
        from winpodx.cli.setup_cmd import _generate_compose
        from winpodx.core.config import Config

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        cfg.rdp.user = "User"
        cfg.rdp.password = "pass\nword\rtest"

        _generate_compose(cfg)

        compose = (tmp_path / "winpodx" / "compose.yaml").read_text()
        for line in compose.splitlines():
            if "PASSWORD" in line:
                assert "\nword" not in line
                assert "\\n" in line or "\\r" in line
                break

    def test_yaml_escape_win_version_quote_injection(self, tmp_path, monkeypatch):
        """Defense-in-depth: even when ``win_version`` somehow bypasses
        the ``__post_init__`` allowlist (e.g. set after load(), or via a
        legacy code path that mutates ``cfg.pod.win_version`` directly),
        the compose writer's ``_yaml_escape`` must neutralise YAML-
        breaking characters so a hand-crafted value cannot inject env
        keys into the dockur service. Bypass validation here with
        ``object.__setattr__`` to simulate the worst case.
        """
        from winpodx.cli.setup_cmd import _generate_compose
        from winpodx.core.config import Config

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        cfg.rdp.user = "User"
        # Bypass __post_init__'s dangerous-char reject — simulate a
        # legacy / future code path that mutates win_version directly
        # without re-running validation.
        object.__setattr__(cfg.pod, "win_version", '11"\nEVIL_KEY: "x')

        _generate_compose(cfg)

        compose = (tmp_path / "winpodx" / "compose.yaml").read_text()
        # The injected key must NOT appear as a YAML key. Note: the
        # escaped value still contains the literal substring
        # ``EVIL_KEY`` inside the quoted VERSION scalar (the escape
        # neutralises the syntactic break, not the bytes). What we
        # care about is that ``EVIL_KEY`` is not parsed as a key.
        for line in compose.splitlines():
            stripped = line.strip()
            # Real YAML key would be at zero or canonical indentation
            # like "      EVIL_KEY:" — never inside a quoted value.
            assert not stripped.startswith("EVIL_KEY:"), (
                f"YAML injection succeeded: line {line!r} parses as a key"
            )

    def test_yaml_escape_container_name_image_disk_size(self, tmp_path, monkeypatch):
        """Same defense-in-depth contract for ``cfg.pod.container_name``,
        ``cfg.pod.image``, and ``cfg.pod.disk_size``. All three land in
        YAML double-quoted scalars (container_name in name field, image
        in service.image, disk_size in DISK_SIZE env). A bypass-validation
        injection must not break out of its scalar.
        """
        from winpodx.cli.setup_cmd import _generate_compose
        from winpodx.core.config import Config

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        cfg.rdp.user = "User"
        object.__setattr__(cfg.pod, "container_name", 'win"\nCONTAINER_EVIL: "x')
        object.__setattr__(cfg.pod, "image", 'image"\nIMAGE_EVIL: "x')
        object.__setattr__(cfg.pod, "disk_size", '64G"\nDISK_EVIL: "x')

        _generate_compose(cfg)

        compose = (tmp_path / "winpodx" / "compose.yaml").read_text()
        for evil_key in ("CONTAINER_EVIL:", "IMAGE_EVIL:", "DISK_EVIL:"):
            for line in compose.splitlines():
                assert not line.strip().startswith(evil_key), (
                    f"YAML injection succeeded: {evil_key} appears as a key in {line!r}"
                )


class TestAgentTokenStaging:
    """Phase 1 agent: setup must stage the OEM token with mode 0600."""

    def test_setup_stages_oem_token_0600(self, tmp_path, monkeypatch):
        import argparse

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))

        from winpodx.cli.setup_cmd import handle_setup
        from winpodx.core.config import Config
        from winpodx.utils.deps import DepCheck

        cfg = Config()
        cfg.save()

        oem_dir = tmp_path / "oem"
        oem_dir.mkdir()
        monkeypatch.setattr("winpodx.cli.setup_cmd._find_oem_dir", lambda: str(oem_dir))
        monkeypatch.setattr(
            "winpodx.cli.setup_cmd.check_all",
            lambda **_: {"freerdp": DepCheck(name="freerdp", found=True, note="stub")},
        )

        args = argparse.Namespace(backend=None, non_interactive=True)
        handle_setup(args)

        staged = oem_dir / "agent_token.txt"
        assert staged.exists()
        assert (staged.stat().st_mode & 0o777) == 0o600


class TestAgentPortMapping:
    """Phase 1 agent: both forwarding-chain legs must be present in compose."""

    def test_compose_includes_user_ports_env(self, tmp_path, monkeypatch):
        from winpodx.cli.setup_cmd import _generate_compose
        from winpodx.core.config import Config

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        _generate_compose(cfg)

        compose = (tmp_path / "winpodx" / "compose.yaml").read_text()
        # Agent port + guest SMB port (reverse-open guest-disk, #616).
        assert 'USER_PORTS: "8765,445"' in compose

    def test_compose_includes_host_port_mapping(self, tmp_path, monkeypatch):
        from winpodx.cli.setup_cmd import _generate_compose
        from winpodx.core.config import Config

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        _generate_compose(cfg)

        compose = (tmp_path / "winpodx" / "compose.yaml").read_text()
        assert "127.0.0.1:8765:8765/tcp" in compose

    def test_compose_port_mapping_loopback_only(self, tmp_path, monkeypatch):
        from winpodx.cli.setup_cmd import _generate_compose
        from winpodx.core.config import Config

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        _generate_compose(cfg)

        compose = (tmp_path / "winpodx" / "compose.yaml").read_text()
        assert "0.0.0.0:8765" not in compose
        assert ":8765:8765" not in compose.replace("127.0.0.1:8765:8765", "")


# --- Password filter ---


class TestPasswordFilter:
    def test_filter_installed_on_handlers(self):
        from winpodx.utils.logging import PasswordFilter, setup_logging

        root = logging.getLogger("winpodx")
        root.handlers.clear()
        setup_logging(log_file=False)
        assert any(isinstance(flt, PasswordFilter) for h in root.handlers for flt in h.filters)
        root.handlers.clear()


# --- Config set integer validation ---


class TestPowerShellEscape:
    """v0.1.9.5: _change_windows_password migrated from podman-exec to
    windows_exec.run_in_windows. Tests now mock the new entry point and
    inspect the payload string instead of subprocess argv."""

    def test_username_single_quote_escaped(self, monkeypatch):
        from winpodx.core.config import Config
        from winpodx.core.windows_exec import WindowsExecResult

        cfg = Config()
        cfg.rdp.user = "admin'; whoami; '"
        cfg.rdp.password = "current-pw"
        cfg.pod.backend = "podman"

        captured: dict[str, str] = {}

        def fake(cfg_inner, payload, **_kw):
            captured["payload"] = payload
            return WindowsExecResult(rc=0, stdout="password set", stderr="")

        monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", fake)
        from winpodx.core.provisioner import _change_windows_password

        assert _change_windows_password(cfg, "newpass123") is True
        assert "admin''; whoami; ''" in captured["payload"]

    def test_normal_username_unchanged(self, monkeypatch):
        from winpodx.core.config import Config
        from winpodx.core.windows_exec import WindowsExecResult

        cfg = Config()
        cfg.rdp.user = "User"
        cfg.rdp.password = "current-pw"
        cfg.pod.backend = "podman"

        captured: dict[str, str] = {}

        def fake(cfg_inner, payload, **_kw):
            captured["payload"] = payload
            return WindowsExecResult(rc=0, stdout="password set", stderr="")

        monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", fake)
        from winpodx.core.provisioner import _change_windows_password

        assert _change_windows_password(cfg, "testpass") is True
        assert "net user 'User' 'testpass'" in captured["payload"]


class TestPasswordTimestamp:
    def test_naive_timestamp_no_crash(self, tmp_path, monkeypatch):
        from winpodx.core.config import Config

        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        cfg.rdp.password = "testpass"
        cfg.rdp.password_max_age = 7
        cfg.rdp.password_updated = "2020-01-01T00:00:00"  # naive, no timezone
        cfg.pod.backend = "podman"

        from winpodx.core.provisioner import _auto_rotate_password

        result = _auto_rotate_password(cfg)
        assert result is not None


class TestConfigSetValidation:
    def test_int_coercion_error(self, tmp_path, monkeypatch, capsys):
        monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
        cfg = Config()
        cfg.save()

        from winpodx.cli.config_cmd import _set

        with pytest.raises(SystemExit):
            _set("rdp.port", "not-a-number")