furtka/tests/test_wireguard.py

"""Tests for furtka.wireguard — tunnel-config write/remove + systemctl toggle.

Same shape as test_https.py: stub the systemctl call so we assert the conf
file is written/removed and that a reload failure rolls the conf file back
to its pre-call state.
"""

import subprocess
from pathlib import Path

import pytest

from furtka import wireguard

WG = {
    "private_key": "cHJpdmF0ZS1rZXk=",
    "address": "10.8.0.2/24",
    "server_public_key": "c2VydmVyLXB1YmxpYy1rZXk=",
    "endpoint": "gateway.example.com:51820",
    "allowed_ips": "10.8.0.0/24",
}


def test_conf_content_shape():
    content = wireguard._conf_content(WG)
    assert "[Interface]" in content
    assert "PrivateKey = cHJpdmF0ZS1rZXk=" in content
    assert "Address = 10.8.0.2/24" in content
    assert "[Peer]" in content
    assert "PublicKey = c2VydmVyLXB1YmxpYy1rZXk=" in content
    assert "Endpoint = gateway.example.com:51820" in content
    assert "AllowedIPs = 10.8.0.0/24" in content
    assert "PersistentKeepalive = 25" in content


def test_conf_content_omits_preshared_key_when_absent():
    # WG has no "preshared_key" key at all — the v14-gateway shape.
    content = wireguard._conf_content(WG)
    assert "PresharedKey" not in content


def test_conf_content_includes_preshared_key_when_present():
    # v15-gateway shape (see furtka-gateway's wgeasy_v15.py) — the server
    # enforces this even if the client's own conf omits it, so dropping it
    # here would break the handshake silently.
    wg = {**WG, "preshared_key": "cHJlc2hhcmVkLWtleQ=="}
    content = wireguard._conf_content(wg)
    assert "PresharedKey = cHJlc2hhcmVkLWtleQ==" in content
    # Must sit between PublicKey and Endpoint, inside [Peer].
    peer_section = content.split("[Peer]\n", 1)[1]
    assert (
        peer_section.index("PublicKey")
        < peer_section.index("PresharedKey")
        < peer_section.index("Endpoint")
    )


def test_enable_requires_wg_config(tmp_path):
    with pytest.raises(ValueError, match="wg config is required"):
        wireguard.set_tunnel_enabled(True, conf_path=tmp_path / "furtka-gw.conf")


def test_enable_surfaces_clean_error_on_unreadable_conf_dir(tmp_path, monkeypatch):
    # Simulated rather than via a 0o000 directory: CI runs as root, which
    # ignores directory permissions.
    def denied(self, *a, **kw):
        raise PermissionError(13, "Permission denied", str(self))

    monkeypatch.setattr(Path, "mkdir", denied)

    def must_not_reload(action):
        raise AssertionError("systemctl must not run when the conf can't be written")

    conf_path = tmp_path / "restricted" / "furtka-gw.conf"
    with pytest.raises(wireguard.WireguardError, match="could not access"):
        wireguard.set_tunnel_enabled(True, wg=WG, conf_path=conf_path, reload=must_not_reload)


def test_default_conf_path_uses_env_override(monkeypatch, tmp_path):
    monkeypatch.setenv("FURTKA_WIREGUARD_CONF", str(tmp_path / "custom.conf"))
    assert wireguard.default_conf_path() == tmp_path / "custom.conf"


def test_default_conf_path_default_when_unset(monkeypatch):
    monkeypatch.delenv("FURTKA_WIREGUARD_CONF", raising=False)
    assert wireguard.default_conf_path() == wireguard.DEFAULT_CONF_PATH


def test_enable_writes_conf_and_calls_reload(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"
    calls = []

    def fake_reload(action):
        calls.append(action)

    result = wireguard.set_tunnel_enabled(True, wg=WG, conf_path=conf_path, reload=fake_reload)

    assert result is True
    assert calls == ["enable"]
    content = conf_path.read_text()
    assert "PrivateKey = cHJpdmF0ZS1rZXk=" in content
    assert oct(conf_path.stat().st_mode)[-3:] == "600"


def test_disable_removes_conf_and_calls_reload(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"
    conf_path.write_text(wireguard._conf_content(WG))
    calls = []

    result = wireguard.set_tunnel_enabled(
        False, conf_path=conf_path, reload=lambda action: calls.append(action)
    )

    assert result is False
    assert calls == ["disable"]
    assert not conf_path.exists()


def test_disable_is_idempotent_when_already_off(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"
    result = wireguard.set_tunnel_enabled(False, conf_path=conf_path, reload=lambda action: None)
    assert result is False
    assert not conf_path.exists()


def test_reload_failure_rolls_back_enable(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"

    def failing_reload(action):
        raise subprocess.CalledProcessError(1, ["systemctl"], stderr="bad config")

    with pytest.raises(wireguard.WireguardError, match="systemctl failed: bad config"):
        wireguard.set_tunnel_enabled(True, wg=WG, conf_path=conf_path, reload=failing_reload)
    # No conf existed before this call, so none should exist after rollback.
    assert not conf_path.exists()


def test_reload_failure_rolls_back_disable(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"
    original = wireguard._conf_content(WG)
    conf_path.write_text(original)

    def failing_reload(action):
        raise subprocess.CalledProcessError(1, ["systemctl"], stderr="bad config")

    with pytest.raises(wireguard.WireguardError):
        wireguard.set_tunnel_enabled(False, conf_path=conf_path, reload=failing_reload)
    assert conf_path.read_text() == original


def test_systemctl_missing_raises_and_rolls_back(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"

    def missing_systemctl(action):
        raise FileNotFoundError(2, "No such file", "systemctl")

    with pytest.raises(wireguard.WireguardError, match="systemctl not available"):
        wireguard.set_tunnel_enabled(True, wg=WG, conf_path=conf_path, reload=missing_systemctl)
    assert not conf_path.exists()


def test_status_not_configured(tmp_path):
    s = wireguard.status(conf_path=tmp_path / "does-not-exist.conf")
    assert s == {"configured": False, "latest_handshake": None}


def test_status_survives_unreadable_conf_path(tmp_path, monkeypatch):
    """Regression: an inaccessible parent directory raises PermissionError
    even for a path that doesn't exist yet — confirmed by actually hitting
    this against a real box's /etc/wireguard/ as a non-root user, which
    crashed the whole /api/gateway/status request with a raw traceback.
    A read-only status check must degrade to "not configured", never 500.
    """

    # Simulated rather than via a 0o000 directory: CI runs as root, which
    # ignores directory permissions.
    def denied(self):
        raise PermissionError(13, "Permission denied", str(self))

    monkeypatch.setattr(Path, "is_file", denied)

    s = wireguard.status(conf_path=tmp_path / "restricted" / "furtka-gw.conf")
    assert s == {"configured": False, "latest_handshake": None}


def test_status_configured_reports_handshake(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"
    conf_path.write_text(wireguard._conf_content(WG))

    s = wireguard.status(conf_path=conf_path, wg_show=lambda: "1234567890")

    assert s == {"configured": True, "latest_handshake": "1234567890"}


def test_status_configured_but_wg_tools_missing(tmp_path):
    conf_path = tmp_path / "furtka-gw.conf"
    conf_path.write_text(wireguard._conf_content(WG))

    def missing_wg():
        raise FileNotFoundError(2, "No such file", "wg")

    s = wireguard.status(conf_path=conf_path, wg_show=missing_wg)

    assert s == {"configured": True, "latest_handshake": None}