172 lines
8.4 KiB
Python
172 lines
8.4 KiB
Python
"""Setup tests use temporary files and isolated mock commands, never a real firewall."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
import watchdog as w
|
|
|
|
|
|
def module_from(filename, name):
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / filename)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class OracleRendererTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.renderer = module_from("configure-oracle.py", "configure_oracle")
|
|
|
|
def test_sni_domain_is_escaped_and_bounded(self):
|
|
# Public resolver IP is only a validation fixture; no socket is opened.
|
|
result = self.renderer.render("Example.COM.", "8.8.8.8")
|
|
expression = next(line.split()[-1] for line in result.splitlines() if "acl allowed_sni" in line)
|
|
for good in ("example.com", "app.example.com", "a.b.example.com"):
|
|
self.assertIsNotNone(re.fullmatch(expression, good))
|
|
for bad in ("exampleXcom", "example.com.attacker.test", "other.test", "notexample.com"):
|
|
self.assertIsNone(re.fullmatch(expression, bad))
|
|
self.assertIn("server home 8.8.8.8:443", result)
|
|
self.assertNotIn("@@", result)
|
|
executable = "\n".join(line for line in result.splitlines() if not line.lstrip().startswith("#"))
|
|
self.assertNotRegex(executable, r"\b(?:ssl|crt|send-proxy)\b")
|
|
|
|
def test_injection_and_invalid_destinations_rejected(self):
|
|
for domain, address in (("example.com\nbind :80", "8.8.8.8"),
|
|
("*.example.com", "8.8.8.8"),
|
|
("example.com", "127.0.0.1"),
|
|
("example.com", "203.0.113.10"),
|
|
("example.com", "8.8.8.8:443")):
|
|
with self.subTest(domain=domain, address=address), self.assertRaises(w.SafetyError):
|
|
self.renderer.render(domain, address)
|
|
|
|
def test_rendered_file_is_private_and_existing_file_replaced_atomically(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
destination = Path(directory) / "private" / "haproxy.cfg"
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
self.renderer.main(["--domain", "example.com", "--home-ip", "8.8.8.8", "--output", str(destination)])
|
|
self.renderer.main(["--domain", "example.net", "--home-ip", "1.1.1.1", "--output", str(destination)])
|
|
self.assertEqual(destination.stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(destination.parent.stat().st_mode & 0o777, 0o700)
|
|
self.assertIn(r"example\.net", destination.read_text())
|
|
self.assertEqual([path.name for path in destination.parent.iterdir()], ["haproxy.cfg"])
|
|
|
|
|
|
class HomeWizardTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.wizard = module_from("configure-home.py", "configure_home_general")
|
|
|
|
def test_selected_apps_and_custom_canary_generate_only_requested_references(self):
|
|
records = [
|
|
{"id": "1" * 32, "type": "A", "name": "check.example.net", "content": "1.1.1.1", "proxied": True, "ttl": 1},
|
|
{"id": "2" * 32, "type": "A", "name": "app.example.net", "content": "8.8.8.8", "proxied": False, "ttl": 60},
|
|
{"id": "3" * 32, "type": "A", "name": "example.net", "content": "8.8.8.8", "proxied": False, "ttl": 60},
|
|
]
|
|
api = mock.Mock()
|
|
api.inventory.return_value = records
|
|
with tempfile.TemporaryDirectory() as directory, \
|
|
mock.patch.object(self.wizard, "__file__", str(Path(directory) / "configure-home.py")), \
|
|
mock.patch.object(sys, "argv", ["configure-home.py", "--local", "--health-host", "check.example.net", "--app", "app.example.net"]), \
|
|
mock.patch("builtins.input", side_effect=["example.net", "8.8.8.8", "1.1.1.1", "f" * 32, "10.50.0.10"]), \
|
|
mock.patch.object(self.wizard.getpass, "getpass", return_value="fixture-token"), \
|
|
mock.patch.object(self.wizard, "Cloudflare", return_value=api), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.wizard.main()
|
|
cfg = json.loads((Path(directory) / "runtime/watchdog.json").read_text())
|
|
self.assertEqual(cfg["canary"]["name"], "check.example.net")
|
|
self.assertEqual(cfg["health"]["primary_host"], "check.example.net")
|
|
self.assertEqual(cfg["apps"], [{"id": "2" * 32, "name": "app.example.net"}])
|
|
self.assertEqual(api.mock_calls, [mock.call.inventory()])
|
|
|
|
def test_outside_domain_is_rejected_before_token_or_api_access(self):
|
|
for option in ("--app", "--health-host"):
|
|
with self.subTest(option=option), tempfile.TemporaryDirectory() as directory, \
|
|
mock.patch.object(self.wizard, "__file__", str(Path(directory) / "configure-home.py")), \
|
|
mock.patch.object(sys, "argv", ["configure-home.py", "--local", option, "attacker.test"]), \
|
|
mock.patch("builtins.input", side_effect=["example.net", "8.8.8.8", "1.1.1.1"]), \
|
|
mock.patch.object(self.wizard.getpass, "getpass") as token, \
|
|
mock.patch.object(self.wizard, "Cloudflare") as api:
|
|
with self.assertRaises(w.SafetyError):
|
|
self.wizard.main()
|
|
token.assert_not_called()
|
|
api.assert_not_called()
|
|
self.assertFalse((Path(directory) / "runtime/cf-token").exists())
|
|
|
|
|
|
class FirewallPreflightTests(unittest.TestCase):
|
|
def run_profile(self, profile):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
binary = Path(directory)
|
|
(binary / "python3").symlink_to(sys.executable)
|
|
dispatcher = """#!/usr/bin/env python3
|
|
import json, os, sys
|
|
from pathlib import Path
|
|
name = Path(sys.argv[0]).name
|
|
profile = os.environ['PREFLIGHT_PROFILE']
|
|
with open(os.environ['PREFLIGHT_CALLS'], 'a') as stream:
|
|
stream.write(name + ' ' + ' '.join(sys.argv[1:]) + '\\n')
|
|
if name == 'systemctl':
|
|
assert sys.argv[1:3] == ['is-active', '--quiet']
|
|
sys.exit(0 if profile == 'active' and sys.argv[-1] == 'ufw.service' else 3)
|
|
if name == 'dpkg-query':
|
|
print('install ok installed' if profile == 'installed' else '')
|
|
elif name in ('iptables-save', 'ip6tables-save'):
|
|
print('*filter\\n:INPUT ACCEPT [0:0]\\n' + (':ufw-before-input - [0:0]\\n' if profile == 'leftover' else '') + 'COMMIT')
|
|
elif name == 'nft':
|
|
assert sys.argv[1:] == ['-j', 'list', 'tables']
|
|
print(json.dumps({'nftables': [{'table': {'family': 'inet' if profile == 'native' else 'ip', 'name': 'custom' if profile == 'native' else 'filter'}}]}))
|
|
else:
|
|
raise RuntimeError('Unexpected command')
|
|
"""
|
|
for name in ("systemctl", "dpkg-query", "iptables-save", "ip6tables-save", "nft"):
|
|
path = binary / name
|
|
path.write_text(dispatcher)
|
|
path.chmod(0o700)
|
|
env = {**os.environ, "PATH": directory, "PREFLIGHT_PROFILE": profile,
|
|
"PREFLIGHT_CALLS": str(binary / "calls")}
|
|
result = subprocess.run(["/bin/bash", str(ROOT / "preflight-oracle.sh")], env=env,
|
|
text=True, capture_output=True, timeout=15)
|
|
calls = (binary / "calls").read_text()
|
|
self.assertNotIn(" stop ", calls)
|
|
self.assertNotIn(" mask ", calls)
|
|
return result
|
|
|
|
def test_clean_compatibility_rules_pass(self):
|
|
result = self.run_profile("clean")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_active_manager_stops(self):
|
|
result = self.run_profile("active")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("ufw is active", result.stderr)
|
|
|
|
def test_installed_but_inactive_ufw_stops(self):
|
|
result = self.run_profile("installed")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("UFW is installed", result.stderr)
|
|
|
|
def test_leftover_chains_stop(self):
|
|
result = self.run_profile("leftover")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("chains found", result.stderr)
|
|
|
|
def test_native_table_stops(self):
|
|
result = self.run_profile("native")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("native nftables table", result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|