275 lines
14 KiB
Python
275 lines
14 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SPEC = importlib.util.spec_from_file_location('guard', Path(__file__).parents[1] / 'dashboard-public-guard.py')
|
|
guard = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(guard)
|
|
SOURCE = '''# quoted braces must not confuse the parser
|
|
server { listen 80 default_server; server_name _; location / { return 200 "{}"; } }
|
|
server { listen 443 ssl default_server; server_name _; location / { return 200 "a}"; } }
|
|
server { listen 80; server_name public.example; location / { return 200 "app"; } }
|
|
'''
|
|
|
|
|
|
class GuardTests(unittest.TestCase):
|
|
def test_active_site_copy_and_symlink_target_are_resolved(self):
|
|
for symlink in [False, True]:
|
|
with self.subTest(symlink=symlink), tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
available = root / 'sites-available/archipelago'
|
|
enabled = root / 'sites-enabled/archipelago'
|
|
available.parent.mkdir(); enabled.parent.mkdir()
|
|
available.write_text(SOURCE)
|
|
if symlink:
|
|
enabled.symlink_to(available)
|
|
else:
|
|
enabled.write_text(SOURCE + '# active custom copy\n')
|
|
active = guard.active_dashboard(root)
|
|
self.assertEqual(active, available if symlink else enabled)
|
|
before = available.read_bytes()
|
|
def command(args, **kwargs):
|
|
return subprocess.CompletedProcess(args, 0)
|
|
guard.apply(active, command, root / 'lock')
|
|
self.assertIn(guard.BEGIN, enabled.read_text())
|
|
self.assertEqual(enabled.is_symlink(), symlink)
|
|
if not symlink:
|
|
self.assertEqual(available.read_bytes(), before)
|
|
|
|
def test_all_defaults_guarded_named_apps_untouched_idempotent(self):
|
|
updated = guard.guarded(SOURCE)
|
|
self.assertEqual(updated.count(guard.CHECK), 2)
|
|
self.assertIn(SOURCE.splitlines()[-1], updated)
|
|
self.assertEqual(guard.guarded(updated), updated)
|
|
self.assertIn('geo $realip_remote_addr', updated)
|
|
|
|
def test_incomplete_and_ambiguous_config_rejected(self):
|
|
for source in [SOURCE.replace('listen 443 ssl default_server;', 'listen 8443 ssl;'),
|
|
SOURCE + '\n' + guard.BEGIN, SOURCE + '\nserver {',
|
|
guard.END + '\n' + guard.BEGIN + '\n' + SOURCE]:
|
|
with self.assertRaises(ValueError):
|
|
guard.guarded(source)
|
|
|
|
def test_legacy_address_specific_https_dashboard(self):
|
|
source = SOURCE.replace('listen 443 ssl default_server;', 'listen 192.168.1.10:443 ssl;')
|
|
updated = guard.guarded(source)
|
|
self.assertEqual(updated.count(guard.CHECK), 2)
|
|
self.assertEqual(guard.guarded(updated), updated)
|
|
self.assertIn(SOURCE.splitlines()[-1], updated)
|
|
|
|
def test_syntax_and_reload_failure_restore_exact_previous_bytes(self):
|
|
for failure in ['nginx', 'systemctl']:
|
|
with self.subTest(failure=failure), tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'archipelago'
|
|
path.write_text(SOURCE)
|
|
calls = []
|
|
def command(args, **kwargs):
|
|
calls.append(args)
|
|
# Only the candidate's first matching command fails.
|
|
fail = args[0] == failure and sum(x[0] == failure for x in calls) == 1
|
|
return subprocess.CompletedProcess(args, int(fail))
|
|
with self.assertRaises(RuntimeError):
|
|
guard.apply(path, command, Path(tmp) / 'nginx.lock')
|
|
self.assertEqual(path.read_text(), SOURCE)
|
|
backups = list(Path(tmp).glob('*.before-management-guard-*'))
|
|
self.assertEqual(len(backups), 1)
|
|
self.assertEqual(backups[0].read_text(), SOURCE)
|
|
self.assertEqual(backups[0].stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(calls[-1], ['systemctl', 'reload', 'nginx'])
|
|
|
|
def test_second_application_does_not_reload(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'archipelago'
|
|
path.write_text(SOURCE)
|
|
calls = []
|
|
def command(args, **kwargs):
|
|
calls.append(args)
|
|
return subprocess.CompletedProcess(args, 0)
|
|
self.assertTrue(guard.apply(path, command, Path(tmp) / 'nginx.lock'))
|
|
self.assertFalse(guard.apply(path, command, Path(tmp) / 'nginx.lock'))
|
|
self.assertEqual(len(calls), 2)
|
|
|
|
def test_legacy_runtime_install_is_guarded_before_any_validation_or_reload(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'active'
|
|
source = Path(tmp) / 'legacy-runtime'
|
|
path.write_text(guard.guarded(SOURCE))
|
|
source.write_text(SOURCE + '# legacy runtime revision\n')
|
|
calls = []
|
|
def command(args, **kwargs):
|
|
# Even the first nginx -t must see a complete guarded candidate.
|
|
current = path.read_text()
|
|
self.assertEqual(current.count(guard.CHECK), 2)
|
|
self.assertEqual(guard.guarded(current), current)
|
|
self.assertIn('# legacy runtime revision', current)
|
|
calls.append(args)
|
|
return subprocess.CompletedProcess(args, 0)
|
|
self.assertTrue(guard.apply(path, command, Path(tmp) / 'lock', source))
|
|
self.assertFalse(guard.apply(path, command, Path(tmp) / 'lock', source))
|
|
self.assertEqual(len(calls), 2)
|
|
self.assertEqual(source.read_text(), SOURCE + '# legacy runtime revision\n')
|
|
|
|
def test_invalid_runtime_never_replaces_protected_site(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'active'
|
|
source = Path(tmp) / 'legacy-runtime'
|
|
previous = guard.guarded(SOURCE)
|
|
path.write_text(previous)
|
|
source.write_text('server { listen 80 default_server; server_name _; }')
|
|
def command(*args, **kwargs):
|
|
self.fail('invalid candidate must be rejected before any command')
|
|
with self.assertRaises(ValueError):
|
|
guard.apply(path, command, Path(tmp) / 'lock', source)
|
|
self.assertEqual(path.read_text(), previous)
|
|
|
|
def test_runtime_validation_or_reload_failure_preserves_previous_guard(self):
|
|
for failure in ['nginx', 'systemctl']:
|
|
with self.subTest(failure=failure), tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'active'
|
|
source = Path(tmp) / 'legacy-runtime'
|
|
previous = guard.guarded(SOURCE)
|
|
path.write_text(previous)
|
|
source.write_text(SOURCE + '# incoming revision\n')
|
|
calls = []
|
|
def command(args, **kwargs):
|
|
self.assertEqual(path.read_text().count(guard.CHECK), 2)
|
|
calls.append(args)
|
|
fail = args[0] == failure and sum(x[0] == failure for x in calls) == 1
|
|
return subprocess.CompletedProcess(args, int(fail))
|
|
with self.assertRaises(RuntimeError):
|
|
guard.apply(path, command, Path(tmp) / 'lock', source)
|
|
self.assertEqual(path.read_text(), previous)
|
|
|
|
def test_runtime_bootstrap_uses_guarded_installer_instead_of_raw_copy(self):
|
|
# Wiring matters: a safe helper does not help if bootstrap bypasses it.
|
|
source = (Path(__file__).parents[2] / 'core/archipelago/src/bootstrap.rs').read_text()
|
|
block = source.split('let nginx_src = configs.join("nginx-archipelago.conf");', 1)[1].split('// archipelago-host-secrets-audit', 1)[0]
|
|
self.assertIn('scripts/dashboard-public-guard.py', block)
|
|
self.assertIn('"--install-runtime"', block)
|
|
self.assertNotIn('"install",', block)
|
|
|
|
def test_legacy_template_protection_is_idempotent_and_rollback_uses_independent_owner(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
template = Path(tmp) / 'legacy.conf'
|
|
template.write_text(SOURCE)
|
|
template.chmod(0o640)
|
|
self.assertTrue(guard.protect_template(template))
|
|
self.assertEqual(template.read_text(), guard.guarded(SOURCE))
|
|
self.assertEqual(template.stat().st_mode & 0o777, 0o640)
|
|
self.assertFalse(guard.protect_template(template))
|
|
invalid = 'server { listen 80 default_server; }'
|
|
template.write_text(invalid)
|
|
with self.assertRaises(ValueError):
|
|
guard.protect_template(template)
|
|
self.assertEqual(template.read_text(), invalid)
|
|
source = (Path(__file__).parents[2] / 'core/archipelago/src/update.rs').read_text()
|
|
rollback = source.split('pub async fn rollback_update', 1)[1].split('\n}', 1)[0]
|
|
# Public recovery now restores the retained code/routing transaction.
|
|
# No old in-process raw-copy rollback may bypass that owner. Actual
|
|
# guarded routing/template retention is exercised by RuntimeTransaction.
|
|
self.assertIn('return public_admission::request_recovery().await;', rollback)
|
|
self.assertNotIn('host_sudo(&["cp"', rollback)
|
|
|
|
|
|
class RuntimePreservationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.active = self.root / 'active'
|
|
self.template = self.root / 'runtime.conf'
|
|
self.template.write_text(SOURCE)
|
|
self.calls = []
|
|
|
|
def command(self, args, **kwargs):
|
|
self.calls.append(args)
|
|
return subprocess.CompletedProcess(args, 0)
|
|
|
|
def install(self, command=None):
|
|
return guard.install_runtime(self.active, self.template, command or self.command, self.root / 'lock')
|
|
|
|
def test_fresh_site_installs_guarded_managed_template(self):
|
|
self.assertTrue(self.install())
|
|
self.assertEqual(self.active.read_text(), guard.guarded(SOURCE))
|
|
self.assertEqual(self.template.read_bytes(), self.active.read_bytes())
|
|
self.assertIn(['systemctl', 'reload', 'nginx'], self.calls)
|
|
|
|
def test_recognized_current_template_is_idempotent_without_reload(self):
|
|
content = guard.guarded(SOURCE)
|
|
self.active.write_text(content); self.template.write_text(content)
|
|
self.assertFalse(self.install())
|
|
self.assertEqual(self.calls, [['nginx', '-t']])
|
|
|
|
def test_custom_current_routes_survive_and_old_installer_cannot_revert(self):
|
|
custom = SOURCE.replace('return 200 "{}";', 'return 200 "current";') + '''
|
|
# retained rental/peer/signer/NPM routing
|
|
server { listen 80; server_name custom.example; location /api/rental-playback/ { return 401; }
|
|
location /api/peer-content/ { return 403; } location /nostr-signer { return 200 "signer"; } }
|
|
'''
|
|
self.active.write_text(guard.guarded(custom)); before = self.active.read_bytes()
|
|
self.assertTrue(self.install())
|
|
self.assertEqual(self.active.read_bytes(), before)
|
|
self.assertEqual(self.template.read_bytes(), before)
|
|
self.assertFalse(guard.apply(self.active, self.command, self.root / 'lock', self.template))
|
|
backups = list(self.root.glob('runtime-template.before-preservation-*'))
|
|
self.assertEqual(len(backups), 1); self.assertEqual(backups[0].read_text(), SOURCE)
|
|
self.assertEqual(backups[0].stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_security_guard_is_not_skipped_for_custom_site(self):
|
|
self.active.write_text(SOURCE + '# custom\n')
|
|
self.install()
|
|
self.assertEqual(self.active.read_text().count(guard.CHECK), 2)
|
|
self.assertIn('# custom', self.active.read_text())
|
|
|
|
def test_explicit_managed_migration_remains_available(self):
|
|
self.active.write_text(guard.guarded(SOURCE + '# old managed\n'))
|
|
self.template.write_text(SOURCE + '# intentional new managed revision\n')
|
|
self.assertTrue(guard.apply(self.active, self.command, self.root / 'lock', self.template))
|
|
self.assertIn('# intentional new managed revision', self.active.read_text())
|
|
|
|
def test_invalid_active_does_not_use_runtime_as_repair(self):
|
|
self.active.write_text('server { listen 80; }')
|
|
with self.assertRaises(ValueError): self.install()
|
|
self.assertEqual(self.active.read_text(), 'server { listen 80; }')
|
|
self.assertEqual(self.template.read_text(), SOURCE)
|
|
|
|
def test_failed_syntax_or_reload_restores_active_and_retains_runtime(self):
|
|
for failure in ('nginx', 'systemctl'):
|
|
with self.subTest(failure=failure):
|
|
self.active.write_text(SOURCE + '# custom\n'); self.calls = []
|
|
before = self.active.read_bytes()
|
|
def command(args, **kwargs):
|
|
self.calls.append(args)
|
|
fail = args[0] == failure and sum(c[0] == failure for c in self.calls) == 1
|
|
return subprocess.CompletedProcess(args, int(fail))
|
|
with self.assertRaises(RuntimeError): self.install(command)
|
|
self.assertEqual(self.active.read_bytes(), before)
|
|
self.assertEqual(self.template.read_text(), SOURCE)
|
|
|
|
def test_guarded_but_invalid_nginx_never_enters_rollback_template(self):
|
|
self.active.write_text(guard.guarded(SOURCE + '\ninvalid_directive;\n'))
|
|
with self.assertRaises(RuntimeError):
|
|
self.install(lambda args, **kw: subprocess.CompletedProcess(args, 1))
|
|
self.assertEqual(self.template.read_text(), SOURCE)
|
|
|
|
def test_fresh_failure_removes_only_new_active_file(self):
|
|
for failure in ('nginx', 'systemctl'):
|
|
self.calls = []
|
|
def command(args, **kwargs):
|
|
self.calls.append(args)
|
|
fail = args[0] == failure and sum(c[0] == failure for c in self.calls) == 1
|
|
return subprocess.CompletedProcess(args, int(fail))
|
|
with self.assertRaises(RuntimeError): self.install(command)
|
|
self.assertFalse(self.active.exists()); self.assertEqual(self.template.read_text(), SOURCE)
|
|
|
|
def test_dangling_enabled_link_is_not_treated_as_fresh_site(self):
|
|
enabled = self.root / 'sites-enabled/archipelago'; enabled.parent.mkdir()
|
|
enabled.symlink_to(self.root / 'missing')
|
|
with self.assertRaises(FileNotFoundError): guard.runtime_dashboard(self.root)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|