mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(lib): fix acme.renew deploy call and hoist wireguard imports

Browse Source 
- Add deploy(domain) to acme.renew() to register deploy hook after
  renewal, matching the pattern in issue()
- Hoist run and run_proc imports to module level in wireguard,
  removing 4 inline imports for consistency
This commit is contained in:
Mike Teehan
2026-05-23 03:56:13 +00:00
parent 37039351be
commit cf8115bb0d
3 changed files with 126 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/acme.py
+44 -47
View File
@@ -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),
}
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),
}
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",
]
lib/wireguard.py
+54 -73
View File
@@ -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,28 +30,7 @@ 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 {
DEFAULT_CONFIG: dict[str, Any] = {
"interface": {
"name": "wg0",
"listen_port": 51820,
@@ -67,34 +47,14 @@ def _default_config() -> dict:
# --- 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",
]
tests/test_wireguard.py
+14 -14
View File
@@ -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,