mteehan
5025dfaf30
- Register, view, and deactivate ACME accounts via API and UI - 16-check validation framework for certificate issuance readiness - DNS resolution, port, nginx, and firewall pre-flight checks - External IP detection with NAT support and fallback providers - Account card and settings modal in certificates page - Guard certificate issuance behind account registration - Update modal CSS to overlay-based approach - 1000+ lines of tests for validation and account handlers
1091 lines
39 KiB
Python
1091 lines
39 KiB
Python
"""Tests for daemon/handlers/acme.py — handler endpoint logic."""
|
|
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from daemon.handlers.acme import (
|
|
_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,
|
|
register_account,
|
|
)
|
|
|
|
|
|
class TestGenerateSelfSigned:
|
|
def test_generate_creates_files(self, tmp_path):
|
|
with (
|
|
patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"),
|
|
):
|
|
result = generate_self_signed(None, {"domain": "test.local"})
|
|
|
|
assert result["domain"] == "test.local"
|
|
assert result["generated"] is True
|
|
cert_dir = tmp_path / "acme" / "test.local"
|
|
assert result["cert"] == str(cert_dir / "fullchain.cer")
|
|
assert result["key"] == str(cert_dir / "test.local.key")
|
|
assert (cert_dir / "fullchain.cer").is_file()
|
|
assert (cert_dir / "test.local.key").is_file()
|
|
|
|
def test_generate_idempotent_skips_existing(self, tmp_path):
|
|
cert_dir = tmp_path / "acme" / "test.local"
|
|
cert_dir.mkdir(parents=True)
|
|
(cert_dir / "fullchain.cer").write_text("dummy-cert")
|
|
(cert_dir / "test.local.key").write_text("dummy-key")
|
|
|
|
with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"):
|
|
result = generate_self_signed(None, {"domain": "test.local"})
|
|
|
|
assert result["generated"] is False
|
|
|
|
def test_generate_partial_existing(self, tmp_path):
|
|
cert_dir = tmp_path / "acme" / "test.local"
|
|
cert_dir.mkdir(parents=True)
|
|
(cert_dir / "fullchain.cer").write_text("dummy-cert")
|
|
# key missing -> should regenerate
|
|
|
|
with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"):
|
|
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._ACME_HOME", tmp_path / "acme"),
|
|
patch("subprocess.run") as mock_run,
|
|
):
|
|
|
|
def _create_files(*args, **kwargs):
|
|
cert_dir = tmp_path / "acme" / "test.local"
|
|
cert_dir.mkdir(parents=True, exist_ok=True)
|
|
(cert_dir / "fullchain.cer").touch()
|
|
(cert_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"
|
|
|
|
cert_dir = tmp_path / "acme" / "test.local"
|
|
if (cert_dir / "fullchain.cer").is_file():
|
|
assert cert_dir.is_dir()
|
|
|
|
def test_generate_creates_directory(self, tmp_path):
|
|
with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"):
|
|
generate_self_signed(None, {"domain": "test.local"})
|
|
|
|
assert (tmp_path / "acme" / "test.local").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, msg = _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()
|
|
|
|
pid_file = Path("/var/run/nginx.pid")
|
|
with patch("subprocess.run", side_effect=run_side_effect):
|
|
with patch.object(Path, "is_file") as mock_is_file:
|
|
with 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, msg = _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, msg = _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, msg = _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, msg = _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, msg = _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, msg = _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, msg = _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, msg = _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, msg = _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, msg = _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, msg = _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 192.168.1.1"
|
|
)
|
|
with (
|
|
patch("socket.gethostbyname", return_value="192.168.1.1"),
|
|
patch("subprocess.run", return_value=mock_result),
|
|
):
|
|
passed, msg = _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("socket.gethostbyname", return_value="192.168.1.1"),
|
|
patch("subprocess.run", return_value=mock_result),
|
|
):
|
|
passed, msg = _check_dns_public("example.com")
|
|
assert passed is False
|
|
|
|
|
|
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 = [c for c in result["checks"] if c["name"] == "nginx_running"][0]
|
|
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 = [c for c in result["checks"] if c["name"] == "dns_public"][0]
|
|
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()
|