mteehan
75b86fd60d
acme:
- acme.sh chmods its tree to owner-only (700/600) every run, which
broke the two-user model: a tree left owner-only by one user made
every acme.sh call of the other exit 2
- normalize_acme_home() reopens group access (sudo chmod g+rwX,
files only — setgid dirs trip RestrictSUIDSGID); _run_acme_preflight
is the choke point before every daemon acme.sh call + startup
- acme service now runs as the daemon user; --log persists the raw CA
transcript; SYS_LOG=6 journals manual issue/renew runs
- timer daily-only: two runs/day landed inside ZeroSSL's 24h
validation backoff (Retry-After: 86400) — a permanent renewal lockout
- _collect_acme no longer raises on cert-list failure; reports
status.error (AcmeState.status) so the certs page can surface it
firewall: re-stamp the applied baseline on live zone mutations
(interfaces/services/rich-rules/masquerade/forward-ports) so cancel-all
reverts to post-mutation state, not a stale install-era snapshot;
set_masquerade syncs the declarative config for existing zones;
add_forward_port records toaddr only with toport
status: apply-all accepts {"force": true} (forwarded to the firewall
apply only); ApplyConfirm force checkbox; applyResultToasts() — the
errors map wins over the 200; ActionButton checks errors before the
success toast; dashboard uses ApplyConfirm
system_import: drift re-imports carry the existing apply-meta; first
import stamps the adopted content as applied (it is the running state)
— no phantom pending changes
nginx: get_config only re-saves when migration actually changed the
config (no more owner/mtime churn on every read)
install: repair mis-owned top-level system dirs (tmpfiles
unsafe-path-transition), warn with a full-repair command for deeper
mis-ownership
daemon/server: loop.get_exception_handler() (aiohttp API fix)
tests: 888 pytest + 24 node passing; ruff clean
1378 lines
49 KiB
Python
1378 lines
49 KiB
Python
"""Tests for daemon/handlers/acme.py — handler endpoint logic."""
|
|
|
|
import asyncio
|
|
import inspect
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import daemon.handlers.acme as acme_mod
|
|
from daemon.handlers.acme import (
|
|
IssueRequest,
|
|
_check_account_registered,
|
|
_check_acme_account,
|
|
_check_acme_home_writable,
|
|
_check_dns_public,
|
|
_check_dns_resolves,
|
|
_check_firewall_port_80,
|
|
_check_nginx_config,
|
|
_check_nginx_running,
|
|
_check_openssl_available,
|
|
_check_port_80_listening,
|
|
_get_account_info,
|
|
_get_external_ip,
|
|
_is_private_ip,
|
|
_validate,
|
|
deactivate_account,
|
|
generate_self_signed,
|
|
get_account,
|
|
get_renew_status,
|
|
issue_cert,
|
|
register_account,
|
|
renew_cert,
|
|
)
|
|
from daemon.server import ConflictError, NotFoundError
|
|
|
|
|
|
def _await_renew(body):
|
|
"""Start a renewal via renew_cert and await its background task."""
|
|
|
|
async def _run():
|
|
result = renew_cert(None, body)
|
|
task = acme_mod._ISSUANCE_TASKS.get(result.get("request_id"))
|
|
if task is not None:
|
|
await task
|
|
return result
|
|
|
|
return asyncio.run(_run())
|
|
|
|
|
|
class TestGenerateSelfSigned:
|
|
def test_generate_creates_files(self, tmp_path):
|
|
with (
|
|
patch("daemon.handlers.acme.PROJECT_DIR", tmp_path),
|
|
):
|
|
result = generate_self_signed(None, {"domain": "test.local"})
|
|
|
|
assert result["domain"] == "test.local"
|
|
assert result["generated"] is True
|
|
certs_dir = tmp_path / "data" / "certs"
|
|
assert result["cert"] == str(certs_dir / "test.local.crt")
|
|
assert result["key"] == str(certs_dir / "test.local.key")
|
|
assert (certs_dir / "test.local.crt").is_file()
|
|
assert (certs_dir / "test.local.key").is_file()
|
|
|
|
def test_generate_idempotent_skips_existing(self, tmp_path):
|
|
certs_dir = tmp_path / "data" / "certs"
|
|
certs_dir.mkdir(parents=True)
|
|
(certs_dir / "test.local.crt").write_text("dummy-cert")
|
|
(certs_dir / "test.local.key").write_text("dummy-key")
|
|
|
|
with patch("daemon.handlers.acme.PROJECT_DIR", tmp_path):
|
|
result = generate_self_signed(None, {"domain": "test.local"})
|
|
|
|
assert result["generated"] is False
|
|
|
|
def test_generate_partial_existing(self, tmp_path):
|
|
certs_dir = tmp_path / "data" / "certs"
|
|
certs_dir.mkdir(parents=True)
|
|
(certs_dir / "test.local.crt").write_text("dummy-cert")
|
|
# key missing -> should regenerate
|
|
|
|
with patch("daemon.handlers.acme.PROJECT_DIR", tmp_path):
|
|
result = generate_self_signed(None, {"domain": "test.local"})
|
|
|
|
assert result["generated"] is True
|
|
|
|
def test_generate_custom_days(self, tmp_path):
|
|
with (
|
|
patch("daemon.handlers.acme.PROJECT_DIR", tmp_path),
|
|
patch("subprocess.run") as mock_run,
|
|
):
|
|
|
|
def _create_files(*args, **kwargs):
|
|
certs_dir = tmp_path / "data" / "certs"
|
|
certs_dir.mkdir(parents=True, exist_ok=True)
|
|
(certs_dir / "test.local.crt").touch()
|
|
(certs_dir / "test.local.key").touch()
|
|
return Path("")
|
|
|
|
mock_run.side_effect = _create_files
|
|
generate_self_signed(None, {"domain": "test.local", "days": 730})
|
|
args = mock_run.call_args[0][0]
|
|
assert "-days" in args
|
|
idx = args.index("-days")
|
|
assert args[idx + 1] == "730"
|
|
|
|
certs_dir = tmp_path / "data" / "certs"
|
|
if (certs_dir / "test.local.crt").is_file():
|
|
assert certs_dir.is_dir()
|
|
|
|
def test_generate_creates_directory(self, tmp_path):
|
|
with patch("daemon.handlers.acme.PROJECT_DIR", tmp_path):
|
|
generate_self_signed(None, {"domain": "test.local"})
|
|
|
|
assert (tmp_path / "data" / "certs").is_dir()
|
|
|
|
def test_generate_requires_domain(self):
|
|
with pytest.raises(ValueError, match="domain"):
|
|
generate_self_signed(None, {"foo": "bar"})
|
|
|
|
def test_generate_requires_body(self):
|
|
with pytest.raises(ValueError, match="body"):
|
|
generate_self_signed(None, None)
|
|
|
|
|
|
class TestCheckNginxRunning:
|
|
def test_via_systemctl_active(self):
|
|
from unittest.mock import patch
|
|
|
|
mock_result = MagicMock(returncode=0, stdout="active\n")
|
|
with patch("subprocess.run", return_value=mock_result):
|
|
passed, msg = _check_nginx_running()
|
|
assert passed is True
|
|
assert "running" in msg
|
|
|
|
def test_via_systemctl_inactive(self, tmp_path):
|
|
from unittest.mock import patch
|
|
|
|
mock_result = MagicMock(returncode=3, stdout="inactive\n")
|
|
with (
|
|
patch("subprocess.run", return_value=mock_result),
|
|
patch.object(Path, "is_file", return_value=False),
|
|
):
|
|
passed, _ = _check_nginx_running()
|
|
assert passed is False
|
|
|
|
def test_via_pid_file(self):
|
|
from unittest.mock import patch
|
|
|
|
def run_side_effect(cmd, **kwargs):
|
|
raise FileNotFoundError()
|
|
|
|
with (
|
|
patch("subprocess.run", side_effect=run_side_effect),
|
|
patch.object(Path, "read_text", return_value="1234\n"),
|
|
):
|
|
|
|
def fake_is_file(self):
|
|
if self == Path("/var/run/nginx.pid"):
|
|
return True
|
|
if str(self) == "/proc/1234/status":
|
|
return True
|
|
return Path(self).is_file()
|
|
|
|
with patch.object(Path, "is_file", fake_is_file):
|
|
passed, _ = _check_nginx_running()
|
|
assert passed is True
|
|
|
|
|
|
class TestCheckNginxConfig:
|
|
def test_valid_config(self):
|
|
with patch("lib.nginx.test_config", return_value=(True, "test passed")):
|
|
passed, _ = _check_nginx_config()
|
|
assert passed is True
|
|
|
|
def test_invalid_config(self):
|
|
with patch("lib.nginx.test_config", return_value=(False, "test failed: blah")):
|
|
passed, msg = _check_nginx_config()
|
|
assert passed is False
|
|
assert "blah" in msg
|
|
|
|
|
|
class TestCheckFirewallPort80:
|
|
def test_port_open_via_services(self):
|
|
from unittest.mock import patch
|
|
|
|
def proc_side_effect(cmd, **kwargs):
|
|
if "--get-active-zones" in cmd:
|
|
return MagicMock(returncode=0, stdout="public\n eth0\n")
|
|
if "--list-services" in cmd:
|
|
return MagicMock(returncode=0, stdout="http https dns ssh\n")
|
|
return MagicMock(returncode=1, stdout="")
|
|
|
|
with (
|
|
patch("lib.common.run_proc", side_effect=proc_side_effect),
|
|
):
|
|
passed, _ = _check_firewall_port_80()
|
|
assert passed is True
|
|
|
|
def test_blocked_by_firewall(self):
|
|
from unittest.mock import patch
|
|
|
|
def proc_side_effect(cmd, **kwargs):
|
|
if "--get-active-zones" in cmd:
|
|
return MagicMock(returncode=0, stdout="public\n eth0\n")
|
|
if "--list-services" in cmd:
|
|
return MagicMock(returncode=0, stdout="https dns ssh\n")
|
|
if "--list-ports" in cmd:
|
|
return MagicMock(returncode=0, stdout="443/tcp\n")
|
|
return MagicMock(returncode=1, stdout="")
|
|
|
|
with (
|
|
patch("lib.common.run_proc", side_effect=proc_side_effect),
|
|
):
|
|
passed, msg = _check_firewall_port_80()
|
|
assert passed is False
|
|
assert "80" in msg
|
|
|
|
def test_firewalld_not_detected(self):
|
|
from unittest.mock import patch
|
|
|
|
with patch(
|
|
"lib.common.run_proc", return_value=MagicMock(returncode=1, stdout="")
|
|
):
|
|
passed, msg = _check_firewall_port_80()
|
|
assert passed is True
|
|
assert "skipping" in msg
|
|
|
|
|
|
class TestCheckAcmeHomeWritable:
|
|
def test_writable(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
passed, _ = _check_acme_home_writable()
|
|
assert passed is True
|
|
|
|
|
|
class TestCheckAcmeHomeWritable_Missing:
|
|
def test_missing_dir(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
passed, msg = _check_acme_home_writable()
|
|
assert passed is False
|
|
assert "does not exist" in msg
|
|
|
|
|
|
class TestCheckAcmeHomeWritable_Permissions:
|
|
def test_not_writable(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
acme_dir.chmod(0o444)
|
|
try:
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
passed, _ = _check_acme_home_writable()
|
|
assert passed is False
|
|
finally:
|
|
acme_dir.chmod(0o755)
|
|
|
|
|
|
class TestCheckOpensslAvailable:
|
|
def test_available(self):
|
|
from unittest.mock import patch
|
|
|
|
mock_result = MagicMock(returncode=0, stdout="OpenSSL 3.0.0\n")
|
|
with (
|
|
patch("shutil.which", return_value="/usr/bin/openssl"),
|
|
patch("subprocess.run", return_value=mock_result),
|
|
):
|
|
passed, msg = _check_openssl_available()
|
|
assert passed is True
|
|
assert "OpenSSL" in msg
|
|
|
|
def test_not_found(self):
|
|
with patch("shutil.which", return_value=None):
|
|
passed, _ = _check_openssl_available()
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckPort80Listening:
|
|
def test_listening(self):
|
|
from unittest.mock import patch
|
|
|
|
mock_sock = MagicMock()
|
|
mock_sock.connect_ex.return_value = 0
|
|
mock_sock.__enter__ = MagicMock(return_value=mock_sock)
|
|
mock_sock.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
passed, _ = _check_port_80_listening()
|
|
assert passed is True
|
|
|
|
def test_not_listening(self):
|
|
from unittest.mock import patch
|
|
|
|
mock_sock = MagicMock()
|
|
mock_sock.connect_ex.return_value = 111
|
|
mock_sock.__enter__ = MagicMock(return_value=mock_sock)
|
|
mock_sock.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
passed, _ = _check_port_80_listening()
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckAcmeAccount:
|
|
def test_via_acme_info(self, tmp_path):
|
|
from unittest.mock import patch
|
|
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
(acme_dir / "acme.sh").write_text("#!/bin/sh\nexit 0")
|
|
(acme_dir / "acme.sh").chmod(0o755)
|
|
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", acme_dir),
|
|
patch(
|
|
"daemon.handlers.acme._find_acme_bin",
|
|
return_value=str(acme_dir / "acme.sh"),
|
|
),
|
|
patch("subprocess.run", return_value=MagicMock(returncode=0, stdout="ok")),
|
|
):
|
|
passed, _ = _check_acme_account()
|
|
assert passed is True
|
|
|
|
def test_via_account_conf(self, tmp_path):
|
|
from unittest.mock import patch
|
|
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n"
|
|
)
|
|
|
|
def run_side_effect(cmd, **kwargs):
|
|
raise FileNotFoundError()
|
|
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", acme_dir),
|
|
patch("subprocess.run", side_effect=run_side_effect),
|
|
):
|
|
passed, _ = _check_acme_account()
|
|
assert passed is True
|
|
|
|
def test_not_configured(self, tmp_path):
|
|
from unittest.mock import patch
|
|
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
|
|
def run_side_effect(cmd, **kwargs):
|
|
raise FileNotFoundError()
|
|
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", acme_dir),
|
|
patch("subprocess.run", side_effect=run_side_effect),
|
|
):
|
|
passed, _ = _check_acme_account()
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckDnsPublic:
|
|
def test_resolves_correctly(self):
|
|
from unittest.mock import patch
|
|
|
|
mock_result = MagicMock(
|
|
returncode=0, stdout="example.com has address 52.14.150.110"
|
|
)
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._get_local_ips", return_value={"52.14.150.110"}
|
|
),
|
|
patch("subprocess.run", return_value=mock_result),
|
|
):
|
|
passed, _ = _check_dns_public("example.com")
|
|
assert passed is True
|
|
|
|
def test_does_not_resolve(self):
|
|
from unittest.mock import patch
|
|
|
|
mock_result = MagicMock(returncode=1, stdout="NXDOMAIN")
|
|
with (
|
|
patch("daemon.handlers.acme._get_local_ips", return_value={"8.8.8.8"}),
|
|
patch("subprocess.run", return_value=mock_result),
|
|
):
|
|
passed, _ = _check_dns_public("example.com")
|
|
assert passed is False
|
|
|
|
def test_nat_detected_skips_check(self):
|
|
from unittest.mock import patch
|
|
|
|
with patch(
|
|
"daemon.handlers.acme._get_local_ips",
|
|
return_value={"192.168.1.1"},
|
|
):
|
|
passed, msg = _check_dns_public("example.com")
|
|
assert passed is True
|
|
assert "NAT" in msg
|
|
|
|
|
|
class TestCheckDomainFormat:
|
|
def test_valid(self):
|
|
from daemon.handlers.acme import _check_domain_format
|
|
|
|
passed, _ = _check_domain_format("example.com")
|
|
assert passed is True
|
|
|
|
def test_invalid(self):
|
|
from daemon.handlers.acme import _check_domain_format
|
|
|
|
passed, _ = _check_domain_format("-bad.com")
|
|
assert passed is False
|
|
|
|
|
|
class TestValidate:
|
|
def test_returns_all_checks(self):
|
|
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_installed", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_openssl_available",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_home_writable",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_account_registered",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_email_configured",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_account", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_running", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_config", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_challenge_config",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_port_80_listening",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_firewall_port_80",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_domain_format", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_existing_cert", return_value=(True, "ok")
|
|
),
|
|
):
|
|
result = _validate("example.com")
|
|
|
|
assert result["domain"] == "example.com"
|
|
assert result["ready"] is True
|
|
assert len(result["checks"]) == 16
|
|
for c in result["checks"]:
|
|
assert c["passed"] is True
|
|
|
|
def test_failing_check_blocks_ready(self):
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_installed", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_openssl_available",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_home_writable",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_account_registered",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_email_configured",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_account", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_running",
|
|
return_value=(False, "not running"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_config", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_challenge_config",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_port_80_listening",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_firewall_port_80",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_existing_cert", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_domain_format", return_value=(True, "ok")
|
|
),
|
|
):
|
|
result = _validate("example.com")
|
|
|
|
assert result["ready"] is False
|
|
nginx_check = next(c for c in result["checks"] if c["name"] == "nginx_running")
|
|
assert nginx_check["passed"] is False
|
|
assert nginx_check["blocking"] is True
|
|
|
|
def test_non_blocking_failure_allows_ready(self):
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_installed", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_openssl_available",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_home_writable",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_account_registered",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_email_configured",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_account", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_running", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_config", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_challenge_config",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_port_80_listening",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_firewall_port_80",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_dns_public",
|
|
return_value=(False, "skipped"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_existing_cert", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_domain_format", return_value=(True, "ok")
|
|
),
|
|
):
|
|
result = _validate("example.com")
|
|
|
|
assert result["ready"] is True
|
|
dns_pub = next(c for c in result["checks"] if c["name"] == "dns_public")
|
|
assert dns_pub["passed"] is False
|
|
assert dns_pub["blocking"] is False
|
|
|
|
|
|
class TestExternalIp:
|
|
def test_success(self):
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = b"93.184.216.34"
|
|
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
|
mock_resp.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
|
result = _get_external_ip()
|
|
assert result == "93.184.216.34"
|
|
|
|
def test_primary_fails_fallback_succeeds(self):
|
|
def urlopen_side_effect(url, timeout=None):
|
|
if "ipify" in url.full_url:
|
|
raise urllib.error.URLError("primary down")
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = b"93.184.216.50"
|
|
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
|
mock_resp.__exit__ = MagicMock(return_value=False)
|
|
return mock_resp
|
|
|
|
with patch("urllib.request.urlopen", side_effect=urlopen_side_effect):
|
|
result = _get_external_ip()
|
|
assert result == "93.184.216.50"
|
|
|
|
def test_both_fail_returns_none(self):
|
|
with patch("urllib.request.urlopen", side_effect=urllib.error.URLError("fail")):
|
|
result = _get_external_ip()
|
|
assert result is None
|
|
|
|
def test_env_override_url(self):
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = b"8.8.8.8"
|
|
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
|
mock_resp.__exit__ = MagicMock(return_value=False)
|
|
|
|
with (
|
|
patch.dict(
|
|
"os.environ", {"VACUUM_WALL_EXTERNAL_IP_URL": "https://my.ip.api"}
|
|
),
|
|
patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen,
|
|
):
|
|
result = _get_external_ip()
|
|
assert result == "8.8.8.8"
|
|
call_url = mock_urlopen.call_args[0][0]
|
|
assert "my.ip.api" in call_url.full_url
|
|
|
|
|
|
class TestIsPrivateIp:
|
|
def test_public_ip(self):
|
|
assert _is_private_ip("8.8.8.8") is False
|
|
|
|
def test_private_10(self):
|
|
assert _is_private_ip("10.0.0.1") is True
|
|
|
|
def test_private_192(self):
|
|
assert _is_private_ip("192.168.1.1") is True
|
|
|
|
def test_private_172(self):
|
|
assert _is_private_ip("172.16.0.1") is True
|
|
|
|
def test_invalid_ip(self):
|
|
assert _is_private_ip("not-an-ip") is False
|
|
|
|
def test_doc_range_203(self):
|
|
assert _is_private_ip("203.0.113.5") is True
|
|
|
|
|
|
class TestCheckDnsResolves:
|
|
def test_match_local_ip(self):
|
|
mock_results = [("AF_INET", "STREAM", 6, "", ("192.168.1.10", 80))]
|
|
with (
|
|
patch("socket.getaddrinfo", return_value=mock_results),
|
|
patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}),
|
|
):
|
|
passed, msg = _check_dns_resolves("example.com")
|
|
assert passed is True
|
|
assert "DNS resolves correctly" in msg
|
|
|
|
def test_match_external_ip(self):
|
|
mock_results = [("AF_INET", "STREAM", 6, "", ("93.184.216.34", 80))]
|
|
with (
|
|
patch("socket.getaddrinfo", return_value=mock_results),
|
|
patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}),
|
|
patch(
|
|
"daemon.handlers.acme._get_external_ip", return_value="93.184.216.34"
|
|
),
|
|
):
|
|
passed, msg = _check_dns_resolves("example.com")
|
|
assert passed is True
|
|
assert "NAT" in msg
|
|
|
|
def test_external_ip_mismatch(self):
|
|
mock_results = [("AF_INET", "STREAM", 6, "", ("93.184.216.99", 80))]
|
|
with (
|
|
patch("socket.getaddrinfo", return_value=mock_results),
|
|
patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}),
|
|
patch(
|
|
"daemon.handlers.acme._get_external_ip", return_value="93.184.216.34"
|
|
),
|
|
):
|
|
passed, msg = _check_dns_resolves("example.com")
|
|
assert passed is False
|
|
assert "93.184.216.99" in msg
|
|
assert "93.184.216.34" in msg
|
|
|
|
def test_external_ip_unavailable(self):
|
|
mock_results = [("AF_INET", "STREAM", 6, "", ("93.184.216.34", 80))]
|
|
with (
|
|
patch("socket.getaddrinfo", return_value=mock_results),
|
|
patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}),
|
|
patch("daemon.handlers.acme._get_external_ip", return_value=None),
|
|
):
|
|
passed, msg = _check_dns_resolves("example.com")
|
|
assert passed is False
|
|
assert "Cannot verify via external IP" in msg
|
|
|
|
def test_private_ip(self):
|
|
mock_results = [("AF_INET", "STREAM", 6, "", ("192.168.1.50", 80))]
|
|
with (
|
|
patch("socket.getaddrinfo", return_value=mock_results),
|
|
patch("daemon.handlers.acme._get_local_ips", return_value={"10.0.0.1"}),
|
|
patch("daemon.handlers.acme._get_external_ip", return_value=None),
|
|
):
|
|
passed, msg = _check_dns_resolves("example.com")
|
|
assert passed is False
|
|
assert "private IP" in msg
|
|
|
|
def test_unresolved(self):
|
|
import socket
|
|
|
|
with patch("socket.getaddrinfo", side_effect=socket.gaierror("NXDOMAIN")):
|
|
passed, msg = _check_dns_resolves("nonexistent.example")
|
|
assert passed is False
|
|
assert "NXDOMAIN" in msg
|
|
|
|
|
|
class TestCheckAccountRegistered:
|
|
def test_registered_with_keys(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='user@example.com'\nACME_MCA='letsencrypt'\n"
|
|
)
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
passed, msg = _check_account_registered()
|
|
assert passed is True
|
|
assert "registered" in msg
|
|
|
|
def test_missing_account_conf(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
passed, msg = _check_account_registered()
|
|
assert passed is False
|
|
assert "Register" in msg
|
|
|
|
def test_incomplete_account_conf(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
(acme_dir / ".account.conf").write_text("ACME_LEEMAIL='user@example.com'\n")
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
passed, msg = _check_account_registered()
|
|
assert passed is False
|
|
assert "Register" in msg
|
|
|
|
|
|
class TestValidateAccountCheck:
|
|
def test_account_not_registered_blocks(self):
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_installed", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_openssl_available",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_home_writable",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_account_registered",
|
|
return_value=(
|
|
False,
|
|
"Register an ACME account before issuing certificates",
|
|
),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_email_configured",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_account", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_running", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_config", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_challenge_config",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_port_80_listening",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_firewall_port_80",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_domain_format", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_existing_cert", return_value=(True, "ok")
|
|
),
|
|
):
|
|
result = _validate("example.com")
|
|
|
|
assert result["ready"] is False
|
|
acct = next(c for c in result["checks"] if c["name"] == "account_registered")
|
|
assert acct["passed"] is False
|
|
assert acct["blocking"] is True
|
|
|
|
def test_account_registered_passes(self):
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_installed", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_openssl_available",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_home_writable",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_account_registered",
|
|
return_value=(True, "ACME account is registered"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_email_configured",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_account", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_running", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_config", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_challenge_config",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_port_80_listening",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_firewall_port_80",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_domain_format", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_existing_cert", return_value=(True, "ok")
|
|
),
|
|
):
|
|
result = _validate("example.com")
|
|
|
|
assert result["ready"] is True
|
|
acct = next(c for c in result["checks"] if c["name"] == "account_registered")
|
|
assert acct["passed"] is True
|
|
assert acct["blocking"] is True
|
|
|
|
def test_email_missing_non_blocking(self):
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_installed", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_openssl_available",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_home_writable",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_account_registered",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_email_configured",
|
|
return_value=(False, "Contact email not set"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_acme_account", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_running", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_nginx_config", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_challenge_config",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_port_80_listening",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_firewall_port_80",
|
|
return_value=(True, "ok"),
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_domain_format", return_value=(True, "ok")
|
|
),
|
|
patch(
|
|
"daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok")
|
|
),
|
|
patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")),
|
|
patch(
|
|
"daemon.handlers.acme._check_existing_cert", return_value=(True, "ok")
|
|
),
|
|
):
|
|
result = _validate("example.com")
|
|
|
|
assert result["ready"] is True
|
|
email_chk = next(c for c in result["checks"] if c["name"] == "email_configured")
|
|
assert email_chk["passed"] is False
|
|
assert email_chk["blocking"] is False
|
|
|
|
|
|
class TestGetAccount:
|
|
def test_registered_account(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='admin@example.com'\nACME_MCA='letsencrypt'\nACME_CERTKEYSIZE='2048'\n"
|
|
)
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
result = _get_account_info()
|
|
|
|
assert result["registered"] is True
|
|
assert result["email"] == "admin@example.com"
|
|
assert result["ca"] == "Let's Encrypt"
|
|
assert result["key_length"] == 2048
|
|
|
|
def test_unregistered_account(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
result = _get_account_info()
|
|
|
|
assert result["registered"] is False
|
|
assert result["email"] == ""
|
|
assert result["ca"] == ""
|
|
assert result["key_length"] is None
|
|
|
|
def test_zerossl_account(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='user@zerossl.com'\nACME_MCA='zerossl'\n"
|
|
)
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
result = _get_account_info()
|
|
|
|
assert result["ca"] == "ZeroSSL"
|
|
|
|
def test_get_account_endpoint(self, tmp_path):
|
|
acme_dir = tmp_path / "acme"
|
|
acme_dir.mkdir()
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n"
|
|
)
|
|
with patch("daemon.handlers.acme._ACME_HOME", acme_dir):
|
|
result = get_account(None, None)
|
|
|
|
assert result["registered"] is True
|
|
assert result["email"] == "test@example.com"
|
|
|
|
|
|
class TestRegisterAccount:
|
|
def test_success(self, tmp_path):
|
|
proj = tmp_path / "project"
|
|
proj.mkdir()
|
|
acme_dir = proj / "data" / "acme"
|
|
acme_dir.mkdir(parents=True)
|
|
config_dir = proj / "config" / "acme"
|
|
config_dir.mkdir(parents=True)
|
|
(config_dir / "config.json").write_text("{}\n")
|
|
|
|
def fake_run_acme(args):
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n"
|
|
)
|
|
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", acme_dir),
|
|
patch("daemon.handlers.acme.PROJECT_DIR", proj),
|
|
patch("daemon.handlers.acme._run_acme", side_effect=fake_run_acme),
|
|
patch("daemon.handlers.acme.refresh_state"),
|
|
):
|
|
result = register_account(
|
|
None, {"email": "test@example.com", "server": "letsencrypt"}
|
|
)
|
|
|
|
assert result["registered"] is True
|
|
assert result["email"] == "test@example.com"
|
|
assert result["ca"] == "letsencrypt"
|
|
|
|
def test_missing_email(self):
|
|
with pytest.raises(ValueError, match="email"):
|
|
register_account(None, {"foo": "bar"})
|
|
|
|
def test_server_default_letsencrypt(self, tmp_path):
|
|
proj = tmp_path / "project"
|
|
proj.mkdir()
|
|
acme_dir = proj / "data" / "acme"
|
|
acme_dir.mkdir(parents=True)
|
|
config_dir = proj / "config" / "acme"
|
|
config_dir.mkdir(parents=True)
|
|
(config_dir / "config.json").write_text("{}\n")
|
|
|
|
captured_args = []
|
|
|
|
def fake_run_acme(args):
|
|
captured_args.append(args)
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n"
|
|
)
|
|
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", acme_dir),
|
|
patch("daemon.handlers.acme.PROJECT_DIR", proj),
|
|
patch("daemon.handlers.acme._run_acme", side_effect=fake_run_acme),
|
|
patch("daemon.handlers.acme.refresh_state"),
|
|
):
|
|
register_account(None, {"email": "test@example.com"})
|
|
|
|
assert "--server" in captured_args[0]
|
|
assert "letsencrypt" in captured_args[0]
|
|
|
|
|
|
class TestDeactivateAccount:
|
|
def test_success(self, tmp_path):
|
|
proj = tmp_path / "project"
|
|
proj.mkdir()
|
|
config_dir = proj / "config" / "acme"
|
|
config_dir.mkdir(parents=True)
|
|
(config_dir / "config.json").write_text(
|
|
'{"email": "test@example.com", "ca": "letsencrypt"}\n'
|
|
)
|
|
|
|
with (
|
|
patch("daemon.handlers.acme.PROJECT_DIR", proj),
|
|
patch("daemon.handlers.acme._run_acme"),
|
|
patch("daemon.handlers.acme.refresh_state"),
|
|
):
|
|
result = deactivate_account(None, None)
|
|
|
|
assert result["email"] == ""
|
|
cfg_text = (config_dir / "config.json").read_text()
|
|
assert "email" not in cfg_text
|
|
assert "ca" not in cfg_text
|
|
|
|
def test_cleans_account_conf_files(self, tmp_path):
|
|
proj = tmp_path / "project"
|
|
proj.mkdir()
|
|
config_dir = proj / "config" / "acme"
|
|
config_dir.mkdir(parents=True)
|
|
(config_dir / "config.json").write_text(
|
|
'{"email": "test@example.com", "ca": "letsencrypt"}\n'
|
|
)
|
|
acme_dir = proj / "data" / "acme"
|
|
acme_dir.mkdir(parents=True)
|
|
(acme_dir / ".account.conf").write_text(
|
|
"ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n"
|
|
)
|
|
(acme_dir / "account.conf").write_text(
|
|
"ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n"
|
|
)
|
|
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", acme_dir),
|
|
patch("daemon.handlers.acme.PROJECT_DIR", proj),
|
|
patch("daemon.handlers.acme._run_acme"),
|
|
patch("daemon.handlers.acme.refresh_state"),
|
|
):
|
|
deactivate_account(None, None)
|
|
|
|
assert not (acme_dir / ".account.conf").is_file()
|
|
assert not (acme_dir / "account.conf").is_file()
|
|
|
|
|
|
class TestIssueCertExistingCerts:
|
|
"""Phase 3: issue_cert blocks when cert expires today (days == 0) or tomorrow (days == 1)."""
|
|
|
|
def test_days_zero_blocks(self):
|
|
"""days_until_expiry returns 0 (expires today) — should block."""
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._validate",
|
|
return_value={"ready": True, "checks": []},
|
|
),
|
|
patch(
|
|
"lib.acme.list_certs",
|
|
return_value=[{"domain": "example.com", "days_until_expiry": 0}],
|
|
),
|
|
pytest.raises(ConflictError, match="0 days remaining"),
|
|
):
|
|
asyncio.run(issue_cert(None, {"domain": "example.com"}))
|
|
|
|
def test_days_one_blocks(self):
|
|
"""days_until_expiry returns 1 (expires tomorrow) — should still block."""
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._validate",
|
|
return_value={"ready": True, "checks": []},
|
|
),
|
|
patch(
|
|
"lib.acme.list_certs",
|
|
return_value=[{"domain": "example.com", "days_until_expiry": 1}],
|
|
),
|
|
pytest.raises(ConflictError, match="1 day remaining"),
|
|
):
|
|
asyncio.run(issue_cert(None, {"domain": "example.com"}))
|
|
|
|
def test_days_negative_one_allows(self):
|
|
"""days_until_expiry returns -1 (already expired) — should not block."""
|
|
|
|
async def _fake_run_issue(req):
|
|
pass
|
|
|
|
with (
|
|
patch(
|
|
"daemon.handlers.acme._validate",
|
|
return_value={"ready": True, "checks": []},
|
|
),
|
|
patch(
|
|
"lib.acme.list_certs",
|
|
return_value=[{"domain": "example.com", "days_until_expiry": -1}],
|
|
),
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
patch(
|
|
"daemon.handlers.acme._run_issue",
|
|
new=MagicMock(side_effect=_fake_run_issue),
|
|
) as mock_run_issue,
|
|
):
|
|
result = asyncio.run(issue_cert(None, {"domain": "example.com"}))
|
|
|
|
assert result["domain"] == "example.com"
|
|
assert "request_id" in result
|
|
mock_run_issue.assert_called_once()
|
|
|
|
|
|
class TestRenewCert:
|
|
def test_requires_body(self):
|
|
with pytest.raises(ValueError, match="Request body required"):
|
|
renew_cert(None, None)
|
|
|
|
def test_requires_domain(self):
|
|
with pytest.raises(ValueError, match="'domain' is required"):
|
|
renew_cert(None, {"force": True})
|
|
|
|
def test_starts_background_task(self):
|
|
with (
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True),
|
|
patch(
|
|
"daemon.handlers.acme._run_acme", return_value="Renewed 'example.com'"
|
|
),
|
|
patch("daemon.handlers.acme.refresh_state") as mock_refresh,
|
|
):
|
|
result = _await_renew({"domain": "example.com"})
|
|
req = acme_mod._ISSUANCES[result["request_id"]]
|
|
|
|
assert result["domain"] == "example.com"
|
|
assert "request_id" in result
|
|
assert req.status == "completed"
|
|
assert [s.status for s in req.steps] == ["done", "done", "done"]
|
|
mock_refresh.assert_called_once_with(["acme"])
|
|
|
|
def test_skip_when_not_due(self):
|
|
output = (
|
|
"[Thu Aug 20 01:27:41 AM UTC 2026] Skipping. "
|
|
"Next renewal time is: 1785068261 (2026-07-26T12:17:41Z)"
|
|
)
|
|
with (
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True),
|
|
patch("daemon.handlers.acme._run_acme", return_value=output),
|
|
patch("daemon.handlers.acme.refresh_state") as mock_refresh,
|
|
):
|
|
result = _await_renew({"domain": "example.com"})
|
|
req = acme_mod._ISSUANCES[result["request_id"]]
|
|
|
|
assert req.status == "skipped"
|
|
assert req.steps[0].status == "done"
|
|
mock_refresh.assert_not_called()
|
|
|
|
def test_failure_marks_failed(self):
|
|
with (
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True),
|
|
patch(
|
|
"daemon.handlers.acme._run_acme",
|
|
side_effect=RuntimeError("acme.sh failed with exit code 1: boom"),
|
|
),
|
|
):
|
|
result = _await_renew({"domain": "example.com"})
|
|
req = acme_mod._ISSUANCES[result["request_id"]]
|
|
|
|
assert req.status == "failed"
|
|
assert req.steps[0].status == "error"
|
|
assert "boom" in req.steps[0].message
|
|
|
|
def test_force_appends_flag(self):
|
|
with (
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True),
|
|
patch("daemon.handlers.acme._run_acme", return_value="ok") as mock_acme,
|
|
patch("daemon.handlers.acme.refresh_state"),
|
|
):
|
|
_await_renew({"domain": "example.com", "force": True})
|
|
|
|
renew_args = mock_acme.call_args_list[0].args[0]
|
|
assert "--force" in renew_args
|
|
|
|
def test_dedup_running_returns_existing(self):
|
|
existing = IssueRequest(request_id="existing123", domain="example.com")
|
|
with (
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True),
|
|
patch("daemon.handlers.acme._run_acme") as mock_acme,
|
|
):
|
|
acme_mod._ISSUANCES["existing123"] = existing
|
|
result = renew_cert(None, {"domain": "example.com"})
|
|
|
|
assert result == {
|
|
"request_id": "existing123",
|
|
"domain": "example.com",
|
|
"status": "existing",
|
|
}
|
|
mock_acme.assert_not_called()
|
|
|
|
|
|
class TestGetRenewStatus:
|
|
def test_missing_id(self):
|
|
with pytest.raises(ValueError, match="'id' is required"):
|
|
get_renew_status(None, None)
|
|
with pytest.raises(ValueError, match="'id' is required"):
|
|
get_renew_status(None, {})
|
|
|
|
def test_unknown_id(self):
|
|
with (
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
pytest.raises(NotFoundError, match="not found"),
|
|
):
|
|
get_renew_status(None, {"id": "unknown"})
|
|
|
|
def test_returns_request_dict(self):
|
|
with (
|
|
patch.dict("daemon.handlers.acme._ISSUANCES", clear=True),
|
|
patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True),
|
|
patch("daemon.handlers.acme._run_acme", return_value="Renewed"),
|
|
patch("daemon.handlers.acme.refresh_state"),
|
|
):
|
|
result = _await_renew({"domain": "example.com"})
|
|
status = get_renew_status(None, {"id": result["request_id"]})
|
|
|
|
assert status["request_id"] == result["request_id"]
|
|
assert status["domain"] == "example.com"
|
|
assert status["status"] == "completed"
|
|
assert [s["name"] for s in status["steps"]] == ["renew", "deploy", "refresh"]
|
|
|
|
|
|
class TestNormalizeAcmeHome:
|
|
def test_normalize_invokes_sudo_chmod_on_files(self, tmp_path):
|
|
f1 = tmp_path / "account.conf"
|
|
f1.write_text("x")
|
|
(tmp_path / "sub").mkdir()
|
|
f2 = tmp_path / "sub" / "dom.key"
|
|
f2.write_text("x")
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", tmp_path),
|
|
patch(
|
|
"lib.common.run_proc",
|
|
return_value=MagicMock(returncode=0, stderr=""),
|
|
) as mock_proc,
|
|
):
|
|
acme_mod.normalize_acme_home()
|
|
args = mock_proc.call_args.args[0]
|
|
assert args[:2] == ["chmod", "g+rwX"]
|
|
assert set(args[2:]) == {str(f1), str(f2)}
|
|
mock_proc.assert_called_once()
|
|
|
|
def test_normalize_empty_tree_skips_sudo(self, tmp_path):
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", tmp_path),
|
|
patch("lib.common.run_proc") as mock_proc,
|
|
):
|
|
acme_mod.normalize_acme_home()
|
|
mock_proc.assert_not_called()
|
|
|
|
def test_normalize_failure_does_not_raise(self, tmp_path):
|
|
(tmp_path / "a.conf").write_text("x")
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", tmp_path),
|
|
patch(
|
|
"lib.common.run_proc",
|
|
return_value=MagicMock(returncode=1, stderr="denied"),
|
|
),
|
|
patch.object(acme_mod, "logger"),
|
|
):
|
|
acme_mod.normalize_acme_home()
|
|
|
|
def test_preflight_normalizes_before_run(self):
|
|
calls = []
|
|
with (
|
|
patch.object(
|
|
acme_mod,
|
|
"normalize_acme_home",
|
|
side_effect=lambda: calls.append("normalize"),
|
|
),
|
|
patch.object(
|
|
acme_mod,
|
|
"_run_acme",
|
|
side_effect=lambda args: calls.append("run:" + " ".join(args)) or "ok",
|
|
),
|
|
):
|
|
out = acme_mod._run_acme_preflight(["--list", "--listraw"])
|
|
assert calls == ["normalize", "run:--list --listraw"]
|
|
assert out == "ok"
|
|
|
|
|
|
class TestPreflightWiring:
|
|
def test_issue_uses_preflight(self):
|
|
source = inspect.getsource(acme_mod._run_issue)
|
|
assert "_run_acme_preflight" in source
|
|
assert "normalize_acme_home" in source
|
|
|
|
def test_renew_uses_preflight(self):
|
|
source = inspect.getsource(acme_mod._run_renew)
|
|
assert "_run_acme_preflight" in source
|
|
assert "normalize_acme_home" in source
|