"""Non-secret infrastructure capability projection and change impact.""" from __future__ import annotations from dataclasses import asdict, dataclass from datetime import UTC, datetime from typing import Mapping from urllib.parse import urlsplit from .model import InstallationSpec CAPABILITY_DOCUMENT_SCHEMA_VERSION = 1 CAPABILITY_STATES = frozenset( { "configured", "available_unconfigured", "externally_supplied", "unavailable", } ) DEPENDENCY_INVENTORY_SCHEMA_VERSION = 1 DEPENDENCY_STATES = frozenset( {"active", "inactive", "data_present", "pending_work", "runtime_binding"} ) @dataclass(frozen=True, slots=True) class InfrastructureCapability: id: str label: str state: str source: str detail: str endpoint: Mapping[str, object] secret_refs: tuple[str, ...] dependent_modules: tuple[str, ...] def to_dict(self) -> dict[str, object]: value = asdict(self) value["endpoint"] = dict(self.endpoint) value["secret_refs"] = list(self.secret_refs) value["dependent_modules"] = list(self.dependent_modules) return value @dataclass(frozen=True, slots=True) class CapabilityChangeImpact: capability_id: str action: str previous_state: str desired_state: str previous_source: str desired_source: str dependent_modules: tuple[str, ...] detail: str required_action: str actual_dependencies: tuple["CapabilityDependency", ...] = () inventory_inspected: bool = False def to_dict(self) -> dict[str, object]: value = asdict(self) value["dependent_modules"] = list(self.dependent_modules) value["actual_dependencies"] = [ item.to_dict() for item in self.actual_dependencies ] return value @dataclass(frozen=True, slots=True) class CapabilityDependency: capability_id: str module_id: str dependency_type: str dependency_ref: str state: str scope: str summary: str metrics: Mapping[str, int] required_action: str def to_dict(self) -> dict[str, object]: return { "capability_id": self.capability_id, "module_id": self.module_id, "dependency_type": self.dependency_type, "dependency_ref": self.dependency_ref, "state": self.state, "scope": self.scope, "summary": self.summary, "metrics": dict(sorted(self.metrics.items())), "required_action": self.required_action, } @dataclass(frozen=True, slots=True) class InfrastructureDependencyInventory: installation_id: str generated_at: datetime complete: bool inspected_capability_ids: tuple[str, ...] provider_count: int dependencies: tuple[CapabilityDependency, ...] def dependencies_for( self, capability_id: str, ) -> tuple[CapabilityDependency, ...]: return tuple( item for item in self.dependencies if item.capability_id == capability_id ) def infrastructure_dependency_inventory_from_mapping( value: object, ) -> InfrastructureDependencyInventory: if ( not isinstance(value, Mapping) or value.get("schema_version") != DEPENDENCY_INVENTORY_SCHEMA_VERSION ): raise ValueError("Infrastructure dependency inventory schema is unsupported.") installation_id = _inventory_text(value, "installation_id", maximum=100) generated_at_text = _inventory_text(value, "generated_at", maximum=100) try: generated_at = datetime.fromisoformat(generated_at_text.replace("Z", "+00:00")) except ValueError as exc: raise ValueError( "Infrastructure dependency inventory timestamp is invalid." ) from exc if generated_at.tzinfo is None: raise ValueError("Infrastructure dependency inventory timestamp needs a timezone.") generated_at = generated_at.astimezone(UTC) complete = value.get("complete") if type(complete) is not bool: raise ValueError("Infrastructure dependency inventory completion state is invalid.") inspected = _inventory_string_list( value.get("inspected_capability_ids"), maximum_items=100, maximum_length=120, ) if len(inspected) != len(set(inspected)): raise ValueError("Infrastructure dependency inventory repeats a capability id.") providers = value.get("providers") if not isinstance(providers, list) or len(providers) > 100: raise ValueError("Infrastructure dependency provider reports are invalid.") provider_states: list[str] = [] provider_declarations: dict[str, tuple[str, ...]] = {} provider_counts: dict[str, int] = {} for provider in providers: if not isinstance(provider, Mapping): raise ValueError("Infrastructure dependency provider report is invalid.") module_id = _inventory_text(provider, "module_id", maximum=120) if module_id in provider_declarations: raise ValueError("Infrastructure dependency provider is repeated.") state = _inventory_text(provider, "state", maximum=40) if state not in {"complete", "error"}: raise ValueError("Infrastructure dependency provider state is invalid.") provider_states.append(state) count = provider.get("dependency_count") if type(count) is not int or count < 0: raise ValueError("Infrastructure dependency provider count is invalid.") capability_ids = _inventory_string_list( provider.get("capability_ids"), maximum_items=30, maximum_length=120, ) if len(capability_ids) != len(set(capability_ids)): raise ValueError("Infrastructure dependency provider capability is repeated.") provider_declarations[module_id] = capability_ids provider_counts[module_id] = count if complete and any(state != "complete" for state in provider_states): raise ValueError("Complete dependency inventory contains a failed provider.") raw_dependencies = value.get("dependencies") if not isinstance(raw_dependencies, list) or len(raw_dependencies) > 10_000: raise ValueError("Infrastructure dependency records are invalid.") dependencies = tuple(_inventory_dependency(item) for item in raw_dependencies) if any( capability_id not in inspected for capability_ids in provider_declarations.values() for capability_id in capability_ids ): raise ValueError( "Infrastructure dependency provider was not covered by the inspection." ) if any(item.capability_id not in inspected for item in dependencies): raise ValueError("Dependency record was not covered by the inventory inspection.") identities = { (item.capability_id, item.module_id, item.dependency_type, item.dependency_ref) for item in dependencies } if len(identities) != len(dependencies): raise ValueError("Infrastructure dependency inventory repeats a record.") observed_counts = {module_id: 0 for module_id in provider_counts} for dependency in dependencies: declarations = provider_declarations.get(dependency.module_id) if declarations is None or dependency.capability_id not in declarations: raise ValueError( "Infrastructure dependency is outside its provider declaration." ) observed_counts[dependency.module_id] += 1 if observed_counts != provider_counts: raise ValueError("Infrastructure dependency provider count does not match records.") return InfrastructureDependencyInventory( installation_id=installation_id, generated_at=generated_at, complete=complete, inspected_capability_ids=inspected, provider_count=len(providers), dependencies=dependencies, ) def infrastructure_capability_document( spec: InstallationSpec, environment: Mapping[str, str], ) -> dict[str, object]: """Project installer choices without copying credentials or secret URLs.""" capabilities = tuple( sorted( ( _postgres_capability(spec, environment), _redis_capability(spec, environment), _mail_capability(spec), _storage_capability(spec, environment), _load_balancer_capability(spec), _ingress_capability(spec), ), key=lambda item: item.id, ) ) tasks = _post_install_tasks(spec, capabilities) return { "schema_version": CAPABILITY_DOCUMENT_SCHEMA_VERSION, "installation_id": spec.installation_id, "profile": spec.profile, "capabilities": [item.to_dict() for item in capabilities], "post_install_tasks": tasks, } def capability_change_impacts( previous_document: object, desired_document: Mapping[str, object], *, dependency_inventory: InfrastructureDependencyInventory | None = None, ) -> tuple[CapabilityChangeImpact, ...]: previous = _capability_map(previous_document) desired = _capability_map(desired_document) if not previous: return () impacts: list[CapabilityChangeImpact] = [] for capability_id in sorted(set(previous) | set(desired)): before = previous.get(capability_id) after = desired.get(capability_id) if before is None or after is None: continue previous_state = str(before.get("state") or "unavailable") desired_state = str(after.get("state") or "unavailable") previous_source = str(before.get("source") or "unknown") desired_source = str(after.get("source") or "unknown") previous_endpoint = _endpoint_signature(before.get("endpoint")) desired_endpoint = _endpoint_signature(after.get("endpoint")) previous_secret_refs = tuple( sorted(_string_items(before.get("secret_refs"))) ) desired_secret_refs = tuple( sorted(_string_items(after.get("secret_refs"))) ) if ( previous_state == desired_state and previous_source == desired_source and previous_endpoint == desired_endpoint and previous_secret_refs == desired_secret_refs ): continue action = ( "remove" if previous_state != "unavailable" and desired_state == "unavailable" else "replace" if previous_source != desired_source else "reconfigure" ) dependents = tuple( sorted( { *(_string_items(before.get("dependent_modules"))), *(_string_items(after.get("dependent_modules"))), } ) ) dependent_label = ", ".join(dependents) or "no declared module consumers" binding_change = _binding_change_label( previous_endpoint, desired_endpoint, previous_secret_refs, desired_secret_refs, ) actual_dependencies = ( dependency_inventory.dependencies_for(capability_id) if dependency_inventory is not None else () ) inventory_inspected = bool( dependency_inventory is not None and capability_id in dependency_inventory.inspected_capability_ids ) if inventory_inspected and actual_dependencies: references = ", ".join( f"{item.module_id}:{item.dependency_ref}" for item in actual_dependencies ) inventory_detail = ( f" Provider inventory reports {len(actual_dependencies)} persisted " f"dependency record(s): {references}." ) elif inventory_inspected: inventory_detail = ( " Provider inventory reports no persisted module-owned dependencies." ) else: inventory_detail = " Provider inventory did not inspect this capability." dependency_actions = tuple( dict.fromkeys( item.required_action for item in actual_dependencies if item.required_action.strip() ) ) required_action = ( "Review module-owned configuration and data migration or recovery " "evidence before apply." ) if dependency_actions: required_action = f"{required_action} {' '.join(dependency_actions)}" impacts.append( CapabilityChangeImpact( capability_id=capability_id, action=action, previous_state=previous_state, desired_state=desired_state, previous_source=previous_source, desired_source=desired_source, dependent_modules=dependents, detail=( f"{capability_id} changes from {previous_state}/{previous_source} " f"to {desired_state}/{desired_source}{binding_change}; " f"declared consumers: {dependent_label}.{inventory_detail}" ), required_action=required_action, actual_dependencies=actual_dependencies, inventory_inspected=inventory_inspected, ) ) return tuple(impacts) def _postgres_capability( spec: InstallationSpec, environment: Mapping[str, str], ) -> InfrastructureCapability: managed = spec.components.postgres.mode == "managed" endpoint = ( {"scheme": "postgresql", "host": "postgres", "port": 5432} if managed else _redacted_endpoint(environment.get("DATABASE_URL", ""), default_port=5432) ) return InfrastructureCapability( id="database.postgresql", label="PostgreSQL database", state="configured" if managed else "externally_supplied", source="installer-managed" if managed else "operator-supplied", detail=( "The installer manages the database service." if managed else "The deployment binds an externally operated PostgreSQL service." ), endpoint=endpoint, secret_refs=("env:POSTGRES_PASSWORD",) if managed else ("env:DATABASE_URL",), dependent_modules=("core", *tuple(sorted(spec.enabled_modules))), ) def _redis_capability( spec: InstallationSpec, environment: Mapping[str, str], ) -> InfrastructureCapability: mode = spec.components.redis.mode consumers = _enabled_consumers( spec, { "campaigns", "dataflow", "files", "mail", "notifications", "scheduling", "workflow_engine", }, include_core=True, ) if mode == "disabled": return InfrastructureCapability( id="coordination.redis", label="Redis coordination and queues", state="unavailable", source="disabled", detail="Distributed queues and coordination are disabled.", endpoint={}, secret_refs=(), dependent_modules=consumers, ) managed = mode == "managed" endpoint = ( {"scheme": "redis", "host": "redis", "port": 6379} if managed else _redacted_endpoint(environment.get("REDIS_URL", ""), default_port=6379) ) return InfrastructureCapability( id="coordination.redis", label="Redis coordination and queues", state="configured" if managed else "externally_supplied", source="installer-managed" if managed else "operator-supplied", detail=( "The installer manages the Redis service." if managed else "The deployment binds an externally operated Redis service." ), endpoint=endpoint, secret_refs=("env:REDIS_PASSWORD",) if managed else ("env:REDIS_URL",), dependent_modules=consumers, ) def _mail_capability(spec: InstallationSpec) -> InfrastructureCapability: mode = spec.components.mail.mode consumers = _enabled_consumers( spec, {"campaigns", "mail", "notifications"}, ) if mode == "disabled": return InfrastructureCapability( id="mail.smtp", label="SMTP delivery", state="unavailable", source="disabled", detail="No SMTP infrastructure was selected.", endpoint={}, secret_refs=(), dependent_modules=consumers, ) if mode == "test-mail": return InfrastructureCapability( id="mail.smtp", label="SMTP delivery", state="available_unconfigured", source="installer-managed-test", detail="GreenMail is reachable, but Mail still owns profile and credential configuration.", endpoint={"scheme": "smtp", "host": "test-mail", "port": 3025}, secret_refs=(), dependent_modules=consumers, ) return InfrastructureCapability( id="mail.smtp", label="SMTP delivery", state="available_unconfigured", source="operator-supplied", detail="An external relay was selected; Mail still needs a reviewed server and credential binding.", endpoint={}, secret_refs=(), dependent_modules=consumers, ) def _storage_capability( spec: InstallationSpec, environment: Mapping[str, str], ) -> InfrastructureCapability: mode = spec.components.storage.mode consumers = _enabled_consumers( spec, {"campaigns", "files", "records", "templates"}, ) if mode == "local": return InfrastructureCapability( id="files.storage", label="Managed file content storage", state="configured", source="host-local", detail="Files use the installer-managed local persistent volume.", endpoint={"kind": "filesystem", "reference": "volume:files-data"}, secret_refs=(), dependent_modules=consumers, ) if mode == "garage": return InfrastructureCapability( id="files.storage", label="Managed file content storage", state="configured", source="installer-managed-garage", detail="Files use the installer-managed single-node Garage service.", endpoint={"scheme": "http", "host": "garage", "port": 3900}, secret_refs=( "env:FILE_STORAGE_S3_ACCESS_KEY_ID", "env:FILE_STORAGE_S3_SECRET_ACCESS_KEY", "env:GARAGE_RPC_SECRET", ), dependent_modules=consumers, ) return InfrastructureCapability( id="files.storage", label="Managed file content storage", state="externally_supplied", source="operator-supplied-s3", detail="Files use an externally operated S3-compatible service.", endpoint=_redacted_endpoint( environment.get("FILE_STORAGE_S3_ENDPOINT_URL", ""), default_port=443, ), secret_refs=( "env:FILE_STORAGE_S3_ACCESS_KEY_ID", "env:FILE_STORAGE_S3_SECRET_ACCESS_KEY", ), dependent_modules=consumers, ) def _load_balancer_capability(spec: InstallationSpec) -> InfrastructureCapability: return InfrastructureCapability( id="runtime.load_balancing", label="Application load balancing", state="configured", source="installer-managed", detail=( f"HAProxy balances {spec.replicas.web} WebUI and {spec.replicas.api} API replica(s)." ), endpoint={"scheme": "http", "host": "load-balancer", "port": 8080}, secret_refs=(), dependent_modules=("core", "ops"), ) def _ingress_capability(spec: InstallationSpec) -> InfrastructureCapability: mode = spec.ingress.mode endpoint = _redacted_endpoint(spec.public_url, default_port=443) if mode == "unconfigured": state = "unavailable" source = "unconfigured" detail = "No supported public ingress boundary is configured." elif mode == "existing-proxy": state = "externally_supplied" source = "operator-supplied-proxy" detail = "An externally operated reverse proxy provides public ingress." else: state = "configured" source = "installer-managed" if mode == "managed" else "host-local" detail = "The installer has a bounded public ingress configuration." return InfrastructureCapability( id="network.ingress", label="Public HTTP ingress", state=state, source=source, detail=detail, endpoint=endpoint, secret_refs=(), dependent_modules=("core", "ops"), ) def _post_install_tasks( spec: InstallationSpec, capabilities: tuple[InfrastructureCapability, ...], ) -> list[dict[str, object]]: by_id = {item.id: item for item in capabilities} tasks: list[dict[str, object]] = [] mail = by_id["mail.smtp"] if mail.state == "available_unconfigured" and "mail" in spec.enabled_modules: tasks.append( { "id": "mail.smtp-profile", "resume_key": f"{spec.installation_id}:mail.smtp-profile:v1", "capability_id": mail.id, "state": "pending", "owner_module": "mail", "summary": "Create or select a Mail SMTP server and credential envelope.", "required_inputs": [ "server endpoint", "transport security policy", "credential envelope reference when authentication is required", ], "secret_boundary": "credential-envelope-reference-only", } ) ingress = by_id["network.ingress"] if ingress.state == "unavailable": tasks.append( { "id": "network.configure-ingress", "resume_key": f"{spec.installation_id}:network.configure-ingress:v1", "capability_id": ingress.id, "state": "pending", "owner_module": "ops", "summary": "Select managed ingress or bind an existing reverse proxy.", "required_inputs": ["public URL", "TLS and proxy trust boundary"], "secret_boundary": "no-secret-material", } ) return tasks def _enabled_consumers( spec: InstallationSpec, candidates: set[str], *, include_core: bool = False, ) -> tuple[str, ...]: consumers = candidates.intersection(spec.enabled_modules) if include_core: consumers.add("core") return tuple(sorted(consumers)) def _redacted_endpoint(value: str, *, default_port: int) -> dict[str, object]: try: parsed = urlsplit(value) host = parsed.hostname port = parsed.port or default_port except ValueError: return {"reference": "unresolved"} if not parsed.scheme or not host: return {"reference": "unresolved"} return {"scheme": parsed.scheme, "host": host, "port": port} def _capability_map(value: object) -> dict[str, Mapping[str, object]]: if not isinstance(value, Mapping): return {} raw_items = value.get("capabilities") if not isinstance(raw_items, list): return {} result: dict[str, Mapping[str, object]] = {} for item in raw_items: if not isinstance(item, Mapping): continue capability_id = str(item.get("id") or "").strip() state = str(item.get("state") or "").strip() if capability_id and state in CAPABILITY_STATES: result[capability_id] = item return result def _string_items(value: object) -> tuple[str, ...]: if not isinstance(value, list): return () return tuple(str(item).strip() for item in value if str(item).strip()) def _endpoint_signature(value: object) -> tuple[tuple[str, str], ...]: if not isinstance(value, Mapping): return () return tuple( sorted( (str(key), str(raw)) for key, raw in value.items() if isinstance(key, str) and isinstance(raw, (str, int, bool)) ) ) def _binding_change_label( previous_endpoint: tuple[tuple[str, str], ...], desired_endpoint: tuple[tuple[str, str], ...], previous_secret_refs: tuple[str, ...], desired_secret_refs: tuple[str, ...], ) -> str: changes: list[str] = [] if previous_endpoint != desired_endpoint: changes.append("endpoint binding") if previous_secret_refs != desired_secret_refs: changes.append("secret-reference binding") return f" with changed {' and '.join(changes)}" if changes else "" def _inventory_text( value: Mapping[str, object], key: str, *, maximum: int, ) -> str: raw = value.get(key) if not isinstance(raw, str): raise ValueError(f"Infrastructure dependency inventory {key} is invalid.") result = raw.strip() if not result or len(result) > maximum or any(ord(char) < 32 for char in result): raise ValueError(f"Infrastructure dependency inventory {key} is invalid.") return result def _inventory_string_list( value: object, *, maximum_items: int, maximum_length: int, ) -> tuple[str, ...]: if not isinstance(value, list) or len(value) > maximum_items: raise ValueError("Infrastructure dependency inventory list is invalid.") items: list[str] = [] for raw in value: if not isinstance(raw, str): raise ValueError("Infrastructure dependency inventory list is invalid.") item = raw.strip() if ( not item or len(item) > maximum_length or any(ord(char) < 32 for char in item) ): raise ValueError("Infrastructure dependency inventory list is invalid.") items.append(item) return tuple(items) def _inventory_dependency(value: object) -> CapabilityDependency: if not isinstance(value, Mapping): raise ValueError("Infrastructure dependency record is invalid.") state = _inventory_text(value, "state", maximum=40) if state not in DEPENDENCY_STATES: raise ValueError("Infrastructure dependency state is invalid.") raw_metrics = value.get("metrics") if not isinstance(raw_metrics, Mapping) or len(raw_metrics) > 20: raise ValueError("Infrastructure dependency metrics are invalid.") metrics: dict[str, int] = {} for raw_key, raw_count in raw_metrics.items(): if not isinstance(raw_key, str): raise ValueError("Infrastructure dependency metric name is invalid.") key = raw_key.strip() if not key or len(key) > 80 or any(ord(char) < 32 for char in key): raise ValueError("Infrastructure dependency metric name is invalid.") if type(raw_count) is not int or raw_count < 0: raise ValueError("Infrastructure dependency metric value is invalid.") metrics[key] = raw_count return CapabilityDependency( capability_id=_inventory_text(value, "capability_id", maximum=120), module_id=_inventory_text(value, "module_id", maximum=120), dependency_type=_inventory_text(value, "dependency_type", maximum=120), dependency_ref=_inventory_text(value, "dependency_ref", maximum=240), state=state, scope=_inventory_text(value, "scope", maximum=120), summary=_inventory_text(value, "summary", maximum=1000), metrics=metrics, required_action=_inventory_text(value, "required_action", maximum=1000), )