diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index f3181b2..1370fec 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -17,6 +17,8 @@ Preserve these compatibility contracts unless a change explicitly documents a mi 3. Run `pre-commit run --all-files`, `python -m pytest --cov`, `mypy set_dns.py`, and `python -m build`. 4. Open a pull request explaining configuration, DNS, and rollback impact. +`python3 set_dns.py --dry-run [--config PATH]` is available as a non-mutating preflight for local configuration checks. It never constructs a Cloudflare client or calls the Cloudflare API, so it does not validate Cloudflare credentials. When the selected configuration is SOPS-encrypted (`cflan_sops_vars.yaml` or `sops_vars.yaml`), it does invoke SOPS locally to decrypt the file, so it exercises SOPS and key availability for the invoking user without writing plaintext to disk. It does not install or execute the actual NetworkManager dispatcher hook, and it is not a substitute for CI. + CI is the validation authority. A passing unit-test suite does not prove that a changed dispatcher hook, SOPS configuration, or Cloudflare token works in an installed host. ## Pull requests diff --git a/README.md b/README.md index a1a53f3..82dee68 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,12 @@ The installer copies `set_dns.py` to `/etc/NetworkManager/dispatcher.d/set_dns` > A Cloudflare DNS record containing an RFC1918 address is useful only for clients that can route to that LAN. CFLAN does not make a private address reachable from the public Internet. +## Preflight dry run + +`set_dns.py --dry-run [--config PATH]` is a non-mutating preflight check. It validates root-volume configuration selection and parsing, the resolved local IPv4 address, the dispatcher positional arguments (`interface`/`action`) when present, and the derived FQDN, then prints the intended reconciliation and exits successfully without constructing a Cloudflare client or performing any Cloudflare API call. + +A dry run never constructs a Cloudflare client or calls the Cloudflare API, so it does not prove the API token is valid. When the selected configuration is SOPS-encrypted (`cflan_sops_vars.yaml` or `sops_vars.yaml`), the dry run does invoke SOPS locally to decrypt it, so it exercises SOPS and key availability for the invoking user without writing plaintext to disk. It does not install or execute the actual NetworkManager dispatcher hook. Normal behavior is unchanged when `--dry-run` is omitted. + ## Behavior and safety - Only NetworkManager `up` events update DNS; other dispatcher events are skipped. diff --git a/install.py b/install.py index 5820363..ce29a40 100644 --- a/install.py +++ b/install.py @@ -6,6 +6,7 @@ import os import shutil import sys +from collections.abc import Sequence from pathlib import Path DISPATCHER_PATH = Path("/etc/NetworkManager/dispatcher.d/set_dns") @@ -18,26 +19,30 @@ ) -def install() -> None: +def install( + source_dir: Path | None = None, + dispatcher_path: Path = DISPATCHER_PATH, + config_files: Sequence[tuple[str, str]] = CONFIG_FILES, +) -> None: """Install only files supplied by the operator; never create configuration values.""" if os.getuid() != 0: sys.exit("Error: Must run as root") - script_dir = Path(__file__).resolve().parent - if not DISPATCHER_PATH.parent.is_dir(): + source_root = Path(__file__).resolve().parent if source_dir is None else source_dir + if not dispatcher_path.parent.is_dir(): sys.exit( - f"Error: NetworkManager dispatcher directory is missing: {DISPATCHER_PATH.parent}" + f"Error: NetworkManager dispatcher directory is missing: {dispatcher_path.parent}" ) print("Deploying NetworkManager dispatcher script...") - shutil.copyfile(script_dir / "set_dns.py", DISPATCHER_PATH) - os.chown(DISPATCHER_PATH, 0, 0) - os.chmod(DISPATCHER_PATH, 0o700) - print(f" Installed: {DISPATCHER_PATH}") + shutil.copyfile(source_root / "set_dns.py", dispatcher_path) + os.chown(dispatcher_path, 0, 0) + os.chmod(dispatcher_path, 0o700) + print(f" Installed: {dispatcher_path}") print("\nDeploying configuration...") - for source_name, target_name in CONFIG_FILES: - source_path = script_dir / source_name + for source_name, target_name in config_files: + source_path = source_root / source_name if not source_path.is_file(): continue diff --git a/set_dns.py b/set_dns.py index f73a40d..5ceac24 100644 --- a/set_dns.py +++ b/set_dns.py @@ -3,6 +3,7 @@ from __future__ import annotations +import argparse import os import socket import subprocess @@ -300,10 +301,70 @@ def set_dns( update_dns_record(client, zone_id, record, record_name, local_ip_addr) +def dry_run( + argv: Sequence[str] | None = None, + config_path: str | None = None, +) -> None: + """Validate inputs and report the intended reconciliation without a client.""" + local_ip_addr = get_local_ip() + if not validate_network_manager_args(local_ip_addr, argv): + return + + config = parse_config(get_yaml_vars(config_path)) + record_name = get_record_name(config) + print( + "Dry run: would reconcile A record " + f"{record_name} to {local_ip_addr} " + f"(ttl={config.ttl}, proxied={config.proxied})." + ) + print("Dry run: no Cloudflare client was constructed and no API calls were made.") + + +def build_argument_parser() -> argparse.ArgumentParser: + """Build the CLI parser, preserving dispatcher positional arguments.""" + parser = argparse.ArgumentParser( + prog="set_dns", + description="Update one Cloudflare A record for this host's IPv4 address.", + ) + parser.add_argument( + "interface", + nargs="?", + help="NetworkManager dispatcher interface name (optional).", + ) + parser.add_argument( + "action", + nargs="?", + help="NetworkManager dispatcher action; only 'up' updates DNS (optional).", + ) + parser.add_argument( + "--config", + default=None, + help="Override the root-volume configuration path.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Validate inputs and report the intended reconciliation without " + "constructing a Cloudflare client or calling the API.", + ) + return parser + + def main(argv: Sequence[str] | None = None) -> int: """Run the dispatcher entry point without leaking configuration values.""" + raw_args = tuple(sys.argv if argv is None else argv) + namespace = build_argument_parser().parse_args(list(raw_args[1:])) + dispatcher_argv = [raw_args[0]] if raw_args else ["set_dns"] + if namespace.interface is not None: + dispatcher_argv.append(namespace.interface) + if namespace.action is not None: + dispatcher_argv.append(namespace.action) + try: - set_dns(argv=argv) + if namespace.dry_run: + dry_run(argv=dispatcher_argv, config_path=namespace.config) + else: + set_dns(argv=dispatcher_argv, config_path=namespace.config) except CflanError as error: print(f"cflan: {error}", file=sys.stderr) return 1 diff --git a/tests/test_install.py b/tests/test_install.py new file mode 100644 index 0000000..a157838 --- /dev/null +++ b/tests/test_install.py @@ -0,0 +1,136 @@ +from collections.abc import Iterator +from contextlib import contextmanager +from unittest.mock import MagicMock, patch + +import pytest + +import install + + +@contextmanager +def running_as_root() -> Iterator[None]: + """Patch getuid only around install() so tmp_path setup stays unpatched.""" + with patch("os.getuid", return_value=0): + yield + + +@contextmanager +def mocked_privileged_calls() -> Iterator[MagicMock]: + """Capture chown/chmod without touching the filesystem ownership.""" + calls = MagicMock() + with patch("os.chown", calls.chown), patch("os.chmod", calls.chmod): + yield calls + + +@pytest.fixture +def source_dir(tmp_path): + source = tmp_path / "source" + source.mkdir() + (source / "set_dns.py").write_text("#!/usr/bin/env python3\n", encoding="utf-8") + return source + + +@pytest.fixture +def dispatcher_path(tmp_path): + dispatcher_dir = tmp_path / "dispatcher.d" + dispatcher_dir.mkdir() + return dispatcher_dir / "set_dns" + + +class TestInstall: + def test_requires_root(self): + with patch("os.getuid", return_value=1000), pytest.raises(SystemExit): + install.install() + + def test_missing_dispatcher_directory_fails(self, source_dir, tmp_path): + missing_target = tmp_path / "missing" / "set_dns" + + with running_as_root(), pytest.raises(SystemExit): + install.install( + source_dir=source_dir, + dispatcher_path=missing_target, + config_files=(), + ) + + def test_preferred_config_wins_over_aliases( + self, source_dir, dispatcher_path, tmp_path + ): + (source_dir / "cflan_vars.yaml").write_text( + "cf_token: test-token\n", encoding="utf-8" + ) + (source_dir / "vars.yaml").write_text( + "cf_token: test-token\n", encoding="utf-8" + ) + preferred_target = tmp_path / "cflan_vars.yaml" + legacy_target = tmp_path / "vars.yaml" + + with running_as_root(), mocked_privileged_calls(): + install.install( + source_dir=source_dir, + dispatcher_path=dispatcher_path, + config_files=( + ("cflan_vars.yaml", str(preferred_target)), + ("vars.yaml", str(legacy_target)), + ), + ) + + assert dispatcher_path.is_file() + assert preferred_target.is_file() + assert not legacy_target.exists() + + def test_legacy_config_mapping_remains_valid( + self, source_dir, dispatcher_path, tmp_path + ): + (source_dir / "vars.yaml").write_text( + "cf_token: test-token\n", encoding="utf-8" + ) + legacy_target = tmp_path / "vars.yaml" + + with running_as_root(), mocked_privileged_calls(): + install.install( + source_dir=source_dir, + dispatcher_path=dispatcher_path, + config_files=( + ("cflan_vars.yaml", str(tmp_path / "cflan_vars.yaml")), + ("vars.yaml", str(legacy_target)), + ), + ) + + assert legacy_target.is_file() + + def test_installer_reports_targets_with_expected_ownership_and_modes( + self, source_dir, dispatcher_path, tmp_path, capsys + ): + (source_dir / "cflan_vars.yaml").write_text( + "cf_token: test-token\n", encoding="utf-8" + ) + config_target = tmp_path / "cflan_vars.yaml" + + with running_as_root(), mocked_privileged_calls() as privileged_calls: + install.install( + source_dir=source_dir, + dispatcher_path=dispatcher_path, + config_files=(("cflan_vars.yaml", str(config_target)),), + ) + + privileged_calls.chown.assert_any_call(dispatcher_path, 0, 0) + privileged_calls.chown.assert_any_call(config_target, 0, 0) + privileged_calls.chmod.assert_any_call(dispatcher_path, 0o700) + privileged_calls.chmod.assert_any_call(config_target, 0o600) + output = capsys.readouterr().out + assert f"Installed: {dispatcher_path}" in output + assert f"Installed: {config_target}" in output + + def test_missing_config_warns_without_installing_one( + self, source_dir, dispatcher_path, capsys + ): + with running_as_root(), mocked_privileged_calls() as privileged_calls: + install.install( + source_dir=source_dir, + dispatcher_path=dispatcher_path, + config_files=(("cflan_vars.yaml", "/should-not-be-used"),), + ) + + output = capsys.readouterr().out + assert "Warning: No configuration file found." in output + assert privileged_calls.chmod.call_count == 1 diff --git a/tests/test_set_dns.py b/tests/test_set_dns.py index 1974d0a..43069ec 100644 --- a/tests/test_set_dns.py +++ b/tests/test_set_dns.py @@ -217,6 +217,95 @@ def test_matching_record_is_not_mutated(self): client.dns.records.edit.assert_not_called() +class TestDryRun: + @pytest.fixture + def config_file(self, tmp_path): + path = tmp_path / "cflan_vars.yaml" + path.write_text( + "cf_token: test-token\ncf_domain_name: example.com\n", encoding="utf-8" + ) + return path + + @patch("set_dns.Cloudflare") + @patch("set_dns.socket.gethostbyname", return_value="192.168.1.100") + @patch("set_dns.socket.gethostname", return_value="host") + def test_dry_run_avoids_client_construction( + self, _, __, mock_cloudflare, config_file, capsys + ): + assert set_dns.main(["set_dns", "--dry-run", "--config", str(config_file)]) == 0 + + mock_cloudflare.assert_not_called() + assert "no Cloudflare client" in capsys.readouterr().out + + @patch("set_dns.set_dns") + @patch("set_dns.socket.gethostbyname", return_value="192.168.1.100") + @patch("set_dns.socket.gethostname", return_value="host") + def test_dry_run_never_enters_reconciliation( + self, _, __, mock_set_dns, config_file + ): + assert set_dns.main(["set_dns", "--dry-run", "--config", str(config_file)]) == 0 + + mock_set_dns.assert_not_called() + + @patch("set_dns.socket.gethostbyname", return_value="192.168.1.100") + @patch("set_dns.socket.gethostname", return_value="host") + def test_dry_run_reports_intended_action(self, _, __, config_file, capsys): + assert set_dns.main(["set_dns", "--dry-run", "--config", str(config_file)]) == 0 + + output = capsys.readouterr().out + assert "would reconcile A record" in output + assert "host.example.com" in output + assert "192.168.1.100" in output + assert "test-token" not in output + + @patch("set_dns.socket.gethostbyname", return_value="192.168.1.100") + @patch("set_dns.socket.gethostname", return_value="host") + def test_config_override_reaches_config_selection(self, _, __, config_file, capsys): + assert set_dns.main(["set_dns", "--dry-run", "--config", str(config_file)]) == 0 + + assert str(config_file) in capsys.readouterr().out + + @patch("set_dns.socket.gethostbyname", return_value="192.168.1.100") + @patch("set_dns.socket.gethostname", return_value="host") + def test_dry_run_rejects_invalid_config(self, _, __, tmp_path): + bad_config = tmp_path / "cflan_vars.yaml" + bad_config.write_text("cf_domain_name: example.com\n", encoding="utf-8") + + assert set_dns.main(["set_dns", "--dry-run", "--config", str(bad_config)]) == 1 + + @patch("set_dns.socket.gethostbyname", return_value="192.168.1.100") + @patch("set_dns.socket.gethostname", return_value="host") + def test_dry_run_preserves_dispatcher_arguments( + self, _, __, monkeypatch, config_file + ): + mock_netifaces = MagicMock() + mock_netifaces.AF_INET = 2 + mock_netifaces.ifaddresses.return_value = {2: [{"addr": "192.168.1.100"}]} + monkeypatch.setattr(set_dns, "netifaces", mock_netifaces) + + assert ( + set_dns.main( + ["set_dns", "eth0", "up", "--dry-run", "--config", str(config_file)] + ) + == 0 + ) + mock_netifaces.ifaddresses.assert_called_once_with("eth0") + + @patch("set_dns.socket.gethostbyname", return_value="192.168.1.100") + @patch("set_dns.socket.gethostname", return_value="host") + def test_dry_run_skips_non_up_dispatcher_action(self, _, __, config_file, capsys): + assert ( + set_dns.main( + ["set_dns", "eth0", "down", "--dry-run", "--config", str(config_file)] + ) + == 0 + ) + + output = capsys.readouterr().out + assert "Skipping NetworkManager action" in output + assert "would reconcile" not in output + + class TestEntrypoint: @patch("set_dns.set_dns", side_effect=set_dns.CflanError("bad configuration")) def test_main_returns_nonzero_for_expected_failure(self, _):