Initial commit

tests/test_setup.py

@@ -0,0 +1,171 @@
+"""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()