From 77276ff9b5a5bec11a4a2da0774bd827b3a65625 Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 11:42:15 +0200 Subject: [PATCH 1/3] Apply one Domain on its proxy server with cloudfall domains apply cloudfall domains apply [--issue-certificate] [--yes] runs domains.yml limited to the Domain's proxy server and to that Domain, so no baseline, firewall or service task runs. Without --yes it answers a plan after every controller-side check. --issue-certificate refuses tls.mode: disabled, and checks on the controller that the primary name and every alias resolve to the proxy server (domains_dns_unverified); a proxied Domain skips the check. The resolver is a protocol passed to the lifecycle calls, so tests answer DNS offline. migrate's dns-verify shares the proxy address computation. Fixes #117 Co-Authored-By: Claude Opus 5.5 --- changelog.d/117-domains-apply.md | 3 + engine/tests/test_domains_certificate.py | 68 +++ sdk/README.md | 17 +- sdk/src/cloudfall/app.py | 94 ++++ sdk/src/cloudfall/commands.py | 27 +- sdk/src/cloudfall/lifecycle.py | 240 +++++++++- sdk/src/cloudfall/migrate.py | 28 +- sdk/tests/test_commands.py | 1 + sdk/tests/test_domains_apply.py | 583 +++++++++++++++++++++++ 9 files changed, 1023 insertions(+), 38 deletions(-) create mode 100644 changelog.d/117-domains-apply.md create mode 100644 engine/tests/test_domains_certificate.py create mode 100644 sdk/tests/test_domains_apply.py diff --git a/changelog.d/117-domains-apply.md b/changelog.d/117-domains-apply.md new file mode 100644 index 0000000..1c8b3aa --- /dev/null +++ b/changelog.d/117-domains-apply.md @@ -0,0 +1,3 @@ +### Added + +- `cloudfall domains apply [--issue-certificate] [--yes]` converges one Domain: `domains.yml` runs on its proxy server alone and renders only it, so no baseline, firewall or service task runs. Without `--yes` it exits 0 with a plan. `--issue-certificate` refuses a Domain with `tls.mode: disabled`, and first checks on the controller that the primary name and every alias resolve to the proxy server (`PRECONDITION`, `domains_dns_unverified`, naming each failing name); a `proxied` Domain skips the check (`dnsCheck: "skipped-proxied"`). With `--yes` it answers the receipt path under `--deployments` (#117) diff --git a/engine/tests/test_domains_certificate.py b/engine/tests/test_domains_certificate.py new file mode 100644 index 0000000..dd9e8f6 --- /dev/null +++ b/engine/tests/test_domains_certificate.py @@ -0,0 +1,68 @@ +"""``domains apply --issue-certificate``: the role's side (spec S-004). + +Real issuance needs Let's Encrypt and a public name, so the ``when`` +conditions of the certificate tasks are evaluated with Ansible's own +templating, and the receipt tasks run for real twice on localhost. +""" + +from __future__ import annotations + +from pathlib import Path + +from ansible.parsing.dataloader import DataLoader +from ansible.template import Templar, trust_as_template +from test_domains_one_domain import RENDERED, _run, _sha256 + +ROLES = Path(__file__).parents[2] / "engine" / "ansible" / "roles" +ROLE_TASKS = ROLES / "cloudfall_nginx_site" / "tasks" / "main.yml" + + +def _when(name: str, variables: dict[str, object]) -> bool: + """Return whether the named task of the role runs under ``variables``.""" + tasks = DataLoader().load_from_file(str(ROLE_TASKS)) + assert isinstance(tasks, list) + (task,) = (task for task in tasks if task["name"] == name) + conditions = task["when"] + templar = Templar(loader=DataLoader(), variables=variables) + return all( + templar.evaluate_conditional(trust_as_template(condition)) + for condition in (conditions if isinstance(conditions, list) else [conditions]) + ) + + +def _variables(*, certificate_exists: bool) -> dict[str, object]: + return { + "ansible_check_mode": False, + "cloudfall_nginx_site_issue_certificate": True, + "cloudfall_nginx_site_domain": {"tls": {"mode": "required"}}, + "cloudfall_nginx_site_certificate": {"stat": {"exists": certificate_exists}}, + } + + +# proves: S-004-9 (issuance runs on a fresh Domain, and the timer is enabled) +def test_s004_9_a_fresh_domain_issues_and_enables_the_renewal_timer() -> None: + fresh = _variables(certificate_exists=False) + + assert _when("Issue the declared certificate", fresh) + assert _when("Enable the certificate renewal timer", fresh) + + +def test_s004_10_a_second_run_skips_issuance() -> None: + issued = _variables(certificate_exists=True) + + assert not _when("Issue the declared certificate", issued) + assert _when("Enable the certificate renewal timer", issued) + + +def test_s004_10_a_second_run_records_the_vhost_it_rendered_before( + tmp_path: Path, +) -> None: + """The second render writes what the first left, so the hashes agree.""" + vhost = tmp_path / "crm.example.test.conf" + + first = _run(tmp_path, vhost) + second = _run(tmp_path, vhost) + + assert first["previousConfigurationSha256"] is None + assert second["previousConfigurationSha256"] == _sha256(RENDERED) + assert second["configurationSha256"] == second["previousConfigurationSha256"] diff --git a/sdk/README.md b/sdk/README.md index 1a2bdb9..2c099b6 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -57,11 +57,18 @@ uv run cloudfall restart --project config/examples crm-backend --yes uv run cloudfall health --project config/examples crm-backend ``` -`deploy`, `rollback`, `restart`, and `data copy` change servers only with -`--yes`. Without it they run every controller-side check (declared component, -verified artifact, declared database, source URL file) and print a `plan` -naming the target servers, exit `0`, so a request can be reviewed before it -runs. +`deploy`, `rollback`, `restart`, `data copy`, and `domains apply` change +servers only with `--yes`. Without it they run every controller-side check +(declared component, verified artifact, declared database, source URL file, +DNS for a certificate) and print a `plan` naming the target servers, exit `0`, +so a request can be reviewed before it runs. + +`domains apply ` runs `domains.yml` on the Domain's proxy server +alone and renders only that Domain, so no baseline, firewall or service task +runs; it is safe on a server whose baseline another project owns. With +`--issue-certificate` it first checks that the Domain's names resolve to the +proxy server (a `proxied` Domain skips the check), then issues the certificate +when none exists yet. `deploy` refuses to run when the built artifact is missing, misidentified, or fails digest verification, and refuses to report success if the engine wrote diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index 29edc7c..76ea965 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -166,6 +166,7 @@ ERROR_EXECUTION_FAILED, STEP_TIMEOUT_SECONDS, DeployOptions, + DomainApply, DomainsRun, EngineContext, ExecutionStep, @@ -173,6 +174,8 @@ LifecyclePreview, StepRun, StepRunner, + SystemNameResolver, + apply_domain, backup_service, build_release_artifact, converge_domains, @@ -182,6 +185,7 @@ migrate_data, preview_data_migration, preview_deploy, + preview_domain_apply, preview_restart, preview_rollback, restart, @@ -4052,6 +4056,96 @@ def data_copy( return DataMigratePayload({**body, "effect": "updated"}) +domains = app.group( + "domains", description="Converge declared Domains on their proxy servers" +) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class DomainApplyArgs(EngineArgs): + """Arguments of ``domains apply``.""" + + domain: ResourceId = Arg(description="Domain to apply, by its metadata.id") + issue_certificate: bool = Flag( + default=False, + description=( + "Check on the controller that every name of the Domain resolves to " + "its proxy server, then issue its certificate when none exists yet" + ), + ) + deployments: Path = Flag( + default=Path("tmp/deployments"), + description="Deployment receipt directory (default: tmp/deployments)", + ) + yes: bool = Flag( + default=False, + confirm=True, + description=( + "Change the proxy server; without it the command validates the " + "request and only shows what it would do" + ), + ) + + def __post_init__(self) -> None: + """Keep relative paths inside the project.""" + EngineArgs.__post_init__(self) + _inside_project(self.deployments, "deployments") + + +class DomainApplyPayload(Payload): + """``domains apply``: the plan without --yes, the receipt with it.""" + + command = "domains apply" + writes = True + + +@domains.command( + "apply", + description=( + "Render one Domain's virtual host on its proxy server alone, issuing " + "its certificate with --issue-certificate; without --yes it shows the plan" + ), + danger_level="mutating", + exit_codes=_LIFECYCLE_EXIT_CODES, + timeout=None, + has_network_io=True, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0"}, + supports_raw_payload=True, + examples=[ + ("Show the plan", "cloudfall domains apply crm-site"), + ( + "Apply it and issue its certificate", + "cloudfall domains apply crm-site --issue-certificate --yes", + ), + ], +) +def domains_apply( + args: DomainApplyArgs, ctx: Ctx, fleet: Fleet +) -> DomainApplyPayload: + """Converge the Domain on its proxy server alone, or show the plan.""" + context = args.context(fleet, ctx) + request = DomainApply( + domain_id=args.domain, + issue_certificate=args.issue_certificate, + receipt_directory=fleet.path(args.deployments), + ) + try: + if not args.yes: + preview = preview_domain_apply(context, request, SystemNameResolver()) + return DomainApplyPayload( + { + **preview.as_dict(), + "instruction": _PLAN_INSTRUCTION, + "effect": "would_update", + } + ) + applied = apply_domain(context, request, SystemNameResolver()) + except LifecycleError as error: + raise _lifecycle_failed(error) from error + return DomainApplyPayload({**applied.as_dict(), "effect": "updated"}) + + # mcp serve: the fleet check before serving, and the tools beside the commands diff --git a/sdk/src/cloudfall/commands.py b/sdk/src/cloudfall/commands.py index d36f12a..f714dc2 100644 --- a/sdk/src/cloudfall/commands.py +++ b/sdk/src/cloudfall/commands.py @@ -77,9 +77,10 @@ class OutputKey: **dict.fromkeys( ( "action", "application", "component", "database", "detail", - "directory", "environmentFile", "health", "instruction", "output", - "pass", "path", "project", "receipt", "release", "report", - "service", "sha256", "status", "step", "verification", "wouldRun", + "directory", "dnsCheck", "domain", "environmentFile", "health", + "instruction", "output", "pass", "path", "project", "receipt", + "release", "report", "service", "sha256", "status", "step", + "verification", "wouldRun", ), "string", ), @@ -523,6 +524,26 @@ def _plan(subject: str, *extra: str) -> OutputShape: ), ), ), + CommandContract( + "cloudfall", + "domains apply", + CommandEffect.SERVERS, + "render one Domain's virtual host, and its certificate on request, on " + "its proxy server alone; no baseline, firewall or service task runs", + gate=_YES, + output=( + _plan("domain", "dnsCheck?"), + _shape( + "status", + "action", + "domain", + "servers", + "dnsCheck?", + "receipt", + when="with `--yes`; `dnsCheck` only with `--issue-certificate`", + ), + ), + ), CommandContract( "cloudfall", "migrate", diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index 645575a..dd968a2 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -12,16 +12,18 @@ import json import os import shutil +import socket import subprocess import sys from dataclasses import dataclass from enum import StrEnum +from ipaddress import ip_address from pathlib import PurePosixPath from typing import TYPE_CHECKING, Protocol, cast from jsonschema.exceptions import ValidationError -from cloudfall.domain import AbsolutePath +from cloudfall.domain import AbsolutePath, DnsMode, Hostname, IpAddress, TlsMode from cloudfall.inventory import PlatformInventory from cloudfall.secrets import ( SecretsError, @@ -44,14 +46,18 @@ from cloudfall.inventory import ( ComponentInventory, DomainInventory, + ServerInventory, ServiceInventory, ) from cloudfall.secrets import OwnedRolePassword, SopsSecretProvider _ARTIFACT_SCHEMA = "artifact.schema.json" _OUTPUT_TAIL_CHARACTERS = 2000 +_DNS_PROBE_PORT = 443 _ERROR_COMPONENT_MISSING = "lifecycle_component_missing" _ERROR_DOMAIN_MISSING = "lifecycle_domain_missing" +_ERROR_DOMAIN_TLS_DISABLED = "domains_tls_disabled" +_ERROR_DOMAIN_DNS_UNVERIFIED = "domains_dns_unverified" _ERROR_ARTIFACT_MISSING = "lifecycle_artifact_missing" _ERROR_ARTIFACT_METADATA = "lifecycle_artifact_metadata_invalid" _ERROR_ARTIFACT_IDENTITY = "lifecycle_artifact_identity_mismatch" @@ -506,16 +512,229 @@ def converge_domains(context: EngineContext, run: DomainsRun) -> None: One Domain is looked up in the validated config first, so an unknown id is refused before the inventory render and before any SSH connection. """ + if run.domain_id is None: + run_engine_playbook(context, "domains.yml", _domains_vars(run)) + return + _converge_one_domain(context, _domain(context, run.domain_id), run) + + +def _domains_vars(run: DomainsRun) -> dict[str, object]: extra_vars: dict[str, object] = { "cloudfall_domains_issue_certificates": run.issue_certificates, "cloudfall_domains_receipt_directory": str(run.receipt_directory), } - if run.domain_id is None: - run_engine_playbook(context, "domains.yml", extra_vars) - return - proxy_server = _domain(context, run.domain_id).proxy.server_id - extra_vars["cloudfall_domains_domain_id"] = run.domain_id.value - run_engine_playbook(context, "domains.yml", extra_vars, limit=proxy_server) + if run.domain_id is not None: + extra_vars["cloudfall_domains_domain_id"] = run.domain_id.value + return extra_vars + + +def _converge_one_domain( + context: EngineContext, domain: DomainInventory, run: DomainsRun +) -> None: + run_engine_playbook( + context, "domains.yml", _domains_vars(run), limit=domain.proxy.server_id + ) + + +class NameResolver(Protocol): + """Resolves a public name on the controller, as a browser would.""" + + def resolve(self, name: Hostname) -> frozenset[IpAddress]: + """Return every address ``name`` resolves to; empty when none.""" + ... + + +class SystemNameResolver: + """The controller's own resolver, through ``getaddrinfo``.""" + + def resolve(self, name: Hostname) -> frozenset[IpAddress]: + """Return the name's addresses; a name that does not resolve has none.""" + try: + records = socket.getaddrinfo( + name.value, _DNS_PROBE_PORT, type=socket.SOCK_STREAM + ) + except socket.gaierror: + return frozenset() + # A scoped link-local address (fe80::1%en0) is no public answer. + return frozenset( + IpAddress.from_boundary(str(record[4][0])) + for record in records + if "%" not in str(record[4][0]) + ) + + +def proxy_addresses( + server: ServerInventory, resolver: NameResolver +) -> frozenset[IpAddress]: + """Return the addresses a name pointed at ``server`` may resolve to. + + The server's address (or what its hostname resolves to) and its declared + ``network.ipv4``, as the migration's ``dns-verify`` step computes them. + """ + addresses: set[IpAddress] = set() + try: + ip_address(server.address.value) + except ValueError: + addresses.update(resolver.resolve(Hostname(server.address.value))) + else: + addresses.add(IpAddress(server.address.value)) + if server.network is not None: + addresses.add(IpAddress(server.network.ipv4.value)) + return frozenset(addresses) + + +class DnsCheck(StrEnum): + """What the controller-side DNS check of ``domains apply`` found.""" + + VERIFIED = "verified" + SKIPPED_PROXIED = "skipped-proxied" + + +@dataclass(frozen=True, slots=True) +class DomainApply: + """One ``domains apply`` request.""" + + domain_id: ResourceId + issue_certificate: bool + receipt_directory: Path + + +@dataclass(frozen=True, slots=True) +class DomainApplied: + """What ``domains apply`` checked, and would run or ran, for one Domain.""" + + domain_id: ResourceId + server: ResourceId + issue_certificate: bool + dns_check: DnsCheck | None + receipt: Path | None = None + """The receipt the run wrote; ``None`` for a plan.""" + + @property + def would_run(self) -> str: + """Say what the run changes, and where.""" + certificate = ( + ", issue its certificate by webroot when none exists yet, render it " + "again with TLS, and enable certbot.timer" + if self.issue_certificate + else "" + ) + return ( + f"render Domain {self.domain_id} as a managed Nginx virtual host on " + f"{self.server} alone through domains.yml{certificate}; no baseline, " + "firewall or service task runs" + ) + + def as_dict(self) -> dict[str, object]: + """Serialize the plan, or with a receipt the run, for agent consumers.""" + result: dict[str, object] = { + "status": "plan" if self.receipt is None else "ok", + "action": "domains-apply", + "domain": self.domain_id.value, + "servers": [self.server.value], + } + if self.receipt is None: + result["wouldRun"] = self.would_run + else: + result["receipt"] = str(self.receipt) + if self.dns_check is not None: + result["dnsCheck"] = self.dns_check.value + return result + + +def preview_domain_apply( + context: EngineContext, request: DomainApply, resolver: NameResolver +) -> DomainApplied: + """Run every controller-side check of ``domains apply``; change nothing. + + The Domain must be declared, ``--issue-certificate`` needs ``tls.mode: + required``, and a ``dns-only`` Domain's names must resolve to its proxy + server before a certificate is asked for. Nothing here renders the + inventory or contacts a server. + """ + state = validate_config(context.project_directory, context.schema_directory) + inventory = PlatformInventory.from_state(state) + domain = _find_domain(inventory, request.domain_id) + dns_check = None + if request.issue_certificate: + if domain.tls_mode is TlsMode.DISABLED: + detail = ( + f"domain {domain.resource_id} declares tls.mode: disabled, so " + "there is no certificate to issue; drop --issue-certificate" + ) + raise LifecycleError(_ERROR_DOMAIN_TLS_DISABLED, detail) + servers = {server.resource_id: server for server in inventory.servers} + dns_check = _check_domain_dns( + domain, servers[domain.proxy.server_id], resolver + ) + return DomainApplied( + domain_id=domain.resource_id, + server=domain.proxy.server_id, + issue_certificate=request.issue_certificate, + dns_check=dns_check, + ) + + +def apply_domain( + context: EngineContext, request: DomainApply, resolver: NameResolver +) -> DomainApplied: + """Converge one Domain on its proxy server alone, and return its receipt. + + The checks of :func:`preview_domain_apply` run first, so a refused + request renders no inventory and runs no playbook. + """ + checked = preview_domain_apply(context, request, resolver) + run_engine_playbook( + context, + "domains.yml", + _domains_vars( + DomainsRun( + issue_certificates=request.issue_certificate, + receipt_directory=request.receipt_directory, + domain_id=request.domain_id, + ) + ), + limit=checked.server, + ) + receipt = request.receipt_directory / f"{request.domain_id.value}.json" + if not receipt.is_file(): + detail = f"domains.yml reported success but wrote no receipt: {receipt}" + raise LifecycleError(_ERROR_RECEIPT_MISSING, detail) + return DomainApplied( + domain_id=checked.domain_id, + server=checked.server, + issue_certificate=checked.issue_certificate, + dns_check=checked.dns_check, + receipt=receipt, + ) + + +def _check_domain_dns( + domain: DomainInventory, server: ServerInventory, resolver: NameResolver +) -> DnsCheck: + """Require every name of a ``dns-only`` Domain to reach its proxy server.""" + if domain.edge.mode is DnsMode.PROXIED: + return DnsCheck.SKIPPED_PROXIED + expected = proxy_addresses(server, resolver) + failing: list[str] = [] + for name in (domain.primary_name, *domain.aliases): + resolved = resolver.resolve(name) + if not resolved & expected: + answer = _address_list(resolved) if resolved else "nothing" + failing.append(f"{name.value} resolves to {answer}") + if failing: + detail = ( + f"point these names of domain {domain.resource_id} at proxy server " + f"{server.resource_id} ({_address_list(expected) or 'no known address'})" + ", wait for " + "propagation, and rerun: " + "; ".join(failing) + ) + raise LifecycleError(_ERROR_DOMAIN_DNS_UNVERIFIED, detail) + return DnsCheck.VERIFIED + + +def _address_list(addresses: frozenset[IpAddress]) -> str: + return ", ".join(sorted(address.value for address in addresses)) ERROR_SERVICES_SECRET = "lifecycle_services_secret_unavailable" # noqa: S105 - a code @@ -1027,7 +1246,12 @@ def _component( def _domain(context: EngineContext, domain_id: ResourceId) -> DomainInventory: state = validate_config(context.project_directory, context.schema_directory) - inventory = PlatformInventory.from_state(state) + return _find_domain(PlatformInventory.from_state(state), domain_id) + + +def _find_domain( + inventory: PlatformInventory, domain_id: ResourceId +) -> DomainInventory: domain = next( ( candidate diff --git a/sdk/src/cloudfall/migrate.py b/sdk/src/cloudfall/migrate.py index cc42e5e..b20e7a6 100644 --- a/sdk/src/cloudfall/migrate.py +++ b/sdk/src/cloudfall/migrate.py @@ -11,11 +11,9 @@ from __future__ import annotations import json -import socket from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from enum import StrEnum -from ipaddress import ip_address from typing import TYPE_CHECKING, cast from cloudfall.audit import AuditStatus, audit_inventory @@ -30,9 +28,11 @@ from cloudfall.lifecycle import ( DeployOptions, LifecycleError, + SystemNameResolver, build_release_artifact, deploy, migrate_data, + proxy_addresses, run_engine_playbook, run_services_playbook, services_passwords, @@ -666,16 +666,10 @@ def _verify_dns(inventory: PlatformInventory) -> dict[str, object]: def _expected_addresses(server: ServerInventory) -> set[str]: - addresses: set[str] = set() - try: - ip_address(server.address.value) - except ValueError: - addresses.update(_resolve_names(server.address.value)) - else: - addresses.add(server.address.value) - if server.network is not None: - addresses.add(server.network.ipv4.value) - return addresses + return { + address.value + for address in proxy_addresses(server, SystemNameResolver()) + } def _resolved_addresses( @@ -690,16 +684,6 @@ def _resolved_addresses( return {address.value for address in resolved} -def _resolve_names(hostname: str) -> set[str]: - try: - records = socket.getaddrinfo( - hostname, _DNS_PROBE_PORT, type=socket.SOCK_STREAM - ) - except OSError: - return set() - return {str(record[4][0]) for record in records} - - def _verify_audit(config: AgentConfig) -> dict[str, object]: state = validate_config(config.project_directory, config.schema_directory) observations = load_observations( diff --git a/sdk/tests/test_commands.py b/sdk/tests/test_commands.py index 95bdacb..35861be 100644 --- a/sdk/tests/test_commands.py +++ b/sdk/tests/test_commands.py @@ -64,6 +64,7 @@ def test_every_yes_gated_command_is_classified_as_changing_servers() -> None: "rollback", "restart", "data copy", + "domains apply", "migrate", } assert gated <= changing diff --git a/sdk/tests/test_domains_apply.py b/sdk/tests/test_domains_apply.py new file mode 100644 index 0000000..91fa032 --- /dev/null +++ b/sdk/tests/test_domains_apply.py @@ -0,0 +1,583 @@ +"""``cloudfall domains apply``: one Domain on its proxy server (spec S-004). + +DNS answers come from a stand-in resolver passed to the lifecycle calls, so +no test asks a real resolver. The CLI tests take only the paths that resolve +nothing: the plain plan, a proxied Domain, and the refusals before DNS. +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import subprocess +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import pytest +import yaml +from cloudfall.cli import main +from cloudfall.domain import Hostname, IpAddress, ResourceId +from cloudfall.lifecycle import ( + DnsCheck, + DomainApply, + EngineContext, + ExecutionStep, + LifecycleError, + StepRun, + StepRunner, + apply_domain, + preview_domain_apply, +) + +if TYPE_CHECKING: + from collections.abc import Sequence + +ROOT = Path(__file__).parents[2] +SCHEMAS = ROOT / "config" / "schemas" / "v1" +EXAMPLES = ROOT / "config" / "examples" +ENGINE = ROOT / "engine" +PLAYBOOKS = ENGINE / "ansible" / "playbooks" +NGINX_SITE = ENGINE / "ansible" / "roles" / "cloudfall_nginx_site" +# crm-site in the examples: proxy server h2 (address h2.example.internal), +# primaryName crm.example.test, one alias www.crm.example.test. +PROXY = "203.0.113.20" +ELSEWHERE = "198.51.100.7" + + +@dataclass(slots=True) +class _Resolver: + """Answer from a fixed table, keeping every name asked.""" + + answers: dict[str, frozenset[IpAddress]] + asked: list[str] = field(default_factory=list) + + def resolve(self, name: Hostname) -> frozenset[IpAddress]: + self.asked.append(name.value) + return self.answers.get(name.value, frozenset()) + + +def _resolver(**overrides: str) -> _Resolver: + """Every name at the proxy server, unless overridden by name.""" + answers = { + "h2.example.internal": PROXY, + "crm.example.test": PROXY, + "www.crm.example.test": PROXY, + } + answers.update({name.replace("_", "."): value for name, value in overrides.items()}) + return _Resolver( + {name: frozenset({IpAddress(value)}) for name, value in answers.items()} + ) + + +@dataclass(frozen=True, slots=True) +class _Done: + returncode: int + stdout: str = "" + stderr: str = "" + + +@dataclass(slots=True) +class _Recorder: + """Answer each step, writing the receipt a playbook run would write.""" + + receipts: Path + steps: list[ExecutionStep] = field(default_factory=list) + + def __call__(self, step: ExecutionStep) -> _Done: + self.steps.append(step) + if step.description == "run domains.yml": + self.receipts.mkdir(parents=True, exist_ok=True) + (self.receipts / "crm-site.json").write_text("{}", encoding="utf-8") + return _Done(0) + + +def _project(tmp_path: Path, **spec: object) -> Path: + """Copy the examples, with crm-site's spec keys replaced by ``spec``.""" + project = tmp_path / "project" + shutil.copytree(EXAMPLES, project) + path = project / "domains" / "crm-site.yaml" + document = yaml.safe_load(path.read_text(encoding="utf-8")) + document["spec"].update(spec) + path.write_text(yaml.safe_dump(document), encoding="utf-8") + return project + + +def _context(project: Path, run: StepRunner) -> EngineContext: + return EngineContext( + project_directory=project, + schema_directory=SCHEMAS, + engine_directory=ENGINE, + inventory_file=project / "tmp" / "inventory.json", + run=run, + ) + + +def _request(project: Path, *, issue_certificate: bool) -> DomainApply: + return DomainApply( + domain_id=ResourceId("crm-site"), + issue_certificate=issue_certificate, + receipt_directory=project / "tmp" / "deployments", + ) + + +def _extra_vars(argv: Sequence[str]) -> dict[str, object]: + extra_vars = json.loads(argv[list(argv).index("--extra-vars") + 1]) + assert isinstance(extra_vars, dict) + return extra_vars + + +def _cli( + project: Path, capsys: pytest.CaptureFixture[str], *argv: str +) -> tuple[int, dict[str, Any]]: + exit_code = main( + [ + "domains", + "apply", + *argv, + "--project", + str(project), + "--schemas", + str(SCHEMAS), + "--engine", + str(ENGINE), + ] + ) + payload = json.loads(capsys.readouterr().out) + assert isinstance(payload, dict) + return exit_code, payload + + +def _untouched(project: Path) -> None: + """Neither the inventory render nor a playbook ran, and no receipt exists.""" + assert not (project / "tmp" / "ansible-inventory.json").exists() + assert not (project / "tmp" / "inventory.json").exists() + assert not (project / "tmp" / "deployments").exists() + + +# S-004-3: the plan + + +def test_s004_3_without_yes_the_plan_runs_nothing( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + project = _project(tmp_path) + + exit_code, payload = _cli(project, capsys, "crm-site") + + assert exit_code == 0, payload + data = payload["data"] + assert data["status"] == "plan" + assert data["effect"] == "would_update" + assert data["action"] == "domains-apply" + assert data["domain"] == "crm-site" + assert data["servers"] == ["h2"] + assert "h2" in data["would_run"] + assert "--yes" in data["instruction"] + assert data["dns_check"] is None + assert payload["meta"]["dry_run"] is True + _untouched(project) + + +def test_s004_3_the_plan_starts_no_engine_step(tmp_path: Path) -> None: + project = _project(tmp_path) + recorder = _Recorder(project / "tmp" / "deployments") + resolver = _resolver() + + planned = preview_domain_apply( + _context(project, recorder), + _request(project, issue_certificate=True), + resolver, + ) + + assert recorder.steps == [] + assert planned.as_dict()["status"] == "plan" + assert planned.dns_check is DnsCheck.VERIFIED + _untouched(project) + + +@pytest.mark.parametrize("yes", [[], ["--yes"]], ids=["plan", "yes"]) +def test_an_unknown_domain_is_refused_before_any_engine_step( + yes: list[str], tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + project = _project(tmp_path) + + exit_code, payload = _cli(project, capsys, "nope", *yes) + + assert exit_code == 4, payload + assert payload["error"]["code"] == "PRECONDITION" + assert payload["error"]["context"]["code"] == "lifecycle_domain_missing" + _untouched(project) + + +# S-004-5: an invalid config + + +@pytest.mark.parametrize("yes", [[], ["--yes"]], ids=["plan", "yes"]) +def test_s004_5_an_invalid_config_exits_before_any_engine_step( + yes: list[str], tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + project = _project(tmp_path, proxy={"server": "ghost"}) + + exit_code, payload = _cli(project, capsys, "crm-site", *yes) + + assert exit_code != 0 + assert payload["error"]["code"] == "CONFIG_INVALID" + _untouched(project) + + +# S-004-6: names that do not reach the proxy server + + +@pytest.mark.parametrize("yes", [False, True], ids=["plan", "yes"]) +def test_s004_6_an_alias_elsewhere_is_refused_naming_it( + yes: bool, # noqa: FBT001 - a pytest parameter + tmp_path: Path, +) -> None: + project = _project(tmp_path) + recorder = _Recorder(project / "tmp" / "deployments") + resolver = _resolver(www_crm_example_test=ELSEWHERE) + call = apply_domain if yes else preview_domain_apply + + with pytest.raises(LifecycleError) as error: + call( + _context(project, recorder), + _request(project, issue_certificate=True), + resolver, + ) + + assert error.value.code == "domains_dns_unverified" + assert f"www.crm.example.test resolves to {ELSEWHERE}" in error.value.detail + # The primary name reaches the proxy, so only the alias is named. + assert error.value.detail.count("crm.example.test resolves") == 1 + assert PROXY in error.value.detail + assert recorder.steps == [] + _untouched(project) + + +def test_s004_6_a_name_that_resolves_to_nothing_is_refused(tmp_path: Path) -> None: + project = _project(tmp_path) + recorder = _Recorder(project / "tmp" / "deployments") + resolver = _resolver() + del resolver.answers["crm.example.test"] + + with pytest.raises(LifecycleError) as error: + apply_domain( + _context(project, recorder), + _request(project, issue_certificate=True), + resolver, + ) + + assert error.value.code == "domains_dns_unverified" + assert "crm.example.test resolves to nothing" in error.value.detail + assert recorder.steps == [] + + +def test_a_proxy_server_address_that_is_an_ip_is_taken_as_it_is( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + server = project / "servers" / "h2.yaml" + document = yaml.safe_load(server.read_text(encoding="utf-8")) + document["spec"]["address"] = "192.0.2.10" + server.write_text(yaml.safe_dump(document), encoding="utf-8") + resolver = _resolver( + crm_example_test="192.0.2.10", www_crm_example_test="192.0.2.10" + ) + + planned = preview_domain_apply( + _context(project, _Recorder(tmp_path / "unused")), + _request(project, issue_certificate=True), + resolver, + ) + + assert planned.dns_check is DnsCheck.VERIFIED + assert resolver.asked == ["crm.example.test", "www.crm.example.test"] + + +def test_without_issue_certificate_dns_is_not_checked(tmp_path: Path) -> None: + project = _project(tmp_path) + resolver = _resolver(www_crm_example_test=ELSEWHERE) + + planned = preview_domain_apply( + _context(project, _Recorder(tmp_path / "unused")), + _request(project, issue_certificate=False), + resolver, + ) + + assert planned.dns_check is None + assert "dnsCheck" not in planned.as_dict() + assert resolver.asked == [] + + +# S-004-7: a proxied Domain + + +def test_s004_7_a_proxied_domain_skips_the_dns_check( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + project = _project(tmp_path, edge={"provider": "cloudflare", "mode": "proxied"}) + + exit_code, payload = _cli(project, capsys, "crm-site", "--issue-certificate") + + assert exit_code == 0, payload + assert payload["data"]["dns_check"] == "skipped-proxied" + assert payload["data"]["status"] == "plan" + _untouched(project) + + +def test_s004_7_a_proxied_domain_asks_no_resolver(tmp_path: Path) -> None: + project = _project(tmp_path, edge={"provider": "cloudflare", "mode": "proxied"}) + recorder = _Recorder(project / "tmp" / "deployments") + resolver = _resolver(crm_example_test=ELSEWHERE) + + applied = apply_domain( + _context(project, recorder), + _request(project, issue_certificate=True), + resolver, + ) + + assert resolver.asked == [] + assert applied.as_dict()["dnsCheck"] == "skipped-proxied" + + +# S-004-8: no certificate for tls.mode disabled + + +@pytest.mark.parametrize("yes", [[], ["--yes"]], ids=["plan", "yes"]) +def test_s004_8_tls_disabled_refuses_issue_certificate( + yes: list[str], tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + project = _project(tmp_path, tls={"mode": "disabled"}) + + exit_code, payload = _cli(project, capsys, "crm-site", "--issue-certificate", *yes) + + assert exit_code == 4, payload + assert payload["error"]["code"] == "PRECONDITION" + assert payload["error"]["context"]["code"] == "domains_tls_disabled" + _untouched(project) + + +def test_s004_8_tls_disabled_is_refused_before_dns(tmp_path: Path) -> None: + project = _project(tmp_path, tls={"mode": "disabled"}) + recorder = _Recorder(project / "tmp" / "deployments") + resolver = _resolver() + + with pytest.raises(LifecycleError) as error: + apply_domain( + _context(project, recorder), + _request(project, issue_certificate=True), + resolver, + ) + + assert error.value.code == "domains_tls_disabled" + assert resolver.asked == [] + assert recorder.steps == [] + + +# S-004-9: the certificate run + + +def test_s004_9_issue_certificate_passes_the_flag_to_domains_yml( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + receipts = project / "tmp" / "deployments" + recorder = _Recorder(receipts) + + applied = apply_domain( + _context(project, recorder), + _request(project, issue_certificate=True), + _resolver(), + ) + + render, playbook = recorder.steps + assert render.description == "render Ansible inventory" + assert playbook.argv[-1] == str(PLAYBOOKS / "domains.yml") + assert playbook.argv[playbook.argv.index("--limit") + 1] == "h2" + assert _extra_vars(playbook.argv) == { + "cloudfall_domains_issue_certificates": True, + "cloudfall_domains_receipt_directory": str(receipts), + "cloudfall_domains_domain_id": "crm-site", + } + assert applied.as_dict() == { + "status": "ok", + "action": "domains-apply", + "domain": "crm-site", + "servers": ["h2"], + "receipt": str(receipts / "crm-site.json"), + "dnsCheck": "verified", + } + + +def test_a_run_that_writes_no_receipt_is_refused(tmp_path: Path) -> None: + project = _project(tmp_path) + + def no_receipt(_step: ExecutionStep) -> StepRun: + return _Done(0) + + with pytest.raises(LifecycleError) as error: + apply_domain( + _context(project, no_receipt), + _request(project, issue_certificate=False), + _resolver(), + ) + + assert error.value.code == "lifecycle_receipt_missing" + + +# S-004-1: domains.yml for real, on the proxy server only + + +STUB_ROLE = """--- +- name: Record what this host rendered + ansible.builtin.lineinfile: + path: {log} + line: >- + {{{{ cloudfall_server_id }}}}:{{{{ cloudfall_nginx_site_domain.id }}}}:{{{{ + cloudfall_nginx_site_issue_certificate | bool }}}} + create: true + mode: "0644" + delegate_to: localhost + become: false + +- name: Create the receipt directory + ansible.builtin.file: + path: "{{{{ cloudfall_nginx_site_receipt_directory }}}}" + state: directory + mode: "0755" + delegate_to: localhost + become: false + +- name: Write a stand-in receipt + ansible.builtin.copy: + content: "{{}}" + dest: >- + {{{{ cloudfall_nginx_site_receipt_directory }}}}/{{{{ + cloudfall_nginx_site_domain.id }}}}.json + mode: "0644" + delegate_to: localhost + become: false +""" + + +@dataclass(slots=True) +class _LocalRun: + """Run each step for real, with every server a local connection. + + The rendered inventory keeps every host and group; only how Ansible + reaches a host changes. ``cloudfall_nginx_site`` is swapped for a stub + through ``ANSIBLE_ROLES_PATH``, since the real role needs Debian. + """ + + inventory_file: Path + roles: Path + playbook_output: list[str] = field(default_factory=list) + + def __call__(self, step: ExecutionStep) -> StepRun: + environment = {**os.environ, **step.environment} + if step.description.startswith("run "): + self._connect_locally() + environment["ANSIBLE_ROLES_PATH"] = str(self.roles) + done = subprocess.run( # noqa: S603 - argv built by the SDK. + list(step.argv), + check=False, + capture_output=True, + text=True, + timeout=600, + env=environment, + ) + if step.description.startswith("run "): + self.playbook_output.append(done.stdout + done.stderr) + return done + + def _connect_locally(self) -> None: + rendered = json.loads(self.inventory_file.read_text(encoding="utf-8")) + hosts = rendered["all"]["children"]["cloudfall_servers"]["hosts"] + assert sorted(hosts) == ["h1", "h2"] + for host in hosts.values(): + host.update( + { + "ansible_connection": "local", + "ansible_python_interpreter": sys.executable, + "ansible_become": False, + } + ) + self.inventory_file.write_text(json.dumps(rendered), encoding="utf-8") + + +def _local_run(tmp_path: Path, project: Path) -> tuple[_LocalRun, Path]: + log = tmp_path / "hosts.log" + tasks = tmp_path / "roles" / "cloudfall_nginx_site" / "tasks" + tasks.mkdir(parents=True) + (tasks / "main.yml").write_text( + STUB_ROLE.format(log=json.dumps(str(log))), encoding="utf-8" + ) + run = _LocalRun(project / "tmp" / "inventory.json", tmp_path / "roles") + return run, log + + +def _recap_hosts(output: str) -> list[str]: + recap = output.split("PLAY RECAP", 1)[1] + return re.findall(r"^(\S+)\s+:\s+ok=", recap, flags=re.MULTILINE) + + +DOMAINS_TASKS = { + "Gathering Facts", + "Require an upstream on every Domain this run renders", + "Render declared domains as managed virtual hosts", + "cloudfall_nginx_site : Record what this host rendered", + "cloudfall_nginx_site : Create the receipt directory", + "cloudfall_nginx_site : Write a stand-in receipt", +} + + +# proves: S-004-9 (the flag reaches the role, and the receipt is written) +@pytest.mark.parametrize("issue_certificate", [False, True], ids=["http", "tls"]) +def test_s004_1_only_the_proxy_server_runs_domains_yml( + issue_certificate: bool, # noqa: FBT001 - a pytest parameter + tmp_path: Path, +) -> None: + project = _project(tmp_path) + run, log = _local_run(tmp_path, project) + receipts = project / "tmp" / "deployments" + + applied = apply_domain( + _context(project, run), + _request(project, issue_certificate=issue_certificate), + _resolver(), + ) + + (output,) = run.playbook_output + assert _recap_hosts(output) == ["h2"], output + ran = set(re.findall(r"^TASK \[(.+)\]", output, flags=re.MULTILINE)) + assert ran <= DOMAINS_TASKS, output + assert log.read_text(encoding="utf-8").splitlines() == [ + f"h2:crm-site:{issue_certificate}" + ] + assert (receipts / "crm-site.json").is_file() + assert applied.receipt == receipts / "crm-site.json" + + +def test_s004_1_domains_yml_runs_no_baseline_firewall_or_service_role() -> None: + """The stub above stands in for this role; nothing else can run.""" + plays = yaml.safe_load((PLAYBOOKS / "domains.yml").read_text(encoding="utf-8")) + included: list[str] = [] + for play in plays: + assert set(play) & {"roles", "import_playbook", "pre_tasks", "post_tasks"} == ( + set() + ) + for task in play["tasks"]: + role = task.get("ansible.builtin.include_role") + if role is not None: + included.append(role["name"]) + assert included == ["cloudfall_nginx_site"] + meta = yaml.safe_load((NGINX_SITE / "meta" / "main.yml").read_text("utf-8")) + assert meta["dependencies"] == [] + for tasks in (NGINX_SITE / "tasks").glob("*.yml"): + text = tasks.read_text(encoding="utf-8") + for keyword in ("include_role", "import_role", "import_playbook"): + assert keyword not in text, tasks From 144789606de4bf82e4b4a6f2dbe9c815a408c1f4 Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 11:53:05 +0200 Subject: [PATCH 2/3] Keep migrate's dns-verify exactly as before domains apply migrate goes back to its own address computation: any OSError from the server hostname lookup counts as no address, and a scoped link-local address is kept as written. Tests pin both. Co-Authored-By: Claude Opus 5.5 --- sdk/src/cloudfall/migrate.py | 28 ++++++++++++++----- sdk/tests/test_migrate.py | 52 ++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 6 deletions(-) diff --git a/sdk/src/cloudfall/migrate.py b/sdk/src/cloudfall/migrate.py index b20e7a6..cc42e5e 100644 --- a/sdk/src/cloudfall/migrate.py +++ b/sdk/src/cloudfall/migrate.py @@ -11,9 +11,11 @@ from __future__ import annotations import json +import socket from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from enum import StrEnum +from ipaddress import ip_address from typing import TYPE_CHECKING, cast from cloudfall.audit import AuditStatus, audit_inventory @@ -28,11 +30,9 @@ from cloudfall.lifecycle import ( DeployOptions, LifecycleError, - SystemNameResolver, build_release_artifact, deploy, migrate_data, - proxy_addresses, run_engine_playbook, run_services_playbook, services_passwords, @@ -666,10 +666,16 @@ def _verify_dns(inventory: PlatformInventory) -> dict[str, object]: def _expected_addresses(server: ServerInventory) -> set[str]: - return { - address.value - for address in proxy_addresses(server, SystemNameResolver()) - } + addresses: set[str] = set() + try: + ip_address(server.address.value) + except ValueError: + addresses.update(_resolve_names(server.address.value)) + else: + addresses.add(server.address.value) + if server.network is not None: + addresses.add(server.network.ipv4.value) + return addresses def _resolved_addresses( @@ -684,6 +690,16 @@ def _resolved_addresses( return {address.value for address in resolved} +def _resolve_names(hostname: str) -> set[str]: + try: + records = socket.getaddrinfo( + hostname, _DNS_PROBE_PORT, type=socket.SOCK_STREAM + ) + except OSError: + return set() + return {str(record[4][0]) for record in records} + + def _verify_audit(config: AgentConfig) -> dict[str, object]: state = validate_config(config.project_directory, config.schema_directory) observations = load_observations( diff --git a/sdk/tests/test_migrate.py b/sdk/tests/test_migrate.py index 8ce7d8d..79bb811 100644 --- a/sdk/tests/test_migrate.py +++ b/sdk/tests/test_migrate.py @@ -3,21 +3,34 @@ from __future__ import annotations import json +from dataclasses import replace from pathlib import Path from typing import TYPE_CHECKING import pytest from cloudfall.agent_tools import AgentConfig from cloudfall.cli import main +from cloudfall.domain import ( + ConnectionAddress, + Hostname, + Ipv4Address, + Ipv6NetworkCidr, +) +from cloudfall.inventory import PlatformInventory, ServerNetwork from cloudfall.migrate import ( MigrateError, MigrateOptions, + _expected_addresses, + _resolve_names, execute_migration, ) +from cloudfall.validation import validate_config if TYPE_CHECKING: from collections.abc import Callable, Mapping + from cloudfall.inventory import ServerInventory + ROOT = Path(__file__).parents[2] SCHEMAS = ROOT / "config" / "schemas" / "v1" EXAMPLES = ROOT / "config" / "examples" @@ -251,3 +264,42 @@ def test_cli_migrate_rejects_malformed_component_pairs( assert "COMPONENT" in payload["error"]["message"] or "NAME=VALUE" in ( payload["error"]["message"] ) + + +# dns-verify's proxy addresses, pinned as they were before domains apply. + + +def _proxy_server(address: str) -> ServerInventory: + state = validate_config(EXAMPLES, SCHEMAS) + (server,) = ( + server + for server in PlatformInventory.from_state(state).servers + if server.resource_id.value == "h2" + ) + return replace( + server, + address=ConnectionAddress(address), + network=ServerNetwork( + ipv4=Ipv4Address("192.0.2.10"), + ipv6_cidr=Ipv6NetworkCidr("2001:db8::/64"), + reverse_dns=Hostname("h2.example.test"), + ), + ) + + +def test_dns_verify_counts_no_address_for_an_unresolvable_server_hostname() -> None: + """A hostname that does not resolve adds nothing and raises nothing. + + ``.invalid`` never resolves (RFC 6761): offline the lookup fails, online + it answers NXDOMAIN, and both are an ``OSError`` dns-verify swallows. + """ + server = _proxy_server("h2.cloudfall.invalid") + + assert _resolve_names("h2.cloudfall.invalid") == set() + assert _expected_addresses(server) == {"192.0.2.10"} + + +def test_dns_verify_keeps_a_scoped_link_local_address_as_written() -> None: + server = _proxy_server("fe80::1%eth0") + + assert _expected_addresses(server) == {"fe80::1%eth0", "192.0.2.10"} From 51c819313d686e0062cc3149e7b276b7505dc0a9 Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 12:09:36 +0200 Subject: [PATCH 3/3] Compare a non-canonical IPv6 proxy address canonically in domains apply The server schema accepts any IPv6 spelling (2001:DB8::10), but IpAddress requires canonical notation, so --issue-certificate crashed with a ValueError instead of running the DNS check. Canonicalise the parsed address before building the expected set. Co-Authored-By: Claude Opus 5.5 --- sdk/src/cloudfall/lifecycle.py | 5 +++-- sdk/tests/test_domains_apply.py | 20 ++++++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index dd968a2..18d9739 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -573,11 +573,12 @@ def proxy_addresses( """ addresses: set[IpAddress] = set() try: - ip_address(server.address.value) + parsed = ip_address(server.address.value) except ValueError: addresses.update(resolver.resolve(Hostname(server.address.value))) else: - addresses.add(IpAddress(server.address.value)) + # The schema takes any IPv6 spelling (2001:DB8::10); compare canonically. + addresses.add(IpAddress(str(parsed))) if server.network is not None: addresses.add(IpAddress(server.network.ipv4.value)) return frozenset(addresses) diff --git a/sdk/tests/test_domains_apply.py b/sdk/tests/test_domains_apply.py index 91fa032..05dae79 100644 --- a/sdk/tests/test_domains_apply.py +++ b/sdk/tests/test_domains_apply.py @@ -298,6 +298,26 @@ def test_a_proxy_server_address_that_is_an_ip_is_taken_as_it_is( assert resolver.asked == ["crm.example.test", "www.crm.example.test"] +def test_a_non_canonical_ipv6_proxy_address_still_matches(tmp_path: Path) -> None: + """The schema takes ``2001:DB8::10``; a resolver answers ``2001:db8::10``.""" + project = _project(tmp_path) + server = project / "servers" / "h2.yaml" + document = yaml.safe_load(server.read_text(encoding="utf-8")) + document["spec"]["address"] = "2001:DB8:0::10" + server.write_text(yaml.safe_dump(document), encoding="utf-8") + resolver = _resolver( + crm_example_test="2001:db8::10", www_crm_example_test="2001:db8::10" + ) + + planned = preview_domain_apply( + _context(project, _Recorder(tmp_path / "unused")), + _request(project, issue_certificate=True), + resolver, + ) + + assert planned.dns_check is DnsCheck.VERIFIED + + def test_without_issue_certificate_dns_is_not_checked(tmp_path: Path) -> None: project = _project(tmp_path) resolver = _resolver(www_crm_example_test=ELSEWHERE)