490 lines
16 KiB
Python
490 lines
16 KiB
Python
"""Non-secret infrastructure capability projection and change impact."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass
|
|
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",
|
|
}
|
|
)
|
|
|
|
|
|
@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
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
value = asdict(self)
|
|
value["dependent_modules"] = list(self.dependent_modules)
|
|
return value
|
|
|
|
|
|
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],
|
|
) -> 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,
|
|
)
|
|
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}."
|
|
),
|
|
required_action=(
|
|
"Review module-owned configuration and data migration or recovery evidence before apply."
|
|
),
|
|
)
|
|
)
|
|
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 ""
|