furtka/tests/test_gateway.py

"""Tests for furtka.gateway — the box-side register/rotate/deregister client.

Same shape as test_updater.py's network tests: monkeypatch
urllib.request.urlopen directly rather than standing up a real server.
"""

from __future__ import annotations

import json
import urllib.error

import pytest

from furtka import gateway


class _FakeResp:
    def __init__(self, body: dict):
        self._body = json.dumps(body).encode()

    def __enter__(self):
        return self

    def __exit__(self, *a):
        return False

    def read(self):
        return self._body


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",
}


@pytest.fixture
def gw(tmp_path, monkeypatch):
    monkeypatch.setenv("FURTKA_GATEWAY_CONFIG", str(tmp_path / "gateway.json"))
    monkeypatch.setenv("FURTKA_GATEWAY_CONNECT_STATE", str(tmp_path / "gateway-connect-state.json"))
    monkeypatch.setenv("FURTKA_GATEWAY_CONNECT_LOCK", str(tmp_path / "gateway-connect.lock"))
    import importlib

    from furtka import paths as p

    importlib.reload(p)
    importlib.reload(gateway)
    return gateway


def test_load_config_returns_none_when_never_registered(gw):
    assert gw.load_config() is None


def test_register_saves_config(gw, monkeypatch):
    calls = []

    def fake_urlopen(req, timeout):
        calls.append((req.full_url, json.loads(req.data)))
        return _FakeResp({"box_id": "abc123", "box_token": "abc123.secret", "wg": WG})

    monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)

    config = gw.register("https://gw.example.com/", "reg-token", "my-box")

    assert config["gateway_url"] == "https://gw.example.com"  # trailing slash stripped
    assert config["box_id"] == "abc123"
    assert config["box_token"] == "abc123.secret"
    assert config["wg"] == WG
    assert gw.load_config() == config

    url, body = calls[0]
    assert url == "https://gw.example.com/v1/boxes/register"
    assert body == {"registration_token": "reg-token", "box_name": "my-box"}


def test_register_config_file_is_0600(gw, monkeypatch, tmp_path):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )
    gw.register("https://gw.example.com", "tok", "box")
    from furtka.paths import gateway_config_file

    assert oct(gateway_config_file().stat().st_mode)[-3:] == "600"


def test_register_raises_gateway_error_on_http_error(gw, monkeypatch):
    def raising(req, timeout):
        raise urllib.error.HTTPError(req.full_url, 401, "invalid registration token", {}, None)

    monkeypatch.setattr("urllib.request.urlopen", raising)

    with pytest.raises(gateway.GatewayError, match="HTTP 401"):
        gw.register("https://gw.example.com", "bad-token", "my-box")
    assert gw.load_config() is None


def test_rotate_token_updates_stored_config(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.old", "wg": WG}),
    )
    gw.register("https://gw.example.com", "tok", "box")

    def fake_rotate(req, timeout):
        assert req.headers.get("Authorization") == "Bearer abc.old"
        assert req.full_url == "https://gw.example.com/v1/boxes/abc/rotate-token"
        return _FakeResp({"box_token": "abc.new"})

    monkeypatch.setattr("urllib.request.urlopen", fake_rotate)

    new_token = gw.rotate_token()

    assert new_token == "abc.new"
    assert gw.load_config()["box_token"] == "abc.new"


def test_rotate_token_without_prior_registration_raises(gw):
    with pytest.raises(gateway.GatewayError, match="not connected"):
        gw.rotate_token()


def test_deregister_calls_gateway_and_clears_config(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )
    gw.register("https://gw.example.com", "tok", "box")

    calls = []

    def fake_deregister(req, timeout):
        calls.append(req.full_url)
        return _FakeResp({"status": "ok"})

    monkeypatch.setattr("urllib.request.urlopen", fake_deregister)

    gw.deregister()

    assert calls == ["https://gw.example.com/v1/boxes/abc/deregister"]
    assert gw.load_config() is None


def test_deregister_when_never_registered_is_a_noop(gw, monkeypatch):
    calls = []
    monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout: calls.append(1))
    gw.deregister()
    assert calls == []


def test_publish_route_requires_connection(gw):
    with pytest.raises(gateway.GatewayError, match="not connected"):
        gw.publish_route("vaultwarden", "vault.example.com", 8081)


def test_publish_route_posts_to_routes_endpoint(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )
    gw.register("https://gw.example.com", "tok", "box")

    def fake_publish(req, timeout):
        assert req.full_url == "https://gw.example.com/v1/routes"
        assert req.headers.get("Authorization") == "Bearer abc.x"
        assert json.loads(req.data) == {
            "app_name": "vaultwarden",
            "subdomain": "vault.example.com",
            "port": 8081,
        }
        return _FakeResp({"route_id": "r1", "public_url": "https://vault.example.com/"})

    monkeypatch.setattr("urllib.request.urlopen", fake_publish)

    result = gw.publish_route("vaultwarden", "vault.example.com", 8081)

    assert result == {"route_id": "r1", "public_url": "https://vault.example.com/"}


def test_unpublish_route_requires_connection(gw):
    with pytest.raises(gateway.GatewayError, match="not connected"):
        gw.unpublish_route("r1")


def test_unpublish_route_calls_delete_endpoint(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )
    gw.register("https://gw.example.com", "tok", "box")

    calls = []

    def fake_delete(req, timeout):
        calls.append((req.full_url, req.get_method(), req.headers.get("Authorization")))
        return _FakeResp({"status": "ok"})

    monkeypatch.setattr("urllib.request.urlopen", fake_delete)

    gw.unpublish_route("r1")

    assert calls == [("https://gw.example.com/v1/routes/r1", "DELETE", "Bearer abc.x")]


def test_write_and_read_state_round_trip(gw):
    gw.write_state("registering")
    s = gw.read_state()
    assert s["stage"] == "registering"
    assert "updated_at" in s


def test_read_state_missing_file_returns_empty_dict(gw):
    assert gw.read_state() == {}


def test_acquire_lock_contention(gw):
    first = gw.acquire_lock()
    try:
        with pytest.raises(gateway.GatewayError, match="already in progress"):
            gw.acquire_lock()
    finally:
        first.close()


def test_run_connect_happy_path(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )
    tunnel_calls = []
    monkeypatch.setattr(
        gw.wireguard,
        "set_tunnel_enabled",
        lambda enabled, wg=None: tunnel_calls.append((enabled, wg)),
    )

    gw.run_connect("https://gw.example.com", "tok", "my-box")

    assert tunnel_calls == [(True, WG)]
    assert gw.read_state()["stage"] == "done"
    assert gw.load_config()["box_id"] == "abc"


def test_run_connect_registration_failure_writes_error_state(gw, monkeypatch):
    def raising(req, timeout):
        raise urllib.error.HTTPError(req.full_url, 401, "invalid registration token", {}, None)

    monkeypatch.setattr("urllib.request.urlopen", raising)

    gw.run_connect("https://gw.example.com", "bad-tok", "my-box")

    state = gw.read_state()
    assert state["stage"] == "error"
    assert "401" in state["error"]
    assert gw.load_config() is None


def test_run_connect_tunnel_failure_writes_error_state(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )

    def failing_tunnel(enabled, wg=None):
        raise gw.wireguard.WireguardError("systemctl failed: bad config")

    monkeypatch.setattr(gw.wireguard, "set_tunnel_enabled", failing_tunnel)

    gw.run_connect("https://gw.example.com", "tok", "my-box")

    state = gw.read_state()
    assert state["stage"] == "error"
    assert "bad config" in state["error"]
    # Registration itself succeeded — config is still saved even though the
    # tunnel failed to come up, so a retry doesn't need to re-register.
    assert gw.load_config()["box_id"] == "abc"


def test_run_connect_retry_after_tunnel_failure_reuses_registration(gw, monkeypatch):
    register_calls = []

    def urlopen(req, timeout):
        register_calls.append(req.full_url)
        return _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG})

    monkeypatch.setattr("urllib.request.urlopen", urlopen)

    def failing_tunnel(enabled, wg=None):
        raise gw.wireguard.WireguardError("systemctl failed: unit missing")

    monkeypatch.setattr(gw.wireguard, "set_tunnel_enabled", failing_tunnel)
    gw.run_connect("https://gw.example.com", "tok", "my-box")
    assert gw.read_state()["stage"] == "error"

    tunnel_calls = []
    monkeypatch.setattr(
        gw.wireguard,
        "set_tunnel_enabled",
        lambda enabled, wg=None: tunnel_calls.append((enabled, wg)),
    )
    gw.run_connect("https://gw.example.com/", "tok", "my-box")

    # Registered exactly once; the retry only brought the tunnel up.
    assert register_calls == ["https://gw.example.com/v1/boxes/register"]
    assert tunnel_calls == [(True, WG)]
    assert gw.read_state()["stage"] == "done"
    assert gw.load_config()["box_id"] == "abc"


def test_run_connect_refuses_a_different_gateway_while_registered(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )
    gw.register("https://gw.example.com", "tok", "box")

    def must_not_call(req, timeout):
        raise AssertionError("must not register with a second gateway")

    monkeypatch.setattr("urllib.request.urlopen", must_not_call)
    monkeypatch.setattr(gw.wireguard, "set_tunnel_enabled", lambda enabled, wg=None: None)

    gw.run_connect("https://other.example.com", "tok", "box")

    state = gw.read_state()
    assert state["stage"] == "error"
    assert "disconnect first" in state["error"]
    assert gw.load_config()["gateway_url"] == "https://gw.example.com"


def test_deregister_failure_keeps_local_config_for_retry(gw, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda req, timeout: _FakeResp({"box_id": "abc", "box_token": "abc.x", "wg": WG}),
    )
    gw.register("https://gw.example.com", "tok", "box")

    def raising(req, timeout):
        raise urllib.error.URLError("connection refused")

    monkeypatch.setattr("urllib.request.urlopen", raising)

    with pytest.raises(gateway.GatewayError):
        gw.deregister()
    # Local config must survive so the caller can retry later.
    assert gw.load_config() is not None