ServerKit/backend/tests/test_connect_policy.py

"""The panel half of fleet policies.

What matters here is the capability gate: a probe that cannot answer must
leave its bundle out of `capabilities`, because that is what makes ServerKit
Cloud show "not checkable here" instead of marking a healthy server
non-compliant. Every probe is replaced by the test, so nothing touches the
host.
"""
import pytest

from app.services import connect_policy


class _Firewall:
    active = True
    ports = [2222]

    @classmethod
    def get_status(cls):
        return {"any_active": cls.active}

    @classmethod
    def ssh_ports(cls):
        return list(cls.ports)

    @classmethod
    def enable(cls, firewall=None, force=False):
        cls.active = True
        return {"success": True}


class _Fail2ban:
    installed = True
    running = True

    @classmethod
    def get_fail2ban_status(cls):
        return {"installed": cls.installed, "service_running": cls.running}


class _SecurityPolicy:
    required = False

    @classmethod
    def require_2fa_enabled(cls):
        return cls.required

    @classmethod
    def set_require_2fa(cls, enabled, user_id=None):
        cls.required = bool(enabled)
        return cls.required


@pytest.fixture(autouse=True)
def probes(monkeypatch):
    """Point every probe at an in-memory host."""
    import app.services.firewall_service as fw
    import app.services.fail2ban_jail_service as f2b
    import app.services.security_policy_service as sp
    import app.services.panel_update_service as pu

    _Firewall.active = True
    _Firewall.ports = [2222]
    _Fail2ban.installed = True
    _Fail2ban.running = True
    _SecurityPolicy.required = False

    monkeypatch.setattr(fw, "FirewallService", _Firewall)
    monkeypatch.setattr(f2b, "Fail2banJailService", _Fail2ban)
    monkeypatch.setattr(sp, "SecurityPolicyService", _SecurityPolicy)
    monkeypatch.setattr(pu, "get_status", lambda: {"version": "1.9.22"})
    monkeypatch.setattr(connect_policy, "_backup_policy_rows", lambda app=None: [])
    return True


# ---------- facts ----------


def test_a_healthy_host_reports_every_bundle_it_can():
    facts = connect_policy.build_facts()
    assert facts["facts_version"] == 1
    assert facts["security"]["firewall_enabled"] is True
    assert facts["security"]["fail2ban_running"] is True
    assert facts["security"]["ssh_port"] == 2222
    assert facts["security"]["panel_2fa_required"] is False
    assert facts["updates"]["panel_version"] == "1.9.22"
    assert "security" in facts["capabilities"]
    assert "updates" in facts["capabilities"]


def test_packages_are_never_advertised_because_nothing_probes_them():
    """The rule Cloud has for pending security updates must read as
    "not checkable here", not as "no updates pending"."""
    facts = connect_policy.build_facts()
    assert "packages" not in facts["capabilities"]
    assert "os_security_pending" not in facts["updates"]


def test_a_firewall_that_is_off_is_reported_as_off():
    _Firewall.active = False
    assert connect_policy.build_facts()["security"]["firewall_enabled"] is False


def test_fail2ban_installed_but_stopped_is_not_running():
    _Fail2ban.running = False
    assert connect_policy.build_facts()["security"]["fail2ban_running"] is False


def test_a_probe_that_raises_leaves_its_fact_out_rather_than_guessing(monkeypatch):
    import app.services.firewall_service as fw

    class Broken:
        @staticmethod
        def get_status():
            raise RuntimeError("no ufw here")

        @staticmethod
        def ssh_ports():
            raise RuntimeError("no sshd_config")
    monkeypatch.setattr(fw, "FirewallService", Broken)
    facts = connect_policy.build_facts()
    assert "firewall_enabled" not in facts["security"]
    assert "ssh_port" not in facts["security"]
    # The bundle is still advertised: 2FA and fail2ban answered.
    assert "security" in facts["capabilities"]


def test_a_panel_with_no_backup_engine_is_not_judged_on_backups(monkeypatch):
    monkeypatch.setattr(connect_policy, "_backup_policy_rows", lambda app=None: None)

    import app.services.storage_provider_service as sps

    class NoStorage:
        @staticmethod
        def get_config():
            raise RuntimeError("no storage.json")
    monkeypatch.setattr(sps, "StorageProviderService", NoStorage)
    facts = connect_policy.build_facts()
    assert "backups" not in facts["capabilities"]
    assert "backups" not in facts


def test_the_newest_successful_backup_is_the_one_reported(monkeypatch):
    from datetime import datetime, timedelta, timezone
    now = datetime.now(timezone.utc)
    monkeypatch.setattr(connect_policy, "_backup_policy_rows", lambda app=None: [
        {"last_status": "success", "last_run_at": now - timedelta(hours=9),
         "last_drill_at": None},
        {"last_status": "success", "last_run_at": now - timedelta(hours=2),
         "last_drill_at": now - timedelta(days=3)},
        {"last_status": "failed", "last_run_at": now, "last_drill_at": None},
    ])
    facts = connect_policy.build_facts()
    assert facts["backups"]["last_success_at"].startswith(
        (now - timedelta(hours=2)).isoformat()[:13])
    assert facts["backups"]["last_verified_at"] is not None
    assert "backups" in facts["capabilities"]


def test_a_backup_that_never_succeeded_reports_the_key_as_null(monkeypatch):
    """Null, not absent: Cloud tells "never succeeded" apart from "did not
    report" by whether the key is there."""
    monkeypatch.setattr(connect_policy, "_backup_policy_rows", lambda app=None: [
        {"last_status": "failed", "last_run_at": None, "last_drill_at": None},
    ])
    facts = connect_policy.build_facts()
    assert facts["backups"]["last_success_at"] is None
    assert "last_success_at" in facts["backups"]


def test_the_facts_frame_is_the_policy_stream():
    frame = connect_policy.facts_frame("pol1", {"facts_version": 1})
    assert frame == {"s": "pol1", "t": "open", "k": "policy",
                     "p": {"facts": {"facts_version": 1}}}


# ---------- repairs ----------


def test_turning_the_firewall_on_uses_the_panels_own_path():
    _Firewall.active = False
    out = connect_policy.enable_firewall({})
    assert out["ok"] is True
    assert _Firewall.active is True


def test_a_firewall_that_refuses_to_come_up_says_why(monkeypatch):
    import app.services.firewall_service as fw

    class Refuses(_Firewall):
        @classmethod
        def enable(cls, firewall=None, force=False):
            return {"success": False, "error": "enabling this would lock SSH out"}
    monkeypatch.setattr(fw, "FirewallService", Refuses)
    out = connect_policy.enable_firewall({})
    assert out["ok"] is False
    assert "lock SSH out" in out["summary"]


def test_requiring_two_factor_turns_the_policy_on():
    out = connect_policy.require_2fa({})
    assert out["ok"] is True
    assert _SecurityPolicy.required is True
    assert "grace window" in out["summary"]


def test_fail2ban_that_is_not_installed_is_refused_with_what_to_do():
    _Fail2ban.installed = False
    out = connect_policy.enable_fail2ban({})
    assert out["ok"] is False
    assert "not installed" in out["summary"]


def test_security_updates_are_refused_honestly_rather_than_faked():
    out = connect_policy.security_upgrade({})
    assert out["ok"] is False
    assert "does not install operating-system security updates" in out["summary"]


def test_check_now_without_a_relay_connection_says_so(monkeypatch):
    monkeypatch.setattr(connect_policy, "_relay_client", lambda: None)
    out = connect_policy.report({})
    assert out["ok"] is False
    assert "not connected to ServerKit Cloud" in out["summary"]


def test_check_now_publishes_the_document(monkeypatch):
    sent = []

    class Client:
        def publish_policy(self, facts):
            sent.append(facts)
            return True
    monkeypatch.setattr(connect_policy, "_relay_client", lambda: Client())
    out = connect_policy.report({})
    assert out["ok"] is True
    assert sent and sent[0]["facts_version"] == 1


def test_every_policy_action_is_registered_with_its_consent_scope():
    from app.services import connect_commands
    for action in ("policy.report", "security.firewall.enable", "security.fail2ban.enable",
                   "security.2fa.require", "packages.security_upgrade", "backup.verify"):
        assert action in connect_commands.HANDLERS
    assert connect_commands.scope_for("security.firewall.enable") == "security.remediate"
    assert connect_commands.scope_for("packages.security_upgrade") == "packages.upgrade"
    assert connect_commands.scope_for("backup.verify") == "backup.run"
    # policy.report has no entry, so its scope is its own name — the Observe
    # bundle every paired panel already grants.
    assert connect_commands.scope_for("policy.report") == "policy.report"