vps-manager/tests/test_agent_collector.py

"""VPS Agent 安全检查采集测试

覆盖两处「假 pass」回归：

1. ssh_config：只读 /etc/ssh/sshd_config 会漏掉 Include 展开的 sshd_config.d/*.conf
   （Debian/Ubuntu 云镜像把加固项放在那里），且关键字缺失时被当成 unknown 直接放过；
   实际 PasswordAuthentication 的 sshd 默认值就是 yes，这类机器是开着密码登录的。
2. firewall：`iptables -L -n` 即使零规则也会输出链头，拿 stdout 非空判「有规则」
   会永远 pass；改用 `iptables -S` 统计 -A 规则行并结合 INPUT 策略判断。

运行：.venv/bin/pytest tests/test_agent_collector.py -v
"""

import subprocess
import sys
from pathlib import Path

import pytest

AGENT_DIR = Path(__file__).resolve().parent.parent / "agent"
if str(AGENT_DIR) not in sys.path:
    sys.path.insert(0, str(AGENT_DIR))

import collector  # noqa: E402  （agent 目录不是包，按路径导入）


@pytest.fixture()
def sshd_tree(tmp_path, monkeypatch):
    """把 collector.SSHD_CONFIG 指向 tmp 下的假配置树"""
    main = tmp_path / "sshd_config"
    dropin_dir = tmp_path / "sshd_config.d"
    dropin_dir.mkdir()
    monkeypatch.setattr(collector, "SSHD_CONFIG", str(main))
    return main, dropin_dir


def test_ssh_config_reads_included_dropins(sshd_tree):
    """加固项写在 sshd_config.d/*.conf 时必须被读到，不能误报 warn"""
    main, dropin_dir = sshd_tree
    main.write_text(
        f"Include {dropin_dir}/*.conf\n\n# 主文件里是未加固的旧值\nPermitRootLogin yes\n"
    )
    (dropin_dir / "50-cloud.conf").write_text(
        "PermitRootLogin prohibit-password\nPasswordAuthentication no\n"
    )

    result = collector._check_ssh_config()

    assert result["status"] == "pass"
    assert "PermitRootLogin=prohibit-password" in result["detail"]
    assert "PasswordAuthentication=no" in result["detail"]


def test_ssh_config_first_value_wins(sshd_tree):
    """sshd 语义是「首次出现的值生效」：Include 在顶部时 drop-in 优先于主文件"""
    main, dropin_dir = sshd_tree
    main.write_text(f"Include {dropin_dir}/*.conf\nPasswordAuthentication yes\n")
    (dropin_dir / "60-hardening.conf").write_text("PasswordAuthentication no\n")

    result = collector._check_ssh_config()

    assert result["status"] == "pass"
    assert "PasswordAuthentication=no" in result["detail"]


def test_ssh_config_missing_keys_use_sshd_defaults(sshd_tree):
    """关键字没写 ≠ 安全：PasswordAuthentication 默认 yes，必须告警而不是 pass"""
    main, _ = sshd_tree
    main.write_text("# 空配置，两个关键字都没写\nPort 22\n")

    result = collector._check_ssh_config()

    assert result["status"] == "warn"
    assert "PasswordAuthentication=yes" in result["detail"]
    assert "未显式配置" in result["detail"]
    assert "关闭密码登录" in result["suggestion"]


def test_ssh_config_unreadable_is_unknown(monkeypatch):
    """读不到主配置时如实报 unknown，不能默认放行"""
    monkeypatch.setattr(collector, "SSHD_CONFIG", "/nonexistent/sshd_config")

    result = collector._check_ssh_config()

    assert result["status"] == "unknown"
    assert "无法读取" in result["detail"]


def _fake_run(monkeypatch, iptables_output: str):
    """ufw 不存在（走 iptables 分支），iptables -S 返回指定输出"""

    def run(cmd, **kwargs):
        if cmd[0] == "ufw":
            raise FileNotFoundError("ufw not installed")
        return subprocess.CompletedProcess(cmd, 0, stdout=iptables_output, stderr="")

    monkeypatch.setattr(collector.subprocess, "run", run)


def test_firewall_no_rules_is_warn(monkeypatch):
    """零规则时 -S 仍有 -P 行，不能据此判 pass"""
    _fake_run(monkeypatch, "-P INPUT ACCEPT\n-P FORWARD ACCEPT\n-P OUTPUT ACCEPT\n")

    result = collector._check_firewall()

    assert result["status"] == "warn"
    assert "无过滤规则" in result["detail"]
    assert result["suggestion"]


def test_firewall_with_rules_is_pass(monkeypatch):
    _fake_run(
        monkeypatch,
        "-P INPUT ACCEPT\n-A INPUT -i lo -j ACCEPT\n-A INPUT -p tcp -m tcp --dport 22 -j ACCEPT\n",
    )

    result = collector._check_firewall()

    assert result["status"] == "pass"
    assert "2 条规则" in result["detail"]


def test_firewall_drop_policy_without_rules_is_pass(monkeypatch):
    """没有 -A 规则但 INPUT 策略是 DROP，本身就是有效防护"""
    _fake_run(monkeypatch, "-P INPUT DROP\n-P FORWARD DROP\n-P OUTPUT ACCEPT\n")

    result = collector._check_firewall()

    assert result["status"] == "pass"
    assert "INPUT 策略 DROP" in result["detail"]


def test_firewall_unknown_when_no_tool(monkeypatch):
    def run(cmd, **kwargs):
        raise FileNotFoundError(cmd[0])

    monkeypatch.setattr(collector.subprocess, "run", run)

    assert collector._check_firewall()["status"] == "unknown"