feat: TLS certificate management in Vault page
Unit Tests / test (push) Successful in 7m26s

Adds live cert status, one-click ACME renewal, and custom cert upload
directly to the Vault page so users never need to touch Caddy config.

Backend:
- CaddyManager.get_cert_status() now returns domain, domain_mode, and
  cert_type so the UI can render the right controls without a separate
  identity fetch
- CaddyManager.renew_cert() reloads Caddy and invalidates the status
  cache; the frontend polls until the cert turns valid
- CaddyManager.upload_custom_cert() validates PEM, writes cert+key to
  the shared config/caddy/certs/ volume, updates identity (cert_type=custom),
  and regenerates the Caddyfile so Caddy references the new paths
- LAN-mode Caddyfile switches from /etc/caddy/internal/ to the shared
  certs dir automatically when cert_type=custom is set
- ddns_api default no longer includes /api/v1 — the plugin appends it;
  legacy /api/v1 suffix is stripped at write time to keep the Caddyfile clean
- POST /api/caddy/cert-renew and POST /api/caddy/custom-cert routes added

Frontend:
- TLSPanel component at the top of Vault.jsx shows status badge
  (valid/expiring-soon/expired/pending/internal) with domain and expiry
- Renew button visible only for ACME modes; spins during the API call
  then polls GET /api/caddy/cert-status every 10 s until valid
- Upload Custom Cert opens a modal with PEM text areas; works for all modes
- caddyAPI.renewCert() and uploadCustomCert() added to api.js

Tests: 22 new tests across 5 classes covering enriched status,
renew_cert guards, upload_custom_cert validation/writes/persistence,
custom-cert Caddyfile path selection, and ddns_api suffix stripping.
All 2093 existing tests continue to pass.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-08 12:53:42 -04:00
parent 85d265187d
commit 33d255f089
5 changed files with 1127 additions and 482 deletions
+231 -1
View File
@@ -70,7 +70,9 @@ class TestGenerateCaddyfilePicNgo(unittest.TestCase):
self.assertIn('alpha.pic.ngo', out)
# Registration token (not TOTP secret) is embedded — no {$VAR} placeholders
self.assertIn('token TESTSECRET123', out)
self.assertIn('api_base_url https://ddns.pic.ngo/api/v1', out)
# /api/v1 is stripped — the plugin appends it itself
self.assertIn('api_base_url https://ddns.pic.ngo', out)
self.assertNotIn('api_base_url https://ddns.pic.ngo/api/v1', out)
self.assertNotIn('{$PIC_NGO_DDNS_TOKEN}', out)
self.assertNotIn('{$PIC_NGO_DDNS_API}', out)
self.assertIn('email admin@alpha.pic.ngo', out)
@@ -435,5 +437,233 @@ class TestRefreshCertStatus(unittest.TestCase):
mock_ref.assert_not_called()
class TestGetCertStatusEnriched(unittest.TestCase):
"""get_cert_status() returns domain, domain_mode, cert_type alongside tls fields."""
def test_includes_domain_and_mode_for_pic_ngo(self):
mgr = _mgr(identity={
'cell_name': 'alpha',
'domain_mode': 'pic_ngo',
'tls': {'status': 'valid', 'expiry': '2026-12-01T00:00:00+00:00', 'days_remaining': 180},
})
s = mgr.get_cert_status()
self.assertEqual(s['domain_mode'], 'pic_ngo')
self.assertEqual(s['domain'], '*.alpha.pic.ngo')
self.assertEqual(s['cert_type'], 'acme')
self.assertEqual(s['status'], 'valid')
def test_cert_type_is_internal_for_lan_mode(self):
mgr = _mgr(identity={'cell_name': 'x', 'domain_mode': 'lan', 'tls': {}})
s = mgr.get_cert_status()
self.assertEqual(s['cert_type'], 'internal')
self.assertIsNone(s['domain'])
def test_cert_type_is_custom_when_tls_says_so(self):
mgr = _mgr(identity={
'cell_name': 'x',
'domain_mode': 'lan',
'tls': {'cert_type': 'custom', 'status': 'valid',
'expiry': '2027-01-01T00:00:00+00:00', 'days_remaining': 200},
})
s = mgr.get_cert_status()
self.assertEqual(s['cert_type'], 'custom')
def test_domain_label_cloudflare(self):
ident = {'domain_mode': 'cloudflare', 'domain_name': 'example.com'}
self.assertEqual(CaddyManager._domain_label(ident), '*.example.com')
def test_domain_label_duckdns(self):
ident = {'cell_name': 'beta', 'domain_mode': 'duckdns'}
self.assertEqual(CaddyManager._domain_label(ident), '*.beta.duckdns.org')
def test_domain_label_http01(self):
ident = {'domain_mode': 'http01', 'domain_name': 'myhost.noip.me'}
self.assertEqual(CaddyManager._domain_label(ident), 'myhost.noip.me')
def test_domain_label_lan_is_none(self):
self.assertIsNone(CaddyManager._domain_label({'domain_mode': 'lan'}))
class TestRenewCert(unittest.TestCase):
"""renew_cert() — mode guard, reload call, cache invalidation."""
def test_lan_mode_returns_error(self):
mgr = _mgr(identity={'domain_mode': 'lan'})
result = mgr.renew_cert()
self.assertFalse(result['ok'])
self.assertIn('LAN', result['error'])
def test_acme_mode_calls_reload(self):
mgr = _mgr(identity={'domain_mode': 'pic_ngo'})
with patch.object(mgr, 'reload_caddy', return_value=True) as mock_reload:
result = mgr.renew_cert()
mock_reload.assert_called_once()
self.assertTrue(result['ok'])
self.assertEqual(result['status'], 'pending')
def test_reload_failure_propagated(self):
mgr = _mgr(identity={'domain_mode': 'cloudflare'})
with patch.object(mgr, 'reload_caddy', return_value=False):
result = mgr.renew_cert()
self.assertFalse(result['ok'])
self.assertIn('reload failed', result['error'])
def test_invalidates_cache_on_success(self):
import time
mgr = _mgr(identity={'domain_mode': 'pic_ngo'})
mgr._cert_refreshed_at = time.monotonic()
with patch.object(mgr, 'reload_caddy', return_value=True):
mgr.renew_cert()
self.assertIsNone(mgr._cert_refreshed_at)
class TestUploadCustomCert(unittest.TestCase):
"""upload_custom_cert() — validation, file writes, identity persistence, Caddyfile regen."""
def _make_pem_cert(self, days_remaining: int = 90):
"""Return (cert_pem, key_pem) for a self-signed cert."""
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import datetime
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
now = datetime.datetime.now(datetime.timezone.utc)
expiry = now + datetime.timedelta(days=days_remaining)
not_before = (now - datetime.timedelta(days=abs(days_remaining) + 10)
if days_remaining < 0 else now - datetime.timedelta(days=1))
cert = (
x509.CertificateBuilder()
.subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'test.example.com')]))
.issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'test.example.com')]))
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(not_before)
.not_valid_after(expiry)
.sign(key, hashes.SHA256())
)
cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
key_pem = key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.TraditionalOpenSSL,
serialization.NoEncryption(),
).decode()
return cert_pem, key_pem
def test_rejects_invalid_cert_pem(self):
mgr = _mgr()
result = mgr.upload_custom_cert('not a cert', '-----BEGIN PRIVATE KEY-----\nXXX\n-----END PRIVATE KEY-----')
self.assertFalse(result['ok'])
self.assertIn('Invalid certificate', result['error'])
def test_rejects_invalid_key_pem(self):
mgr = _mgr()
cert_pem, _ = self._make_pem_cert()
result = mgr.upload_custom_cert(cert_pem, 'not a key')
self.assertFalse(result['ok'])
self.assertIn('Invalid private key', result['error'])
def test_writes_files_to_certs_dir(self):
mgr = _mgr(identity={'domain_mode': 'lan', 'cell_name': 'x'})
cert_pem, key_pem = self._make_pem_cert()
written = {}
def fake_open(path, mode='r', **kw):
import unittest.mock
m = unittest.mock.mock_open()()
if 'w' in mode:
written[path] = True
return m
with patch('builtins.open', side_effect=fake_open):
with patch('os.makedirs'):
with patch.object(mgr, 'regenerate_with_installed', return_value=True):
mgr.upload_custom_cert(cert_pem, key_pem)
self.assertTrue(any('cert.pem' in p for p in written))
self.assertTrue(any('key.pem' in p for p in written))
def test_persists_custom_cert_type_to_identity(self):
mgr = _mgr(identity={'domain_mode': 'lan', 'cell_name': 'x'})
cert_pem, key_pem = self._make_pem_cert(days_remaining=90)
with patch('builtins.open', unittest.mock.mock_open()):
with patch('os.makedirs'):
with patch.object(mgr, 'regenerate_with_installed', return_value=True):
result = mgr.upload_custom_cert(cert_pem, key_pem)
self.assertTrue(result['ok'])
self.assertEqual(result['cert_type'], 'custom')
self.assertEqual(result['status'], 'valid')
mgr.config_manager.set_identity_field.assert_called_once()
call_args = mgr.config_manager.set_identity_field.call_args
self.assertEqual(call_args[0][0], 'tls')
self.assertEqual(call_args[0][1]['cert_type'], 'custom')
def test_expired_cert_flagged_as_expired(self):
mgr = _mgr(identity={'domain_mode': 'lan', 'cell_name': 'x'})
cert_pem, key_pem = self._make_pem_cert(days_remaining=-5)
with patch('builtins.open', unittest.mock.mock_open()):
with patch('os.makedirs'):
with patch.object(mgr, 'regenerate_with_installed', return_value=True):
result = mgr.upload_custom_cert(cert_pem, key_pem)
self.assertEqual(result['status'], 'expired')
def test_file_write_failure_returns_error(self):
mgr = _mgr(identity={'domain_mode': 'lan'})
cert_pem, key_pem = self._make_pem_cert()
with patch('os.makedirs'):
with patch('builtins.open', side_effect=OSError('no space')):
result = mgr.upload_custom_cert(cert_pem, key_pem)
self.assertFalse(result['ok'])
self.assertIn('Failed to write', result['error'])
class TestCaddyfileLanCustomCert(unittest.TestCase):
"""_caddyfile_lan() uses the custom cert path when cert_type=custom."""
def test_default_uses_internal_cert_path(self):
mgr = _mgr(identity={'cell_name': 'mycell', 'domain_mode': 'lan'})
out = mgr.generate_caddyfile({'cell_name': 'mycell', 'domain_mode': 'lan'}, [])
self.assertIn('/etc/caddy/internal/cert.pem', out)
def test_custom_cert_type_uses_shared_cert_path(self):
mgr = _mgr(identity={
'cell_name': 'mycell',
'domain_mode': 'lan',
'tls': {'cert_type': 'custom'},
})
out = mgr.generate_caddyfile({'cell_name': 'mycell', 'domain_mode': 'lan'}, [])
self.assertIn('/config/caddy/certs/cert.pem', out)
self.assertNotIn('/etc/caddy/internal/cert.pem', out)
class TestDdnsApiStripsLegacySuffix(unittest.TestCase):
"""_caddyfile_pic_ngo strips /api/v1 from ddns_api so the plugin doesn't double it."""
def test_api_v1_suffix_stripped_from_config_url(self):
mgr = _mgr()
mgr.config_manager.configs = {
'ddns': {'token': 'tok', 'url': 'https://ddns.pic.ngo/api/v1'},
}
with patch.dict(os.environ, {}, clear=False):
os.environ.pop('DDNS_URL', None)
out = mgr.generate_caddyfile({'cell_name': 'x', 'domain_mode': 'pic_ngo'}, [])
self.assertIn('api_base_url https://ddns.pic.ngo', out)
self.assertNotIn('api_base_url https://ddns.pic.ngo/api/v1', out)
def test_clean_url_is_unchanged(self):
mgr = _mgr()
mgr.config_manager.configs = {
'ddns': {'token': 'tok', 'url': 'https://ddns.pic.ngo'},
}
with patch.dict(os.environ, {}, clear=False):
os.environ.pop('DDNS_URL', None)
out = mgr.generate_caddyfile({'cell_name': 'x', 'domain_mode': 'pic_ngo'}, [])
self.assertIn('api_base_url https://ddns.pic.ngo', out)
if __name__ == '__main__':
unittest.main()