diff --git a/lib/acme.py b/lib/acme.py index c114d8a..cfa4a33 100644 --- a/lib/acme.py +++ b/lib/acme.py @@ -27,6 +27,8 @@ _ACME_ENVIRON = { ), } +_WEBROOT = PROJECT_DIR / "data" / "acme" / "www" + def _find_acme() -> str: """Locate the acme.sh binary on the system. @@ -147,77 +149,52 @@ def get_email() -> str: return "" -def issue(domain: str, webroot: str | None = None) -> dict: +def issue(domain: str, webroot: str | None = None, email: str | None = None) -> str: """Issue a new SSL certificate for a domain. Args: domain: The primary domain name. webroot: Path to the web root directory for HTTP-01 validation. + email: Contact email. Falls back to configured ACME email if not given. Returns: - A dict with 'success', 'domain', 'message', 'output', and 'error'. + Combined stdout from the acme.sh command. + + Raises: + RuntimeError: If issuance fails. """ args: list[str] = ["--issue", "-d", domain] - - if webroot: - args.extend(["--webroot", webroot]) - - email = get_email() - if email: - args.extend(["-m", email]) + args.extend(["--webroot", webroot or str(_WEBROOT)]) + contact = email or get_email() + if contact: + args.extend(["-m", contact]) args.append("--force") - - try: - output = _run_acme(args) - deploy(domain) - return { - "success": True, - "domain": domain, - "message": f"Certificate for {domain} issued successfully", - "output": output.strip(), - "error": None, - } - except RuntimeError as exc: - return { - "success": False, - "domain": domain, - "message": f"Failed to issue certificate for {domain}", - "output": "", - "error": str(exc), - } + output = _run_acme(args) + deploy(domain) + logger.info("Certificate for %s issued successfully", domain) + return output.strip() -def renew(domain: str, force: bool = False) -> dict: +def renew(domain: str, force: bool = False) -> str: """Renew an existing SSL certificate. Args: domain: The domain whose certificate should be renewed. - force: If True, renew even if the certificate isn't close to expiry. + force: If True, renew even if not close to expiry. Returns: - A dict with 'success', 'domain', 'message', 'output', and 'error'. + Combined stdout from the acme.sh command. + + Raises: + RuntimeError: If renewal fails. """ args: list[str] = ["--renew", "-d", domain] if force: args.append("--force") - - try: - output = _run_acme(args) - return { - "success": True, - "domain": domain, - "message": f"Certificate for {domain} renewed successfully", - "output": output.strip(), - "error": None, - } - except RuntimeError as exc: - return { - "success": False, - "domain": domain, - "message": f"Failed to renew certificate for {domain}", - "output": "", - "error": str(exc), - } + output = _run_acme(args) + deploy(domain) + logger.info("Certificate for %s renewed successfully", domain) + return output.strip() def remove(domain: str) -> str: @@ -274,7 +251,10 @@ def list_certs() -> list[dict]: certs.append( { "domain": main, - "ca": entry.get("CA", ""), + "issuer": entry.get("CA", ""), + "expiry": entry.get("certificate_expires", ""), + "days_remaining": days, + "expired": days is not None and days <= 0, "cert_path": cert_path, "key_path": key_path, "ca_path": ca_path, @@ -494,3 +474,20 @@ def _has_auto_renew(domain: str) -> bool: acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME)) domain_conf = Path(acme_home_env) / f"{domain}.conf" return bool(domain_conf.is_file()) + + +__all__ = [ + "copy_cert", + "days_until_expiry", + "deploy", + "get_cert_info", + "get_cert_paths", + "get_email", + "get_expiry", + "is_expired", + "issue", + "list_certs", + "remove", + "renew", + "set_email", +] diff --git a/lib/wireguard.py b/lib/wireguard.py index 89568e9..361cc22 100644 --- a/lib/wireguard.py +++ b/lib/wireguard.py @@ -1,23 +1,24 @@ -""" -WireGuard Manager for Vacuum Wall SSL Proxy Firewall. +"""WireGuard Manager for Vacuum Wall SSL Proxy Firewall. Generates wg-quick configurations, manages peers, and controls the WireGuard tunnel interface. """ -import json import logging import os -import subprocess +from copy import deepcopy from datetime import UTC, datetime from pathlib import Path +from typing import Any from jinja2 import Environment, FileSystemLoader +from lib.common import deep_merge, load_json, run, run_proc, save_json + logger = logging.getLogger(__name__) PROJECT_DIR = Path(__file__).resolve().parent.parent -CONFIG_PATH = str(PROJECT_DIR / "config" / "wireguard" / "config.json") +CONFIG_PATH = PROJECT_DIR / "config" / "wireguard" / "config.json" WG_CONF_PATH = "/etc/wireguard/wg0.conf" WG_QUICK_BIN = "wg-quick" WG_BIN = "wg" @@ -29,72 +30,31 @@ ENV = Environment( trim_blocks=True, ) - -# --- Helpers --- - - -def _run(cmd: list[str], check: bool = True) -> subprocess.CompletedProcess: - """Run a command via sudo and return the completed process.""" - return subprocess.run( - ["sudo", *cmd], - capture_output=True, - text=True, - check=check, - ) - - -def _ensure_dir(path: str) -> None: - """Create parent directories for *path* if they don't exist.""" - Path(path).parent.mkdir(parents=True, exist_ok=True) - - -def _default_config() -> dict: - """Return the skeleton config with no keys and no peers.""" - return { - "interface": { - "name": "wg0", - "listen_port": 51820, - "private_key": "", - "public_key": "", - "addresses": ["10.137.0.1/24"], - "post_up": None, - "post_down": None, - }, - "peers": {}, - } +DEFAULT_CONFIG: dict[str, Any] = { + "interface": { + "name": "wg0", + "listen_port": 51820, + "private_key": "", + "public_key": "", + "addresses": ["10.137.0.1/24"], + "post_up": None, + "post_down": None, + }, + "peers": {}, +} # --- Core config persistence --- -def get_config() -> dict: +def get_config() -> dict[str, Any]: """Load the current WireGuard configuration from the JSON store.""" - try: - with open(CONFIG_PATH) as f: - cfg = json.load(f) - defaults = _default_config() - cfg.setdefault("interface", defaults["interface"]) - cfg["interface"].setdefault("name", defaults["interface"]["name"]) - cfg["interface"].setdefault("listen_port", defaults["interface"]["listen_port"]) - cfg["interface"].setdefault("private_key", defaults["interface"]["private_key"]) - cfg["interface"].setdefault("public_key", defaults["interface"]["public_key"]) - cfg["interface"].setdefault("addresses", defaults["interface"]["addresses"]) - cfg["interface"].setdefault("post_up", defaults["interface"]["post_up"]) - cfg["interface"].setdefault("post_down", defaults["interface"]["post_down"]) - cfg.setdefault("peers", {}) - return cfg - except (FileNotFoundError, json.JSONDecodeError): - return _default_config() + return deep_merge(deepcopy(DEFAULT_CONFIG), load_json(CONFIG_PATH)) -def save_config(cfg: dict) -> None: +def save_config(cfg: dict[str, Any]) -> None: """Persist *cfg* to the JSON store atomically.""" - _ensure_dir(CONFIG_PATH) - tmp = CONFIG_PATH + ".tmp" - with open(tmp, "w") as f: - json.dump(cfg, f, indent=4) - f.write("\n") - os.replace(tmp, CONFIG_PATH) + save_json(CONFIG_PATH, cfg) # --- Key generation --- @@ -102,9 +62,9 @@ def save_config(cfg: dict) -> None: def generate_keypair() -> tuple[str, str]: """Generate a WireGuard private/public key pair using ``wg`` CLI.""" - res = _run([WG_BIN, "genkey"]) + res = run_proc([WG_BIN, "genkey"], sudo=True) private_key = res.stdout.strip() - res2 = _run([WG_BIN, "pubkey"], input=private_key) + res2 = run_proc([WG_BIN, "pubkey"], sudo=True, input=private_key) public_key = res2.stdout.strip() return private_key, public_key @@ -112,7 +72,7 @@ def generate_keypair() -> tuple[str, str]: # --- wg0.conf generation --- -def generate_conf(cfg: dict) -> str: +def generate_conf(cfg: dict[str, Any]) -> str: """Render a valid wg-quick config file from *cfg* using Jinja2.""" tmpl = ENV.get_template("wireguard.conf") return tmpl.render( @@ -137,11 +97,11 @@ def apply() -> None: with open(local_tmp, "w") as f: f.write(conf_text) os.chmod(local_tmp, 0o600) - _run(["cp", "--", str(local_tmp), WG_CONF_PATH]) - _run(["chown", "root:root", WG_CONF_PATH], check=False) + run(["cp", "--", str(local_tmp), WG_CONF_PATH], sudo=True) + run(["chown", "root:root", WG_CONF_PATH], sudo=True, check=False) local_tmp.unlink(missing_ok=True) - _run([WG_QUICK_BIN, "up", cfg["interface"]["name"]]) + run([WG_QUICK_BIN, "up", cfg["interface"]["name"]], sudo=True) logger.info("WireGuard tunnel '%s' brought up", cfg["interface"]["name"]) @@ -149,34 +109,34 @@ def down() -> None: """Bring the WireGuard tunnel interface down.""" cfg = get_config() name = cfg["interface"]["name"] - _run([WG_QUICK_BIN, "down", name]) + run([WG_QUICK_BIN, "down", name], sudo=True) logger.info("WireGuard tunnel '%s' brought down", name) # --- Status --- -def status() -> dict: +def status() -> dict[str, Any]: """Query the live tunnel state via ``wg show``.""" cfg = get_config() name = cfg["interface"]["name"] - result = { + result: dict[str, Any] = { "up": False, "interface": {}, "peers": [], } try: - proc = _run([WG_BIN, "show", name], check=False) - if proc.returncode != 0: + res = run_proc([WG_BIN, "show", name], sudo=True, check=False) + if res.returncode != 0: return result - raw = proc.stdout.strip() + raw = res.stdout.strip() except Exception: return result - current_peer = None - peers: list[dict] = [] + current_peer: dict[str, Any] | None = None + peers: list[dict[str, Any]] = [] for line in raw.splitlines(): line = line.strip() @@ -261,7 +221,7 @@ def add_peer( allowed_ips: list[str] | None = None, persistent_keepalive: int | None = None, preshared_key: str | None = None, -) -> dict: +) -> dict[str, Any]: """Add (or update) a peer in the configuration.""" cfg = get_config() peers = cfg.setdefault("peers", {}) @@ -301,10 +261,10 @@ def remove_peer(name: str) -> None: logger.info("WireGuard peer '%s' removed", name) -def get_peers() -> list[dict]: +def get_peers() -> list[dict[str, Any]]: """List all configured peers (from the JSON store, *not* live).""" cfg = get_config() - peers = [] + peers: list[dict[str, Any]] = [] for name, info in cfg.get("peers", {}).items(): entry = dict(info) entry["name"] = name @@ -313,7 +273,7 @@ def get_peers() -> list[dict]: return peers -def get_peer_status() -> list[dict]: +def get_peer_status() -> list[dict[str, Any]]: """Return live peer status from ``wg show``.""" st = status() return st.get("peers", []) @@ -393,7 +353,7 @@ def set_post_down(cmd: str | None) -> None: # --- Initialise --- -def initialize() -> dict: +def initialize() -> dict[str, Any]: """Perform first-time WireGuard setup.""" cfg = get_config() @@ -411,10 +371,10 @@ def initialize() -> dict: # --- Utility: parse wg show into structured peer map --- -def _parse_wg_show(output: str) -> dict: +def _parse_wg_show(output: str) -> dict[str, Any]: """Internal parser for ``wg show`` multiline output.""" - peers: dict = {} - current = None + peers: dict[str, dict[str, Any]] = {} + current: dict[str, Any] | None = None for line in output.splitlines(): line = line.strip() @@ -440,3 +400,24 @@ def _parse_wg_show(output: str) -> dict: current["persistent_keepalive"] = line.split(":", 1)[1].strip() return peers + + +__all__ = [ + "DEFAULT_CONFIG", + "add_peer", + "apply", + "down", + "generate_client_conf", + "generate_conf", + "generate_keypair", + "get_config", + "get_peer_status", + "get_peers", + "initialize", + "remove_peer", + "save_config", + "set_listen_port", + "set_post_down", + "set_post_up", + "status", +] diff --git a/tests/test_wireguard.py b/tests/test_wireguard.py index 71e9b6c..980e78c 100644 --- a/tests/test_wireguard.py +++ b/tests/test_wireguard.py @@ -1,5 +1,5 @@ import json -from pathlib import Path +from copy import deepcopy from unittest.mock import MagicMock, patch import pytest @@ -10,14 +10,14 @@ from lib import wireguard @pytest.fixture def temp_config(tmp_path): original = wireguard.CONFIG_PATH - wireguard.CONFIG_PATH = str(tmp_path / "config.json") + wireguard.CONFIG_PATH = tmp_path / "config.json" yield tmp_path wireguard.CONFIG_PATH = original class TestDefaultConfig: def test_returns_skeleton(self): - cfg = wireguard._default_config() + cfg = wireguard.DEFAULT_CONFIG assert cfg["interface"]["name"] == "wg0" assert cfg["interface"]["listen_port"] == 51820 assert cfg["interface"]["private_key"] == "" @@ -31,8 +31,7 @@ class TestGetConfig: assert cfg["peers"] == {} def test_loads_existing_config(self, temp_config): - path = Path(wireguard.CONFIG_PATH) - expected = { + wireguard.CONFIG_PATH.write_text(json.dumps({ "interface": { "name": "wg0", "listen_port": 51820, @@ -43,15 +42,14 @@ class TestGetConfig: "post_down": None, }, "peers": {}, - } - path.write_text(json.dumps(expected)) + })) cfg = wireguard.get_config() assert cfg["interface"]["private_key"] == "existing-key" class TestSaveConfig: def test_save_and_reload(self, temp_config): - cfg = wireguard._default_config() + cfg = deepcopy(wireguard.DEFAULT_CONFIG) cfg["interface"]["listen_port"] = 51821 wireguard.save_config(cfg) loaded = wireguard.get_config() @@ -59,7 +57,7 @@ class TestSaveConfig: class TestGenerateKeyPair: - @patch("lib.wireguard._run") + @patch("lib.wireguard.run_proc") def test_returns_keypair(self, mock_run): mock_run.side_effect = [ MagicMock(returncode=0, stdout="private-key\n"), @@ -68,6 +66,8 @@ class TestGenerateKeyPair: private, public = wireguard.generate_keypair() assert private == "private-key" assert public == "public-key" + assert mock_run.call_count == 2 + assert mock_run.call_args_list[1].kwargs.get("input") == "private-key" class TestGetPeers: @@ -97,7 +97,7 @@ class TestGetPeers: } }, } - Path(wireguard.CONFIG_PATH).write_text(json.dumps(cfg)) + wireguard.CONFIG_PATH.write_text(json.dumps(cfg)) peers = wireguard.get_peers() assert len(peers) == 1 assert peers[0]["name"] == "client1" @@ -180,14 +180,14 @@ class TestInitialize: }, "peers": {}, } - Path(wireguard.CONFIG_PATH).write_text(json.dumps(existing)) + wireguard.CONFIG_PATH.write_text(json.dumps(existing)) cfg = wireguard.initialize() assert cfg["interface"]["private_key"] == "original-private" mock_gen.assert_not_called() class TestStatus: - @patch("lib.wireguard._run") + @patch("lib.wireguard.run_proc") def test_returns_down_when_interface_down(self, mock_run, temp_config): mock_run.return_value = MagicMock( returncode=1, stdout="", stderr="interface not found" @@ -195,7 +195,7 @@ class TestStatus: result = wireguard.status() assert result["up"] is False - @patch("lib.wireguard._run") + @patch("lib.wireguard.run_proc") def test_parses_interface_info(self, mock_run, temp_config): mock_run.return_value = MagicMock( returncode=0, @@ -206,7 +206,7 @@ class TestStatus: assert result["interface"]["public_key"] == "ABCDEF" assert result["interface"]["listen_port"] == 51820 - @patch("lib.wireguard._run") + @patch("lib.wireguard.run_proc") def test_parses_peer_info(self, mock_run, temp_config): mock_run.return_value = MagicMock( returncode=0,