From 617d3b541519e322971a11276caf1f98c42f9605 Mon Sep 17 00:00:00 2001 From: Steven Welch Date: Tue, 1 Sep 2026 18:01:38 -0600 Subject: [PATCH 1/4] feat: harden DNS updates and configuration handling --- .github/workflows/ci.yml | 31 ++- CHANGELOG.md | 13 ++ CONTRIBUTING.md | 24 +++ README.md | 207 ++++--------------- SECURITY.md | 15 ++ install.py | 66 +++--- pyproject.toml | 79 ++++--- requirements-dev.txt | 22 +- requirements.txt | 8 +- set_dns.py | 432 ++++++++++++++++++++++++--------------- tests/test_set_dns.py | 329 ++++++++++++++--------------- 11 files changed, 630 insertions(+), 596 deletions(-) create mode 100644 CHANGELOG.md create mode 100644 CONTRIBUTING.md create mode 100644 SECURITY.md mode change 100755 => 100644 install.py mode change 100755 => 100644 set_dns.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 13b1ce1..9102c23 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,6 +6,9 @@ on: pull_request: branches: [main, master] +permissions: + contents: read + jobs: lint-and-test: runs-on: ubuntu-latest @@ -32,15 +35,13 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -r requirements.txt - pip install -r requirements-dev.txt + python -m pip install '.[dev]' - name: Run pre-commit hooks uses: pre-commit/action@2c7b3805fd2a0fd8c1884dcaebf91fc102a13ecd # v3.0.1 - name: Run tests with pytest - run: | - pytest --cov --cov-report=xml --cov-report=term-missing + run: python -m pytest --cov --cov-report=xml --cov-report=term-missing - name: Upload coverage to Codecov uses: codecov/codecov-action@fb8b3582c8e4def4969c97caa2f19720cb33a72f # v7.0.0 @@ -51,6 +52,25 @@ jobs: env: CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }} + package: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.11" + + - name: Build wheel and source distribution + run: | + python -m pip install --upgrade pip build + python -m build + + - name: Install the wheel and smoke-test the console command + run: | + python -m pip install dist/*.whl + cflan-set-dns --help + type-check: runs-on: ubuntu-latest steps: @@ -64,8 +84,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -r requirements.txt - pip install mypy types-PyYAML types-requests + python -m pip install '.[dev]' - name: Run mypy type checking run: mypy set_dns.py diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..f3faed1 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,13 @@ +# Changelog + +All notable changes to CFLAN are documented here. + +## Unreleased + +### Changed + +- Migrated the updater to the supported Cloudflare Python SDK interface. +- Replaced delete-and-create record changes with an in-place Cloudflare PATCH update. +- Added preferred root-volume configuration names: `cflan_vars.yaml` and `cflan_sops_vars.yaml`. +- Preserved `vars.yaml` and `sops_vars.yaml` as root-volume compatibility aliases. +- Added configuration validation, safer IPv4 checks, duplicate-record protection, package build verification, and public contributor/security guidance. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..f3181b2 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,24 @@ +# Contributing to CFLAN + +## Scope and safety + +CFLAN changes Cloudflare DNS from a NetworkManager dispatcher hook. Do not commit API tokens, decrypted SOPS files, real hostnames, DNS values, or root-volume contents. Do not test changes against production DNS as part of a pull request. + +Preserve these compatibility contracts unless a change explicitly documents a migration: + +- Dispatcher entry point: `/etc/NetworkManager/dispatcher.d/set_dns` +- Preferred root-volume configuration: `/cflan_vars.yaml` or `/cflan_sops_vars.yaml` +- Legacy root-volume aliases: `/vars.yaml` or `/sops_vars.yaml` + +## Development workflow + +1. Create a focused branch and add unit tests for behavior changes. +2. Install development dependencies with `python -m pip install '.[dev]'`. +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. + +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 + +Keep changes narrow. Document any changed default configuration name, API permission, record behavior, package version, or installed-path contract. Reviewers must be able to determine whether the change is source-only or requires a separate installation step. diff --git a/README.md b/README.md index 2b0cfce..a1a53f3 100644 --- a/README.md +++ b/README.md @@ -1,199 +1,82 @@ # CFLAN -[![CI](https://github.com/welchworks/cflan/actions/workflows/ci.yml/badge.svg)](https://github.com/welchworks/cflan/actions/workflows/ci.yml) -[![Python 3.9+](https://img.shields.io/badge/python-3.9+-blue.svg)](https://www.python.org/downloads/) +[![CI](https://github.com/makeitworkcloud/cflan/actions/workflows/ci.yml/badge.svg)](https://github.com/makeitworkcloud/cflan/actions/workflows/ci.yml) +[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/downloads/) [![License: GPL v3](https://img.shields.io/badge/License-GPLv3-blue.svg)](LICENSE) -[![Code style: Ruff](https://img.shields.io/badge/code%20style-ruff-000000.svg)](https://github.com/astral-sh/ruff) -> Integrating LAN infrastructure with Cloudflare services +> NetworkManager-driven Cloudflare DNS updates for LAN hosts. -CFLAN automatically updates Cloudflare DNS records with your local machine's IP address whenever your network connection changes. This is particularly useful for home servers, NAS devices, or any machine that needs a consistent DNS name despite having a dynamic local IP. +CFLAN reconciles one Cloudflare A record with a machine's active IPv4 address when a NetworkManager interface comes up. It is designed for root-owned configuration supplied from the host's root/volume area; it does not create, persist, or print credentials. -## Features +## Requirements -- **Automatic DNS Updates**: Updates Cloudflare DNS A records when network interfaces come up -- **NetworkManager Integration**: Runs as a NetworkManager dispatcher script -- **SOPS Support**: Supports encrypted configuration using [SOPS](https://github.com/getsops/sops) -- **IP Validation**: Ensures the correct interface IP is used before updating -- **Idempotent**: Only updates DNS when the IP address has actually changed +- Linux with NetworkManager dispatcher support +- Python 3.10 or later +- A Cloudflare API token limited to **Zone / DNS / Edit** for the target zone +- Optional: SOPS available and configured for root when using encrypted configuration -## Prerequisites +## Configuration contract -- Python 3.9 or higher -- NetworkManager (for dispatcher script functionality) -- Root access (for installing dispatcher scripts) -- Cloudflare API Token with DNS edit permissions -- (Optional) SOPS for encrypted configuration files +CFLAN searches these root-volume paths in order: -## Installation +1. `/cflan_vars.yaml` — preferred plaintext name +2. `/cflan_sops_vars.yaml` — preferred SOPS-encrypted name +3. `/vars.yaml` — supported legacy alias +4. `/sops_vars.yaml` — supported legacy SOPS alias -### From Source +`CFLAN_CONFIG` can override the path for a controlled deployment. The existing `/vars.yaml` and `/sops_vars.yaml` contract remains supported; no secret is copied into another directory by the updater. -```bash -# Clone the repository -git clone https://github.com/welchworks/cflan.git -cd cflan - -# Create configuration file (see Configuration section below) -# Then install (requires root) -sudo python install.py -``` - -### Development Installation - -```bash -pip install -r requirements.txt -pip install -r requirements-dev.txt -pre-commit install -``` - -## Configuration - -Create a configuration file with your Cloudflare credentials: - -### Option 1: Plain YAML (`vars.yaml`) +Example structure (do **not** commit a real token): ```yaml -cf_token: "your-cloudflare-api-token" -cf_domain_name: "example.com" -``` - -### Option 2: Encrypted with SOPS (`sops_vars.yaml`) - -```bash -# Create encrypted config -cat > vars.yaml < sops_vars.yaml -rm vars.yaml +# Optional full FQDN; defaults to . +# cf_record_name: "host.example.com" +# Optional; 1 means Cloudflare automatic TTL +# cf_ttl: 1 +# Optional; defaults to false to avoid proxying a LAN address +# cf_proxied: false ``` -### Cloudflare API Token Setup - -1. Go to [Cloudflare API Tokens](https://dash.cloudflare.com/profile/api-tokens) -2. Click "Create Token" -3. Use the "Edit zone DNS" template -4. Select your zone (domain) -5. Create the token and copy it for your config - -## Usage - -### As a NetworkManager Dispatcher Script +## Installation -When installed via `install.py`, the script runs automatically when network interfaces change: +The existing installation model is preserved: place one configuration file beside the scripts, then run the root-only installer. ```bash -# Trigger manually (for testing) -sudo /etc/NetworkManager/dispatcher.d/set_dns eth0 up +git clone https://github.com/makeitworkcloud/cflan.git +cd cflan +sudo python3 install.py ``` -### Standalone Execution +The installer copies `set_dns.py` to `/etc/NetworkManager/dispatcher.d/set_dns` with mode `0700`. It copies the first configuration filename it finds in the priority listed above to its matching root-volume path with mode `0600`. -```bash -# From the project directory -python set_dns.py -``` +> 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. -The script will: -1. Detect your local IP address -2. Read configuration from `vars.yaml` or decrypt `sops_vars.yaml` -3. Find the Cloudflare zone for your domain -4. Check for an existing DNS record -5. Create or update the A record with your current IP +## Behavior and safety -## Project Structure - -``` -cflan/ -├── .github/workflows/ # CI/CD configuration -├── tests/ # Test suite -├── set_dns.py # Main application script -├── install.py # Installation script -├── vars.yaml # Configuration (unencrypted) -├── pyproject.toml # Project metadata and tool config -├── requirements.txt # Production dependencies -├── requirements-dev.txt # Development dependencies -└── README.md # This file -``` +- Only NetworkManager `up` events update DNS; other dispatcher events are skipped. +- The interface IPv4 address must equal the resolved primary host IPv4 address. +- Loopback, multicast, unspecified, and malformed addresses are rejected before any API call. +- Exactly one matching zone and zero or one matching A record are required. Duplicate records fail closed. +- Existing records are updated with Cloudflare PATCH rather than delete-and-recreate, preserving the record and avoiding an avoidable DNS gap. +- SOPS plaintext exists only in the updater process memory. ## Development -### Running Tests - -```bash -pytest -``` - -With coverage: - ```bash -pytest --cov=cflan --cov-report=term-missing -``` - -### Code Quality - -This project uses: -- **Ruff**: Fast Python linter and formatter -- **MyPy**: Static type checking -- **Pre-commit**: Git hooks for code quality - -```bash -# Run linting +python -m pip install '.[dev]' +python -m pytest --cov ruff check . -ruff format . - -# Run type checking +ruff format --check . mypy set_dns.py - -# Run all pre-commit hooks -pre-commit run --all-files +python -m build ``` -### Continuous Integration - -GitHub Actions runs the following on every push and PR: -- Pre-commit hooks -- Tests across Python 3.9-3.13 -- Type checking with mypy -- Coverage reporting - -## Security Notes - -- The configuration file (`vars.yaml` or `sops_vars.yaml`) is installed with `600` permissions (readable only by root) -- The dispatcher script is installed with `700` permissions (executable only by root) -- Use SOPS encryption for production deployments to protect API tokens -- Store your Cloudflare API Token securely; it grants DNS edit access - -## Troubleshooting - -### "Must run as root" -The install script requires root privileges to install files to `/etc/NetworkManager/dispatcher.d/`. +CI runs formatting/linting hooks, unit tests and coverage on Python 3.10–3.13, mypy, and a wheel build/install smoke test. Unit tests do not contact Cloudflare or invoke NetworkManager. -### "sops must be installed" -If using `sops_vars.yaml`, ensure SOPS is installed: https://github.com/getsops/sops - -### "The IP address is a value for localhost" -The script prevents updating DNS with localhost addresses (127.0.0.x). Check your network configuration. +See [CONTRIBUTING.md](CONTRIBUTING.md) and [SECURITY.md](SECURITY.md) before opening an issue or pull request. ## License -This project is licensed under the GNU General Public License v3.0 - see the [LICENSE](LICENSE) file for details. - -## Contributing - -Contributions are welcome! Please: -1. Fork the repository -2. Create a feature branch -3. Make your changes -4. Run tests and linting -5. Submit a pull request - -## Author - -**Steven Welch** - steven@makeitwork.cloud - -Project Link: [https://github.com/welchworks/cflan](https://github.com/welchworks/cflan) +CFLAN is licensed under [GPL-3.0](LICENSE). diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..cc42523 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,15 @@ +# Security policy + +## Reporting a vulnerability + +Do **not** disclose Cloudflare tokens, decrypted SOPS data, host-specific paths, internal DNS values, or reproduction data containing credentials in a public issue. + +Use GitHub private vulnerability reporting for this repository when it is available. If it is unavailable, contact the maintainer through the public address in the project metadata and include only the minimum information needed to establish impact. + +## Supported versions + +Security fixes are made against the current `main` branch until versioned releases are published. No release branch or older version currently receives security support. + +## Deployment guidance + +Use a Cloudflare API token restricted to the specific zone and the minimum DNS-edit capability. Keep plaintext and encrypted configuration files root-owned with mode `0600`. CFLAN never needs a global API key. diff --git a/install.py b/install.py old mode 100755 new mode 100644 index 5bc2264..5c03d8d --- a/install.py +++ b/install.py @@ -1,52 +1,60 @@ #!/usr/bin/env python3 +"""Install the cflan NetworkManager dispatcher script and root-volume config.""" + +from __future__ import annotations + import os import shutil import sys +from pathlib import Path + +DISPATCHER_PATH = Path("/etc/NetworkManager/dispatcher.d/set_dns") +CONFIG_FILES: tuple[tuple[str, str], ...] = ( + ("cflan_vars.yaml", "/cflan_vars.yaml"), + ("cflan_sops_vars.yaml", "/cflan_sops_vars.yaml"), + # Legacy root-volume aliases remain supported for the existing installation. + ("vars.yaml", "/vars.yaml"), + ("sops_vars.yaml", "/sops_vars.yaml"), +) def install() -> 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 = os.path.dirname(os.path.abspath(__file__)) + script_dir = Path(__file__).resolve().parent + if not DISPATCHER_PATH.parent.is_dir(): + sys.exit(f"Error: NetworkManager dispatcher directory is missing: {DISPATCHER_PATH.parent}") print("Deploying NetworkManager dispatcher script...") - dispatcher_path = "/etc/NetworkManager/dispatcher.d/set_dns" - shutil.copyfile(os.path.join(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(script_dir / "set_dns.py", DISPATCHER_PATH) + os.chown(DISPATCHER_PATH, 0, 0) + os.chmod(DISPATCHER_PATH, 0o700) + print(f" Installed: {DISPATCHER_PATH}") print("\nDeploying configuration...") - config_deployed = False - - vars_path = os.path.join(script_dir, "vars.yaml") - if os.path.exists(vars_path): - target_path = "/vars.yaml" - shutil.copyfile(vars_path, target_path) - os.chown(target_path, 0, 0) - os.chmod(target_path, 0o600) - print(f" Installed: {target_path}") - config_deployed = True + for source_name, target_name in CONFIG_FILES: + source_path = script_dir / source_name + if not source_path.is_file(): + continue - sops_path = os.path.join(script_dir, "sops_vars.yaml") - if not config_deployed and os.path.exists(sops_path): - target_path = "/sops_vars.yaml" - shutil.copyfile(sops_path, target_path) + target_path = Path(target_name) + shutil.copyfile(source_path, target_path) os.chown(target_path, 0, 0) os.chmod(target_path, 0o600) print(f" Installed: {target_path}") - print(" Note: Ensure SOPS is configured for root user") - config_deployed = True - - if not config_deployed: - print(" Warning: No configuration file found (vars.yaml or sops_vars.yaml)") - print(" Create one before running the script!") + if source_name.startswith("cflan_"): + print(" Using the preferred cflan-prefixed root-volume configuration name.") + if source_name in {"cflan_sops_vars.yaml", "sops_vars.yaml"}: + print(" Note: Ensure SOPS is configured for the root user.") + break + else: + print(" Warning: No configuration file found.") + print(" Expected cflan_vars.yaml or cflan_sops_vars.yaml; legacy aliases remain valid.") print("\nInstallation complete!") - print("\nNext steps:") - print("1. Verify your configuration in /vars.yaml or /sops_vars.yaml") - print("2. Test with: sudo /etc/NetworkManager/dispatcher.d/set_dns eth0 up") + print("Configuration remains root-owned on the root/volume area.") if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index d5e2f54..a100a3d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,20 +4,25 @@ build-backend = "hatchling.build" [project] name = "cflan" -version = "1.0.0" -description = "Integrating LAN infrastructure with Cloudflare services" +version = "1.1.0" +description = "NetworkManager-driven Cloudflare DNS updates for LAN hosts" readme = "README.md" license = {file = "LICENSE"} authors = [ {name = "Steven Welch", email = "steven@makeitwork.cloud"} ] +requires-python = ">=3.10" +dependencies = [ + "cloudflare>=3,<6", + "netifaces>=0.11,<1", + "PyYAML>=6,<7", +] classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: System Administrators", "License :: OSI Approved :: GNU General Public License v3 (GPLv3)", "Operating System :: POSIX :: Linux", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", @@ -25,28 +30,30 @@ classifiers = [ "Topic :: Internet :: Name Service (DNS)", "Topic :: System :: Networking", ] -requires-python = ">=3.10" -dependencies = [ - "cloudflare", - "netifaces", - "PyYAML", -] +keywords = ["cloudflare", "dns", "networkmanager"] [project.optional-dependencies] dev = [ + "build", + "mypy", "pre-commit", "pytest", "pytest-cov", - "mypy", - "vulture", "types-PyYAML", - "types-requests", + "vulture", ] +[project.scripts] +cflan-set-dns = "set_dns:main" + [project.urls] -Homepage = "https://github.com/welchworks/cflan" -Repository = "https://github.com/welchworks/cflan" -Issues = "https://github.com/welchworks/cflan/issues" +Homepage = "https://github.com/makeitworkcloud/cflan" +Repository = "https://github.com/makeitworkcloud/cflan" +Issues = "https://github.com/makeitworkcloud/cflan/issues" +Security = "https://github.com/makeitworkcloud/cflan/security/policy" + +[tool.hatch.build.targets.wheel] +only-include = ["set_dns.py"] [tool.ruff] target-version = "py310" @@ -55,23 +62,17 @@ indent-width = 4 [tool.ruff.lint] select = [ - "E", # pycodestyle errors - "W", # pycodestyle warnings - "F", # Pyflakes - "I", # isort - "N", # pep8-naming - "W", # pycodestyle warnings - "UP", # pyupgrade - "B", # flake8-bugbear - "C4", # flake8-comprehensions - "SIM", # flake8-simplify + "E", + "W", + "F", + "I", + "N", + "UP", + "B", + "C4", + "SIM", ] -ignore = [ - "E501", # Line too long (handled by ruff-format) -] - -[tool.ruff.lint.pydocstyle] -convention = "google" +ignore = ["E501"] [tool.ruff.format] quote-style = "double" @@ -90,11 +91,7 @@ warn_unused_configs = true ignore_missing_imports = true strict = true show_error_codes = true -exclude = [ - "tests/", - "venv/", - ".venv/", -] +exclude = ["tests/", "venv/", ".venv/"] [tool.pytest.ini_options] minversion = "7.0" @@ -103,15 +100,9 @@ testpaths = ["tests"] python_files = ["test_*.py", "*_test.py"] [tool.coverage.run] -source = ["."] +source = ["set_dns"] branch = true -omit = [ - "tests/*", - "setup.py", - ".venv/*", - "venv/*", - "install.py", -] +omit = ["tests/*", "install.py", ".venv/*", "venv/*"] [tool.coverage.report] exclude_lines = [ diff --git a/requirements-dev.txt b/requirements-dev.txt index 5b85a99..fe88d09 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,15 +1,9 @@ -# Development dependencies -# Install with: pip install -r requirements-dev.txt - -# Testing -pytest>=9.1.1 -pytest-cov>=7.1.0 - -# Type checking -mypy>=2.3.1 +# Compatibility development requirements file. +# Packaging metadata in pyproject.toml is canonical; prefer: pip install '.[dev]'. +build +mypy +pre-commit +pytest +pytest-cov types-PyYAML -types-requests - -# Linting and formatting (via pre-commit) -pre-commit>=4.6.2 -vulture>=2.16 +vulture diff --git a/requirements.txt b/requirements.txt index 5410c68..fc20f4b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ -cloudflare -netifaces -PyYAML +# Compatibility requirements file for existing root-volume installations. +# Packaging metadata in pyproject.toml is the canonical dependency definition. +cloudflare>=3,<6 +netifaces>=0.11,<1 +PyYAML>=6,<7 diff --git a/set_dns.py b/set_dns.py old mode 100755 new mode 100644 index 5c6d396..c7eacfe --- a/set_dns.py +++ b/set_dns.py @@ -1,204 +1,304 @@ #!/usr/bin/env python3 -# steven@makeitwork.cloud -# https://github.com/welchworks/cflan/blob/main/set_dns.py -# -# To run as NetworkManager script, place in /etc/NetworkManager/disapatcher.d/ -# Accepts two optional positional arguments: 1) the NIC interface name, 2) the action, i.e. "up" -# -# Requires two YAML variables to be set in vars.yaml or alternatively as sops encrypted values in sops_vars.yaml: -# cf_token - Cloudflare API Token with DNS edit permissions -# cf_domain_name - Name of the DNS Zone in Cloudflare, i.e. mydomain.com +"""Update one Cloudflare A record when a NetworkManager interface comes up.""" +from __future__ import annotations + +import os import socket import subprocess import sys -from typing import Any +from dataclasses import dataclass +from ipaddress import IPv4Address, AddressValueError +from pathlib import Path +from typing import Any, Sequence -import CloudFlare import netifaces import yaml +from cloudflare import Cloudflare + +PREFERRED_PLAIN_CONFIG = Path("/cflan_vars.yaml") +PREFERRED_SOPS_CONFIG = Path("/cflan_sops_vars.yaml") +LEGACY_PLAIN_CONFIG = Path("/vars.yaml") +LEGACY_SOPS_CONFIG = Path("/sops_vars.yaml") +DEFAULT_CONFIG_PATHS: tuple[tuple[Path, bool], ...] = ( + (PREFERRED_PLAIN_CONFIG, False), + (PREFERRED_SOPS_CONFIG, True), + (LEGACY_PLAIN_CONFIG, False), + (LEGACY_SOPS_CONFIG, True), +) + + +class CflanError(RuntimeError): + """An expected configuration, network, or Cloudflare update failure.""" + + +@dataclass(frozen=True) +class CflanConfig: + """Validated configuration needed to update one A record.""" + + token: str + domain_name: str + record_name: str | None + ttl: int + proxied: bool def get_local_ip() -> str: - local_ip_addr = "" + """Resolve the host's primary IPv4 address without accepting loopback values.""" + hostname = socket.gethostname() + errors: list[OSError] = [] + + for candidate in (f"{hostname}.local", f"{hostname}.lan", hostname): + try: + return validate_ipv4(socket.gethostbyname(candidate)) + except OSError as error: + errors.append(error) + + raise CflanError("Could not resolve a usable IPv4 address for this host.") from errors[-1] + + +def validate_ipv4(value: str) -> str: + """Return a usable IPv4 address or raise without contacting Cloudflare.""" try: - local_ip_addr = socket.gethostbyname(socket.gethostname() + ".local") - except Exception: - local_ip_addr = socket.gethostbyname(socket.gethostname() + ".lan") - else: - local_ip_addr = socket.gethostbyname(socket.gethostname()) - return local_ip_addr + address = IPv4Address(value) + except AddressValueError as error: + raise CflanError("The detected address is not a valid IPv4 address.") from error + + if address.is_loopback or address.is_multicast or address.is_unspecified: + raise CflanError("The detected address is not suitable for a DNS A record.") + return str(address) -def validate_network_manager_args(local_ip_addr: str) -> None: - print("Parsing NetworkManager arguments...") +def validate_network_manager_args( + local_ip_addr: str, argv: Sequence[str] | None = None +) -> bool: + """Validate dispatcher arguments; return False for events that should be skipped.""" + args = tuple(sys.argv if argv is None else argv) + if len(args) < 3: + print("NetworkManager arguments were not set; proceeding as a standalone run.") + return True + + interface, action = args[1], args[2] + if action != "up": + print(f"Skipping NetworkManager action {action!r}; only 'up' updates DNS.") + return False + try: - if ( - netifaces.ifaddresses(sys.argv[1])[netifaces.AF_INET][0]["addr"] - != local_ip_addr - ): - print("Failed!") - sys.exit( - "The IP address " - + netifaces.ifaddresses(sys.argv[1])[netifaces.AF_INET][0]["addr"] - + " for the interface " - + sys.argv[1] - + " is not the same as the primary IP address of " - + local_ip_addr - + " ." - ) - if sys.argv[2] != "up": - print("Failed!") - sys.exit( - "The NetworkManager action '" - + sys.argv[2] - + "' does not match the required action of 'up'." - ) - except KeyError: - print("Failed!") - sys.exit("IP address for interface not set.") - except ValueError: - print("Failed!") - sys.exit("Invalid NetworkManager interface value for this script.") - except IndexError: - print("NetworkManager argument(s) were not set. Proceeding...") - - -def get_yaml_vars() -> dict[str, Any]: - print("Getting unencrypted values from vars.yaml ...") + interface_ip = netifaces.ifaddresses(interface)[netifaces.AF_INET][0]["addr"] + except (KeyError, ValueError, IndexError) as error: + raise CflanError("The NetworkManager interface has no IPv4 address.") from error + + if interface_ip != local_ip_addr: + raise CflanError( + "The NetworkManager interface IPv4 address does not match the primary IPv4 address." + ) + return True + + +def resolve_config_path( + override: str | None = None, + config_paths: Sequence[tuple[Path, bool]] = DEFAULT_CONFIG_PATHS, +) -> tuple[Path, bool]: + """Find the root-volume configuration, preferring the cflan-prefixed names.""" + selected = override or os.environ.get("CFLAN_CONFIG") + if selected: + path = Path(selected) + return path, path.name in { + PREFERRED_SOPS_CONFIG.name, + LEGACY_SOPS_CONFIG.name, + } + + for path, encrypted in config_paths: + if path.is_file(): + return path, encrypted + + searched = ", ".join(str(path) for path, _ in config_paths) + raise CflanError(f"No configuration file was found. Searched: {searched}") + + +def read_config_file(path: Path, encrypted: bool) -> dict[str, Any]: + """Read plaintext YAML or decrypt a SOPS YAML file without writing plaintext.""" try: - with open("vars.yaml") as f: - result: dict[str, Any] = yaml.safe_load(f.read()) - return result - except Exception: - print("Failed to get unencrypted values from vars.yaml ...") - print("Getting sops encrypted values from sops_vars.yaml ...") - try: - r = subprocess.run( - ["sops", "decrypt", "sops_vars.yaml"], + if encrypted: + result = subprocess.run( + ["sops", "decrypt", str(path)], + check=True, capture_output=True, + text=True, + timeout=30, ) - if r.returncode != 0: - print(r.stderr.decode("utf-8")) - sys.exit("Failed getting sops values.") - except FileNotFoundError: - print("Failed!") - sys.exit("sops must be installed and configured to use this script.") - - print("Getting YAML variables from sops output...") - sops_result: dict[str, Any] = yaml.safe_load(r.stdout.decode("utf-8")) - return sops_result - - -def get_zone_info(cf: CloudFlare.CloudFlare, domain_name: str) -> tuple[str, str]: - print("Getting CloudFlare DNS Zone ID and Name...") - zone_data = cf.zones.get(params={"per_page": "1", "name": domain_name})[0] - return zone_data["id"], zone_data["name"] - - -def get_dns_record_id( - cf: CloudFlare.CloudFlare, zone_id: str, hostname: str, zone_name: str -) -> str: - print( - "Attempting to get existing DNS record for " - + hostname - + "." - + zone_name - + " ..." + loaded = yaml.safe_load(result.stdout) + else: + loaded = yaml.safe_load(path.read_text(encoding="utf-8")) + except FileNotFoundError as error: + raise CflanError(f"Configuration file or SOPS executable was not found: {path}") from error + except subprocess.CalledProcessError as error: + raise CflanError("SOPS could not decrypt the configuration file.") from error + except subprocess.TimeoutExpired as error: + raise CflanError("SOPS decryption timed out.") from error + except (OSError, yaml.YAMLError) as error: + raise CflanError("Configuration could not be read as YAML.") from error + + if not isinstance(loaded, dict): + raise CflanError("Configuration must be a YAML mapping.") + return loaded + + +def get_yaml_vars(config_path: str | None = None) -> dict[str, Any]: + """Compatibility helper that returns validated-path YAML values.""" + path, encrypted = resolve_config_path(config_path) + print(f"Loading configuration from {path}.") + return read_config_file(path, encrypted) + + +def parse_config(values: dict[str, Any]) -> CflanConfig: + """Validate user-controlled values before creating a Cloudflare client.""" + token = values.get("cf_token") + domain_name = values.get("cf_domain_name") + record_name = values.get("cf_record_name") + ttl = values.get("cf_ttl", 1) + proxied = values.get("cf_proxied", False) + + if not isinstance(token, str) or not token.strip(): + raise CflanError("Configuration requires a non-empty cf_token.") + if not isinstance(domain_name, str) or not domain_name.strip(): + raise CflanError("Configuration requires a non-empty cf_domain_name.") + if record_name is not None and (not isinstance(record_name, str) or not record_name.strip()): + raise CflanError("cf_record_name must be a non-empty string when provided.") + if isinstance(ttl, bool) or not isinstance(ttl, int) or (ttl != 1 and not 60 <= ttl <= 86400): + raise CflanError("cf_ttl must be 1 or an integer from 60 through 86400.") + if not isinstance(proxied, bool): + raise CflanError("cf_proxied must be a boolean.") + + return CflanConfig( + token=token, + domain_name=domain_name.rstrip("."), + record_name=record_name.rstrip(".") if record_name else None, + ttl=ttl, + proxied=proxied, ) - try: - records = cf.zones.dns_records.get( - zone_id, - params={ - "name": hostname + "." + zone_name, - "match": "all", - "type": "A", - }, + + +def get_record_name(config: CflanConfig) -> str: + """Return a fully qualified record name, preserving the legacy hostname default.""" + hostname = config.record_name or socket.gethostname() + if hostname.endswith(f".{config.domain_name}"): + return hostname + return f"{hostname}.{config.domain_name}" + + +def get_zone_info(client: Any, domain_name: str) -> tuple[str, str]: + """Find exactly the configured Cloudflare zone.""" + zones = list(client.zones.list(name=domain_name, per_page=1)) + matches = [zone for zone in zones if getattr(zone, "name", None) == domain_name] + if len(matches) != 1: + raise CflanError("Cloudflare did not return exactly one matching zone.") + + zone = matches[0] + zone_id = getattr(zone, "id", None) + zone_name = getattr(zone, "name", None) + if not isinstance(zone_id, str) or not isinstance(zone_name, str): + raise CflanError("Cloudflare returned an invalid zone response.") + return zone_id, zone_name + + +def get_dns_record(client: Any, zone_id: str, record_name: str) -> Any | None: + """Find one A record, refusing to mutate an ambiguous record set.""" + records = list( + client.dns.records.list( + zone_id=zone_id, + name=record_name, + type="A", + per_page=100, ) - if records and len(records) > 0: - record_id: str = records[0]["id"] - return record_id - except Exception: - pass - return "" + ) + if len(records) > 1: + raise CflanError("More than one matching A record exists; refusing to choose one.") + return records[0] if records else None def create_dns_record( - cf: CloudFlare.CloudFlare, zone_id: str, hostname: str, ip_addr: str + client: Any, + zone_id: str, + record_name: str, + ip_addr: str, + config: CflanConfig, ) -> None: - print("Record not found...") - print("Creating new record for " + hostname + "...") - try: - cf.zones.dns_records.post( - zone_id, - data={"name": hostname, "type": "A", "content": ip_addr}, - ) - except CloudFlare.exceptions.CloudFlareAPIError as e: - sys.exit(f"/zones.dns_records.post {e} - {e} {e}") - print("Success!") + """Create the configured A record with explicit safe defaults.""" + client.dns.records.create( + zone_id=zone_id, + name=record_name, + type="A", + content=ip_addr, + ttl=config.ttl, + proxied=config.proxied, + ) + print(f"Created DNS record {record_name}.") def update_dns_record( - cf: CloudFlare.CloudFlare, zone_id: str, dns_id: str, hostname: str, ip_addr: str + client: Any, + zone_id: str, + record: Any, + record_name: str, + ip_addr: str, ) -> None: - print("Getting IP address for existing record...") - dns_content = cf.zones.dns_records.get( - zone_id, - params={ - "name": hostname, - "match": "all", - "type": "A", - }, - )[0]["content"] - - print("Evaluating if existing record matches current IP address...") - if dns_content == ip_addr: - print("Record matches, exiting...") - sys.exit() - - print("Deleting existing record...") - cf.zones.dns_records.delete(zone_id, dns_id) - - print("Creating new record for " + hostname + "...") - try: - cf.zones.dns_records.post( - zone_id, - data={"name": hostname, "type": "A", "content": ip_addr}, - ) - except CloudFlare.exceptions.CloudFlareAPIError as e: - sys.exit(f"/zones.dns_records.post {e} - {e} {e}") - - print("Success!") + """Patch only the A-record content, retaining the record's existing settings.""" + if getattr(record, "content", None) == ip_addr: + print(f"DNS record {record_name} already matches the local IPv4 address.") + return + + record_id = getattr(record, "id", None) + ttl = getattr(record, "ttl", None) + if not isinstance(record_id, str) or not isinstance(ttl, int): + raise CflanError("Cloudflare returned an invalid DNS record response.") + + client.dns.records.edit( + record_id, + zone_id=zone_id, + name=record_name, + type="A", + content=ip_addr, + ttl=ttl, + proxied=bool(getattr(record, "proxied", False)), + ) + print(f"Updated DNS record {record_name}.") -def set_dns() -> None: +def set_dns( + argv: Sequence[str] | None = None, + config_path: str | None = None, + client_factory: Any = Cloudflare, +) -> None: + """Reconcile the configured Cloudflare A record with this host's IPv4 address.""" local_ip_addr = get_local_ip() + if not validate_network_manager_args(local_ip_addr, argv): + return - print("Using IP address " + local_ip_addr + " ...") - if "127.0.0" in local_ip_addr: - print("Failed!") - sys.exit("The IP address is a value for localhost.") - - validate_network_manager_args(local_ip_addr) - - yaml_vars = get_yaml_vars() + config = parse_config(get_yaml_vars(config_path)) + record_name = get_record_name(config) + client = client_factory(api_token=config.token, timeout=10.0) + zone_id, _ = get_zone_info(client, config.domain_name) + record = get_dns_record(client, zone_id, record_name) - print("Initiating CloudFlare object using API Token...") - cf = CloudFlare.CloudFlare(token=yaml_vars["cf_token"]) - - zone_id, zone_name = get_zone_info(cf, yaml_vars["cf_domain_name"]) - hostname = socket.gethostname() + if record is None: + create_dns_record(client, zone_id, record_name, local_ip_addr, config) + else: + update_dns_record(client, zone_id, record, record_name, local_ip_addr) - dns_id = get_dns_record_id(cf, zone_id, hostname, zone_name) - if not dns_id: - create_dns_record(cf, zone_id, hostname, local_ip_addr) - sys.exit() - else: - update_dns_record( - cf, zone_id, dns_id, hostname + "." + zone_name, local_ip_addr - ) +def main(argv: Sequence[str] | None = None) -> int: + """Run the dispatcher entry point without leaking configuration values.""" + try: + set_dns(argv=argv) + except CflanError as error: + print(f"cflan: {error}", file=sys.stderr) + return 1 + except Exception as error: # Cloudflare SDK exceptions vary by transport and status. + print(f"cflan: Cloudflare update failed ({type(error).__name__}).", file=sys.stderr) + return 1 + return 0 if __name__ == "__main__": - set_dns() + sys.exit(main()) diff --git a/tests/test_set_dns.py b/tests/test_set_dns.py index b4efb66..b91ef7b 100644 --- a/tests/test_set_dns.py +++ b/tests/test_set_dns.py @@ -1,229 +1,214 @@ -import socket -import sys +from pathlib import Path +from types import SimpleNamespace from unittest.mock import MagicMock, mock_open, patch import pytest -# Mock netifaces before import -sys.modules["netifaces"] = MagicMock() -sys.modules["CloudFlare"] = MagicMock() -sys.modules["yaml"] = MagicMock() +import set_dns -from set_dns import ( # noqa: E402 - create_dns_record, - get_dns_record_id, - get_local_ip, - get_yaml_vars, - get_zone_info, - update_dns_record, - validate_network_manager_args, -) +@pytest.fixture +def config_values(): + return { + "cf_token": "test-token", + "cf_domain_name": "example.com", + } -class TestGetLocalIp: - @patch("set_dns.socket.gethostbyname") - @patch("set_dns.socket.gethostname") - def test_get_local_ip_with_local_suffix(self, mock_hostname, mock_gethost): - mock_hostname.return_value = "testhost" - mock_gethost.return_value = "192.168.1.100" - result = get_local_ip() - assert result == "192.168.1.100" - mock_gethost.assert_any_call("testhost.local") +class TestAddresses: @patch("set_dns.socket.gethostbyname") - @patch("set_dns.socket.gethostname") - def test_get_local_ip_fallback_to_lan(self, mock_hostname, mock_gethost): - mock_hostname.return_value = "testhost" - - def side_effect(hostname): - if hostname.endswith(".local"): - raise socket.gaierror("Not found") - return "192.168.1.100" + @patch("set_dns.socket.gethostname", return_value="testhost") + def test_get_local_ip_prefers_local_name(self, _, mock_gethost): + mock_gethost.return_value = "192.168.1.100" - mock_gethost.side_effect = side_effect - result = get_local_ip() - assert result == "192.168.1.100" + assert set_dns.get_local_ip() == "192.168.1.100" + mock_gethost.assert_called_once_with("testhost.local") @patch("set_dns.socket.gethostbyname") - @patch("set_dns.socket.gethostname") - def test_get_local_ip_both_suffixes_fail(self, mock_hostname, mock_gethost): - mock_hostname.return_value = "testhost" - - def side_effect(hostname): - raise socket.gaierror("Not found") + @patch("set_dns.socket.gethostname", return_value="testhost") + def test_get_local_ip_falls_back_to_lan_name(self, _, mock_gethost): + mock_gethost.side_effect = [OSError("missing"), "192.168.1.100"] - mock_gethost.side_effect = side_effect + assert set_dns.get_local_ip() == "192.168.1.100" + assert mock_gethost.call_args_list[1].args == ("testhost.lan",) - with pytest.raises(socket.gaierror): - get_local_ip() + @pytest.mark.parametrize("value", ["127.0.0.1", "0.0.0.0", "224.0.0.1", "not-an-ip"]) + def test_validate_ipv4_rejects_unsuitable_addresses(self, value): + with pytest.raises(set_dns.CflanError): + set_dns.validate_ipv4(value) -class TestValidateNetworkManagerArgs: - @patch("set_dns.netifaces") - @patch("set_dns.sys") - def test_valid_interface_and_action(self, mock_sys, mock_netifaces): - mock_sys.argv = ["script", "eth0", "up"] - mock_netifaces.ifaddresses.return_value = {2: [{"addr": "192.168.1.100"}]} +class TestNetworkManagerArguments: + def test_valid_interface_and_action(self, monkeypatch): + 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) - validate_network_manager_args("192.168.1.100") - - @patch("set_dns.netifaces") - @patch("set_dns.sys") - def test_mismatched_ip_address(self, mock_sys, mock_netifaces): - mock_sys.argv = ["script", "eth0", "up"] - mock_sys.exit.side_effect = SystemExit - mock_netifaces.ifaddresses.return_value = {2: [{"addr": "10.0.0.1"}]} - mock_netifaces.AF_INET = 2 + assert set_dns.validate_network_manager_args( + "192.168.1.100", ["set_dns", "eth0", "up"] + ) - with pytest.raises(SystemExit): - validate_network_manager_args("192.168.1.100") + def test_non_up_action_is_skipped(self): + assert not set_dns.validate_network_manager_args( + "192.168.1.100", ["set_dns", "eth0", "down"] + ) - @patch("set_dns.netifaces") - @patch("set_dns.sys") - def test_wrong_action(self, mock_sys, mock_netifaces): - mock_sys.argv = ["script", "eth0", "down"] - mock_sys.exit.side_effect = SystemExit - mock_netifaces.ifaddresses.return_value = {2: [{"addr": "192.168.1.100"}]} + def test_mismatched_interface_address_is_rejected(self, monkeypatch): + mock_netifaces = MagicMock() mock_netifaces.AF_INET = 2 + mock_netifaces.ifaddresses.return_value = {2: [{"addr": "10.0.0.1"}]} + monkeypatch.setattr(set_dns, "netifaces", mock_netifaces) - with pytest.raises(SystemExit): - validate_network_manager_args("192.168.1.100") + with pytest.raises(set_dns.CflanError): + set_dns.validate_network_manager_args( + "192.168.1.100", ["set_dns", "eth0", "up"] + ) - @patch("set_dns.netifaces") - @patch("set_dns.sys") - def test_no_arguments(self, mock_sys, mock_netifaces): - mock_sys.argv = ["script"] - mock_netifaces.AF_INET = 2 - validate_network_manager_args("192.168.1.100") +class TestConfiguration: + def test_prefers_cflan_prefixed_root_volume_filename(self, tmp_path): + preferred = tmp_path / "cflan_vars.yaml" + legacy = tmp_path / "vars.yaml" + preferred.write_text("cf_token: test-token\ncf_domain_name: example.com\n") + legacy.write_text("cf_token: test-token\ncf_domain_name: example.com\n") + path, encrypted = set_dns.resolve_config_path( + config_paths=((preferred, False), (legacy, False)) + ) -class TestGetYamlVars: - @patch( - "builtins.open", - mock_open(read_data="cf_token: test123\ncf_domain: example.com"), - ) - @patch("set_dns.yaml") - def test_get_unencrypted_yaml(self, mock_yaml): - mock_yaml.safe_load.return_value = { - "cf_token": "test123", - "cf_domain": "example.com", - } + assert path == preferred + assert not encrypted - result = get_yaml_vars() + def test_legacy_root_volume_filename_remains_an_alias(self, tmp_path): + legacy = tmp_path / "vars.yaml" + legacy.write_text("cf_token: test-token\ncf_domain_name: example.com\n") - assert result["cf_token"] == "test123" - assert result["cf_domain"] == "example.com" + path, encrypted = set_dns.resolve_config_path(config_paths=((legacy, False),)) - @patch("set_dns.subprocess.run") - @patch("set_dns.yaml") - @patch("builtins.open") - def test_get_sops_encrypted_yaml(self, mock_open_file, mock_yaml, mock_run): - mock_open_file.side_effect = FileNotFoundError() - - mock_run.return_value = MagicMock( - returncode=0, - stdout=b"cf_token: encrypted123\ncf_domain: example.com", - stderr=b"", - ) - mock_yaml.safe_load.return_value = { - "cf_token": "encrypted123", - "cf_domain": "example.com", - } + assert path == legacy + assert not encrypted - result = get_yaml_vars() + @patch("builtins.open", mock_open(read_data="cf_token: test-token\ncf_domain_name: example.com")) + @patch("set_dns.Path.read_text", return_value="cf_token: test-token\ncf_domain_name: example.com") + def test_reads_plaintext_yaml(self, mock_read, _): + values = set_dns.read_config_file(Path("/cflan_vars.yaml"), encrypted=False) - assert result["cf_token"] == "encrypted123" - mock_run.assert_called_once() + assert values["cf_token"] == "test-token" + mock_read.assert_called_once_with(encoding="utf-8") @patch("set_dns.subprocess.run") - @patch("builtins.open") - def test_sops_not_installed(self, mock_open_file, mock_run): - mock_open_file.side_effect = FileNotFoundError() - mock_run.side_effect = FileNotFoundError("sops not found") - - with pytest.raises(SystemExit): - get_yaml_vars() - + def test_reads_sops_yaml_without_writing_plaintext(self, mock_run): + mock_run.return_value.stdout = "cf_token: test-token\ncf_domain_name: example.com" -class TestGetZoneInfo: - def test_get_zone_info(self): - mock_cf = MagicMock() - mock_cf.zones.get.return_value = [{"id": "zone123", "name": "example.com"}] + values = set_dns.read_config_file(Path("/cflan_sops_vars.yaml"), encrypted=True) - zone_id, zone_name = get_zone_info(mock_cf, "example.com") + assert values["cf_domain_name"] == "example.com" + mock_run.assert_called_once() - assert zone_id == "zone123" - assert zone_name == "example.com" + def test_config_requires_token_and_domain(self): + with pytest.raises(set_dns.CflanError): + set_dns.parse_config({"cf_token": "test-token"}) + def test_config_rejects_invalid_ttl(self, config_values): + config_values["cf_ttl"] = 30 -class TestGetDnsRecordId: - def test_existing_record_found(self): - mock_cf = MagicMock() - mock_cf.zones.dns_records.get.return_value = [{"id": "record123"}] + with pytest.raises(set_dns.CflanError): + set_dns.parse_config(config_values) - result = get_dns_record_id(mock_cf, "zone123", "host", "example.com") + @patch("set_dns.socket.gethostname", return_value="host") + def test_record_name_defaults_to_hostname(self, _, config_values): + config = set_dns.parse_config(config_values) - assert result == "record123" + assert set_dns.get_record_name(config) == "host.example.com" - def test_no_existing_record(self): - mock_cf = MagicMock() - mock_cf.zones.dns_records.get.return_value = [] - result = get_dns_record_id(mock_cf, "zone123", "host", "example.com") +class TestCloudflareReconciliation: + def test_get_zone_info_requires_exact_match(self): + client = MagicMock() + client.zones.list.return_value = [ + SimpleNamespace(id="zone-id", name="example.com") + ] - assert result == "" + assert set_dns.get_zone_info(client, "example.com") == ("zone-id", "example.com") + def test_get_dns_record_rejects_duplicates(self): + client = MagicMock() + client.dns.records.list.return_value = [MagicMock(), MagicMock()] -class TestCreateDnsRecord: - def test_create_record_success(self): - mock_cf = MagicMock() + with pytest.raises(set_dns.CflanError): + set_dns.get_dns_record(client, "zone-id", "host.example.com") - create_dns_record(mock_cf, "zone123", "host", "192.168.1.100") + def test_create_record_uses_explicit_defaults(self, config_values): + client = MagicMock() + config = set_dns.parse_config(config_values) - mock_cf.zones.dns_records.post.assert_called_once_with( - "zone123", - data={"name": "host", "type": "A", "content": "192.168.1.100"}, + set_dns.create_dns_record( + client, "zone-id", "host.example.com", "192.168.1.100", config ) - def test_create_record_api_error(self): - mock_cf = MagicMock() - - class MockAPIError(Exception): - def __str__(self): - return "API Error" - - def __int__(self): - return 400 - - mock_cf.zones.dns_records.post.side_effect = MockAPIError("API Error") - # Also set the exception on the module mock so isinstance check passes - import sys - - sys.modules["CloudFlare"].exceptions.CloudFlareAPIError = MockAPIError + client.dns.records.create.assert_called_once_with( + zone_id="zone-id", + name="host.example.com", + type="A", + content="192.168.1.100", + ttl=1, + proxied=False, + ) - with pytest.raises(SystemExit): - create_dns_record(mock_cf, "zone123", "host", "192.168.1.100") + def test_update_record_uses_patch_without_delete(self): + client = MagicMock() + record = SimpleNamespace( + id="record-id", + content="10.0.0.1", + ttl=300, + proxied=True, + ) + set_dns.update_dns_record( + client, + "zone-id", + record, + "host.example.com", + "192.168.1.100", + ) -class TestUpdateDnsRecord: - def test_update_when_ip_changed(self): - mock_cf = MagicMock() - mock_cf.zones.dns_records.get.return_value = [{"content": "10.0.0.1"}] + client.dns.records.edit.assert_called_once_with( + "record-id", + zone_id="zone-id", + name="host.example.com", + type="A", + content="192.168.1.100", + ttl=300, + proxied=True, + ) + client.dns.records.delete.assert_not_called() + + def test_matching_record_is_not_mutated(self): + client = MagicMock() + record = SimpleNamespace( + id="record-id", + content="192.168.1.100", + ttl=1, + proxied=False, + ) - update_dns_record( - mock_cf, "zone123", "record123", "host.example.com", "192.168.1.100" + set_dns.update_dns_record( + client, + "zone-id", + record, + "host.example.com", + "192.168.1.100", ) - mock_cf.zones.dns_records.delete.assert_called_once_with("zone123", "record123") - mock_cf.zones.dns_records.post.assert_called_once() + client.dns.records.edit.assert_not_called() - def test_no_update_when_ip_unchanged(self): - mock_cf = MagicMock() - mock_cf.zones.dns_records.get.return_value = [{"content": "192.168.1.100"}] - with pytest.raises(SystemExit): - update_dns_record( - mock_cf, "zone123", "record123", "host.example.com", "192.168.1.100" - ) +class TestEntrypoint: + @patch("set_dns.set_dns", side_effect=set_dns.CflanError("bad configuration")) + def test_main_returns_nonzero_for_expected_failure(self, _): + assert set_dns.main(["set_dns"]) == 1 + + @patch("set_dns.set_dns", side_effect=RuntimeError("unexpected")) + def test_main_does_not_render_unexpected_error_text(self, _): + assert set_dns.main(["set_dns"]) == 1 From 264b8fbaa6afef017bebe479db0778119a3e85c9 Mon Sep 17 00:00:00 2001 From: Steven Welch Date: Tue, 1 Sep 2026 18:03:44 -0600 Subject: [PATCH 2/4] test: exercise packaged console entry point --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9102c23..5dd002d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -69,7 +69,7 @@ jobs: - name: Install the wheel and smoke-test the console command run: | python -m pip install dist/*.whl - cflan-set-dns --help + ! cflan-set-dns type-check: runs-on: ubuntu-latest From 3b150c28f340b730bee6b3e45ac918bb2d0630fb Mon Sep 17 00:00:00 2001 From: Steven Welch Date: Tue, 1 Sep 2026 18:08:04 -0600 Subject: [PATCH 3/4] style: apply ruff and ruff-format fixes from pre-commit CI --- install.py | 12 +++++++++--- set_dns.py | 36 +++++++++++++++++++++++++++--------- tests/test_set_dns.py | 23 ++++++++++++++++++----- 3 files changed, 54 insertions(+), 17 deletions(-) diff --git a/install.py b/install.py index 5c03d8d..5820363 100644 --- a/install.py +++ b/install.py @@ -25,7 +25,9 @@ def install() -> None: script_dir = Path(__file__).resolve().parent if not DISPATCHER_PATH.parent.is_dir(): - sys.exit(f"Error: NetworkManager dispatcher directory is missing: {DISPATCHER_PATH.parent}") + sys.exit( + f"Error: NetworkManager dispatcher directory is missing: {DISPATCHER_PATH.parent}" + ) print("Deploying NetworkManager dispatcher script...") shutil.copyfile(script_dir / "set_dns.py", DISPATCHER_PATH) @@ -45,13 +47,17 @@ def install() -> None: os.chmod(target_path, 0o600) print(f" Installed: {target_path}") if source_name.startswith("cflan_"): - print(" Using the preferred cflan-prefixed root-volume configuration name.") + print( + " Using the preferred cflan-prefixed root-volume configuration name." + ) if source_name in {"cflan_sops_vars.yaml", "sops_vars.yaml"}: print(" Note: Ensure SOPS is configured for the root user.") break else: print(" Warning: No configuration file found.") - print(" Expected cflan_vars.yaml or cflan_sops_vars.yaml; legacy aliases remain valid.") + print( + " Expected cflan_vars.yaml or cflan_sops_vars.yaml; legacy aliases remain valid." + ) print("\nInstallation complete!") print("Configuration remains root-owned on the root/volume area.") diff --git a/set_dns.py b/set_dns.py index c7eacfe..f73a40d 100644 --- a/set_dns.py +++ b/set_dns.py @@ -7,10 +7,11 @@ import socket import subprocess import sys +from collections.abc import Sequence from dataclasses import dataclass -from ipaddress import IPv4Address, AddressValueError +from ipaddress import AddressValueError, IPv4Address from pathlib import Path -from typing import Any, Sequence +from typing import Any import netifaces import yaml @@ -54,7 +55,9 @@ def get_local_ip() -> str: except OSError as error: errors.append(error) - raise CflanError("Could not resolve a usable IPv4 address for this host.") from errors[-1] + raise CflanError( + "Could not resolve a usable IPv4 address for this host." + ) from errors[-1] def validate_ipv4(value: str) -> str: @@ -131,7 +134,9 @@ def read_config_file(path: Path, encrypted: bool) -> dict[str, Any]: else: loaded = yaml.safe_load(path.read_text(encoding="utf-8")) except FileNotFoundError as error: - raise CflanError(f"Configuration file or SOPS executable was not found: {path}") from error + raise CflanError( + f"Configuration file or SOPS executable was not found: {path}" + ) from error except subprocess.CalledProcessError as error: raise CflanError("SOPS could not decrypt the configuration file.") from error except subprocess.TimeoutExpired as error: @@ -163,9 +168,15 @@ def parse_config(values: dict[str, Any]) -> CflanConfig: raise CflanError("Configuration requires a non-empty cf_token.") if not isinstance(domain_name, str) or not domain_name.strip(): raise CflanError("Configuration requires a non-empty cf_domain_name.") - if record_name is not None and (not isinstance(record_name, str) or not record_name.strip()): + if record_name is not None and ( + not isinstance(record_name, str) or not record_name.strip() + ): raise CflanError("cf_record_name must be a non-empty string when provided.") - if isinstance(ttl, bool) or not isinstance(ttl, int) or (ttl != 1 and not 60 <= ttl <= 86400): + if ( + isinstance(ttl, bool) + or not isinstance(ttl, int) + or (ttl != 1 and not 60 <= ttl <= 86400) + ): raise CflanError("cf_ttl must be 1 or an integer from 60 through 86400.") if not isinstance(proxied, bool): raise CflanError("cf_proxied must be a boolean.") @@ -213,7 +224,9 @@ def get_dns_record(client: Any, zone_id: str, record_name: str) -> Any | None: ) ) if len(records) > 1: - raise CflanError("More than one matching A record exists; refusing to choose one.") + raise CflanError( + "More than one matching A record exists; refusing to choose one." + ) return records[0] if records else None @@ -294,8 +307,13 @@ def main(argv: Sequence[str] | None = None) -> int: except CflanError as error: print(f"cflan: {error}", file=sys.stderr) return 1 - except Exception as error: # Cloudflare SDK exceptions vary by transport and status. - print(f"cflan: Cloudflare update failed ({type(error).__name__}).", file=sys.stderr) + except ( + Exception + ) as error: # Cloudflare SDK exceptions vary by transport and status. + print( + f"cflan: Cloudflare update failed ({type(error).__name__}).", + file=sys.stderr, + ) return 1 return 0 diff --git a/tests/test_set_dns.py b/tests/test_set_dns.py index b91ef7b..68c5194 100644 --- a/tests/test_set_dns.py +++ b/tests/test_set_dns.py @@ -32,7 +32,9 @@ def test_get_local_ip_falls_back_to_lan_name(self, _, mock_gethost): assert set_dns.get_local_ip() == "192.168.1.100" assert mock_gethost.call_args_list[1].args == ("testhost.lan",) - @pytest.mark.parametrize("value", ["127.0.0.1", "0.0.0.0", "224.0.0.1", "not-an-ip"]) + @pytest.mark.parametrize( + "value", ["127.0.0.1", "0.0.0.0", "224.0.0.1", "not-an-ip"] + ) def test_validate_ipv4_rejects_unsuitable_addresses(self, value): with pytest.raises(set_dns.CflanError): set_dns.validate_ipv4(value) @@ -89,8 +91,14 @@ def test_legacy_root_volume_filename_remains_an_alias(self, tmp_path): assert path == legacy assert not encrypted - @patch("builtins.open", mock_open(read_data="cf_token: test-token\ncf_domain_name: example.com")) - @patch("set_dns.Path.read_text", return_value="cf_token: test-token\ncf_domain_name: example.com") + @patch( + "builtins.open", + mock_open(read_data="cf_token: test-token\ncf_domain_name: example.com"), + ) + @patch( + "set_dns.Path.read_text", + return_value="cf_token: test-token\ncf_domain_name: example.com", + ) def test_reads_plaintext_yaml(self, mock_read, _): values = set_dns.read_config_file(Path("/cflan_vars.yaml"), encrypted=False) @@ -99,7 +107,9 @@ def test_reads_plaintext_yaml(self, mock_read, _): @patch("set_dns.subprocess.run") def test_reads_sops_yaml_without_writing_plaintext(self, mock_run): - mock_run.return_value.stdout = "cf_token: test-token\ncf_domain_name: example.com" + mock_run.return_value.stdout = ( + "cf_token: test-token\ncf_domain_name: example.com" + ) values = set_dns.read_config_file(Path("/cflan_sops_vars.yaml"), encrypted=True) @@ -130,7 +140,10 @@ def test_get_zone_info_requires_exact_match(self): SimpleNamespace(id="zone-id", name="example.com") ] - assert set_dns.get_zone_info(client, "example.com") == ("zone-id", "example.com") + assert set_dns.get_zone_info(client, "example.com") == ( + "zone-id", + "example.com", + ) def test_get_dns_record_rejects_duplicates(self): client = MagicMock() From f153889800d80c587ae15febafedeefbb1df6ecc Mon Sep 17 00:00:00 2001 From: Steven Welch Date: Tue, 1 Sep 2026 18:10:05 -0600 Subject: [PATCH 4/4] test: drop unbound mock parameter in test_reads_plaintext_yaml The builtins.open patch passes an explicit new= mock, so it injects no argument; the extra '_' parameter was treated as a missing pytest fixture. --- tests/test_set_dns.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_set_dns.py b/tests/test_set_dns.py index 68c5194..1974d0a 100644 --- a/tests/test_set_dns.py +++ b/tests/test_set_dns.py @@ -99,7 +99,7 @@ def test_legacy_root_volume_filename_remains_an_alias(self, tmp_path): "set_dns.Path.read_text", return_value="cf_token: test-token\ncf_domain_name: example.com", ) - def test_reads_plaintext_yaml(self, mock_read, _): + def test_reads_plaintext_yaml(self, mock_read): values = set_dns.read_config_file(Path("/cflan_vars.yaml"), encrypted=False) assert values["cf_token"] == "test-token"