563 lines
15 KiB
Python
563 lines
15 KiB
Python
from collections import defaultdict
|
|
from collections.abc import Iterable, Mapping, Sequence
|
|
from dataclasses import replace
|
|
from typing import Protocol
|
|
|
|
from .adapters import (
|
|
CERTIFICATES,
|
|
PROXY_HOSTS,
|
|
Capability,
|
|
CapabilityOperation,
|
|
DNS_RECORDS,
|
|
Endpoint,
|
|
EndpointPlan,
|
|
NormalizedObject,
|
|
ObjectReference,
|
|
Observation,
|
|
PlanAction,
|
|
PlanOperation,
|
|
)
|
|
from .dependencies import order_plan_actions
|
|
|
|
|
|
ObservationGroupKey = tuple[str, str]
|
|
|
|
|
|
class ReconciliationConflict(RuntimeError):
|
|
"""Raised when observations cannot be reconciled safely."""
|
|
|
|
|
|
class PlanningConflict(RuntimeError):
|
|
"""Raised when an endpoint cannot be planned safely."""
|
|
|
|
|
|
class PlanningAdapter(Protocol):
|
|
endpoint: Endpoint
|
|
capabilities: frozenset[Capability]
|
|
|
|
|
|
class ReconciliationPolicy(Protocol):
|
|
object_type: str
|
|
|
|
def reconcile(
|
|
self,
|
|
logical_key: str,
|
|
observations: Sequence[Observation],
|
|
) -> NormalizedObject: ...
|
|
|
|
def equivalent(
|
|
self,
|
|
current: NormalizedObject,
|
|
desired: NormalizedObject,
|
|
) -> bool: ...
|
|
|
|
|
|
def _observation_sort_key(
|
|
observation: Observation,
|
|
) -> tuple[str, str, str]:
|
|
return (
|
|
observation.endpoint_id,
|
|
observation.object.local_key or "",
|
|
observation.observed_at.isoformat(),
|
|
)
|
|
|
|
|
|
def logicalize_observation_references(
|
|
observations: Iterable[Observation],
|
|
) -> tuple[Observation, ...]:
|
|
rows = tuple(observations)
|
|
|
|
local_targets: dict[
|
|
tuple[str, str, str],
|
|
str,
|
|
] = {}
|
|
|
|
for observation in rows:
|
|
object = observation.object
|
|
|
|
if object.local_key is None:
|
|
continue
|
|
|
|
identity = (
|
|
observation.endpoint_id,
|
|
object.object_type,
|
|
object.local_key,
|
|
)
|
|
|
|
existing = local_targets.get(identity)
|
|
|
|
if (
|
|
existing is not None
|
|
and existing != object.logical_key
|
|
):
|
|
raise ReconciliationConflict(
|
|
"Conflicting endpoint-local object identity for "
|
|
f"{observation.endpoint_id} "
|
|
f"{object.object_type} "
|
|
f"local key {object.local_key!r}"
|
|
)
|
|
|
|
local_targets[identity] = object.logical_key
|
|
|
|
logicalized: list[Observation] = []
|
|
|
|
for observation in rows:
|
|
object = observation.object
|
|
|
|
references = []
|
|
|
|
for reference in object.references:
|
|
if reference.logical_key is not None:
|
|
references.append(reference)
|
|
continue
|
|
|
|
target = (
|
|
observation.endpoint_id,
|
|
reference.object_type,
|
|
reference.local_key,
|
|
)
|
|
logical_key = local_targets.get(target)
|
|
|
|
if logical_key is None:
|
|
raise ReconciliationConflict(
|
|
"Unresolved endpoint-local reference from "
|
|
f"{observation.endpoint_id} "
|
|
f"{object.object_type} "
|
|
f"{object.logical_key!r}: "
|
|
f"{reference.relation} -> "
|
|
f"{reference.object_type} "
|
|
f"local key {reference.local_key!r}"
|
|
)
|
|
|
|
references.append(
|
|
ObjectReference(
|
|
relation=reference.relation,
|
|
object_type=reference.object_type,
|
|
logical_key=logical_key,
|
|
)
|
|
)
|
|
|
|
logicalized.append(
|
|
replace(
|
|
observation,
|
|
object=replace(
|
|
object,
|
|
references=tuple(references),
|
|
),
|
|
)
|
|
)
|
|
|
|
return tuple(logicalized)
|
|
|
|
|
|
def group_observations(
|
|
observations: Iterable[Observation],
|
|
) -> dict[ObservationGroupKey, tuple[Observation, ...]]:
|
|
grouped: dict[
|
|
ObservationGroupKey,
|
|
list[Observation],
|
|
] = defaultdict(list)
|
|
|
|
for observation in observations:
|
|
normalized = observation.object
|
|
if not observation.endpoint_id:
|
|
raise RuntimeError("Observation endpoint_id cannot be empty")
|
|
if not normalized.object_type:
|
|
raise RuntimeError("Observation object_type cannot be empty")
|
|
if not normalized.logical_key:
|
|
raise RuntimeError("Observation logical_key cannot be empty")
|
|
|
|
key = (
|
|
normalized.object_type,
|
|
normalized.logical_key,
|
|
)
|
|
grouped[key].append(observation)
|
|
|
|
return {
|
|
key: tuple(sorted(rows, key=_observation_sort_key))
|
|
for key, rows in sorted(grouped.items())
|
|
}
|
|
|
|
|
|
class DnsUnionPolicy:
|
|
object_type = DNS_RECORDS
|
|
|
|
def reconcile(
|
|
self,
|
|
logical_key: str,
|
|
observations: Sequence[Observation],
|
|
) -> NormalizedObject:
|
|
if not observations:
|
|
raise RuntimeError(
|
|
"DNS union policy requires at least one observation"
|
|
)
|
|
|
|
first = observations[0].object
|
|
conflicting_endpoints = {
|
|
observation.endpoint_id
|
|
for observation in observations
|
|
if observation.object.payload != first.payload
|
|
}
|
|
if conflicting_endpoints:
|
|
conflicting_endpoints.add(observations[0].endpoint_id)
|
|
endpoint_list = ", ".join(
|
|
sorted(conflicting_endpoints)
|
|
)
|
|
raise ReconciliationConflict(
|
|
f"Conflicting {DNS_RECORDS} observations for "
|
|
f"{logical_key!r} from {endpoint_list}"
|
|
)
|
|
|
|
return NormalizedObject(
|
|
object_type=DNS_RECORDS,
|
|
logical_key=logical_key,
|
|
payload=dict(first.payload),
|
|
)
|
|
|
|
def equivalent(
|
|
self,
|
|
current: NormalizedObject,
|
|
desired: NormalizedObject,
|
|
) -> bool:
|
|
return (
|
|
current.object_type == desired.object_type
|
|
and current.logical_key == desired.logical_key
|
|
and current.payload == desired.payload
|
|
and current.references == desired.references
|
|
)
|
|
|
|
|
|
class CertificateNewestPolicy:
|
|
object_type = CERTIFICATES
|
|
|
|
_PORTABLE_FIELDS = (
|
|
"domain_names",
|
|
"valid_from",
|
|
"expires_on",
|
|
)
|
|
|
|
@classmethod
|
|
def _portable_state(
|
|
cls,
|
|
object: NormalizedObject,
|
|
) -> tuple[object, ...]:
|
|
return tuple(
|
|
object.payload.get(field)
|
|
for field in cls._PORTABLE_FIELDS
|
|
)
|
|
|
|
@staticmethod
|
|
def _expiry_key(
|
|
object: NormalizedObject,
|
|
) -> tuple[int, str]:
|
|
expires_on = object.payload.get("expires_on")
|
|
|
|
if expires_on is None:
|
|
return (0, "")
|
|
|
|
if not isinstance(expires_on, str):
|
|
raise ReconciliationConflict(
|
|
"Certificate expires_on must be normalized text"
|
|
)
|
|
|
|
return (1, expires_on)
|
|
|
|
def reconcile(
|
|
self,
|
|
logical_key: str,
|
|
observations: Sequence[Observation],
|
|
) -> NormalizedObject:
|
|
if not observations:
|
|
raise RuntimeError(
|
|
"Certificate policy requires at least one observation"
|
|
)
|
|
|
|
newest_key = max(
|
|
self._expiry_key(observation.object)
|
|
for observation in observations
|
|
)
|
|
|
|
newest = [
|
|
observation
|
|
for observation in observations
|
|
if self._expiry_key(observation.object) == newest_key
|
|
]
|
|
|
|
baseline = self._portable_state(newest[0].object)
|
|
|
|
conflicting_endpoints = {
|
|
observation.endpoint_id
|
|
for observation in newest
|
|
if self._portable_state(observation.object) != baseline
|
|
}
|
|
|
|
if conflicting_endpoints:
|
|
conflicting_endpoints.add(newest[0].endpoint_id)
|
|
|
|
raise ReconciliationConflict(
|
|
f"Conflicting {CERTIFICATES} observations for "
|
|
f"{logical_key!r} at newest expiry from "
|
|
+ ", ".join(sorted(conflicting_endpoints))
|
|
)
|
|
|
|
winner = newest[0].object
|
|
|
|
return NormalizedObject(
|
|
object_type=CERTIFICATES,
|
|
logical_key=logical_key,
|
|
payload=dict(winner.payload),
|
|
)
|
|
|
|
def equivalent(
|
|
self,
|
|
current: NormalizedObject,
|
|
desired: NormalizedObject,
|
|
) -> bool:
|
|
return (
|
|
current.object_type == CERTIFICATES
|
|
and desired.object_type == CERTIFICATES
|
|
and current.logical_key == desired.logical_key
|
|
and self._portable_state(current)
|
|
== self._portable_state(desired)
|
|
)
|
|
|
|
|
|
class ProxyHostConsensusPolicy:
|
|
object_type = PROXY_HOSTS
|
|
|
|
def reconcile(
|
|
self,
|
|
logical_key: str,
|
|
observations: Sequence[Observation],
|
|
) -> NormalizedObject:
|
|
if not observations:
|
|
raise RuntimeError(
|
|
"Proxy-host policy requires at least one observation"
|
|
)
|
|
|
|
first = observations[0].object
|
|
|
|
conflicting_endpoints = {
|
|
observation.endpoint_id
|
|
for observation in observations
|
|
if (
|
|
observation.object.payload != first.payload
|
|
or observation.object.references != first.references
|
|
)
|
|
}
|
|
|
|
if conflicting_endpoints:
|
|
conflicting_endpoints.add(observations[0].endpoint_id)
|
|
|
|
raise ReconciliationConflict(
|
|
f"Conflicting {PROXY_HOSTS} observations for "
|
|
f"{logical_key!r} from "
|
|
+ ", ".join(sorted(conflicting_endpoints))
|
|
)
|
|
|
|
for reference in first.references:
|
|
if reference.local_key is not None:
|
|
raise ReconciliationConflict(
|
|
"Proxy-host desired state cannot contain "
|
|
"endpoint-local references"
|
|
)
|
|
|
|
return NormalizedObject(
|
|
object_type=PROXY_HOSTS,
|
|
logical_key=logical_key,
|
|
payload=dict(first.payload),
|
|
references=first.references,
|
|
)
|
|
|
|
def equivalent(
|
|
self,
|
|
current: NormalizedObject,
|
|
desired: NormalizedObject,
|
|
) -> bool:
|
|
return (
|
|
current.object_type == PROXY_HOSTS
|
|
and desired.object_type == PROXY_HOSTS
|
|
and current.logical_key == desired.logical_key
|
|
and current.payload == desired.payload
|
|
and current.references == desired.references
|
|
)
|
|
|
|
|
|
DEFAULT_POLICIES: Mapping[str, ReconciliationPolicy] = {
|
|
DNS_RECORDS: DnsUnionPolicy(),
|
|
CERTIFICATES: CertificateNewestPolicy(),
|
|
PROXY_HOSTS: ProxyHostConsensusPolicy(),
|
|
}
|
|
|
|
|
|
def reconcile_observations(
|
|
observations: Iterable[Observation],
|
|
policies: Mapping[str, ReconciliationPolicy] | None = None,
|
|
) -> list[NormalizedObject]:
|
|
selected_policies = (
|
|
DEFAULT_POLICIES
|
|
if policies is None
|
|
else policies
|
|
)
|
|
desired: list[NormalizedObject] = []
|
|
|
|
logicalized = logicalize_observation_references(
|
|
observations
|
|
)
|
|
|
|
for (object_type, logical_key), rows in (
|
|
group_observations(logicalized).items()
|
|
):
|
|
policy = selected_policies.get(object_type)
|
|
if policy is None:
|
|
raise RuntimeError(
|
|
f"No reconciliation policy for {object_type!r}"
|
|
)
|
|
if policy.object_type != object_type:
|
|
raise RuntimeError(
|
|
f"Policy for {object_type!r} advertises "
|
|
f"{policy.object_type!r}"
|
|
)
|
|
|
|
desired.append(policy.reconcile(logical_key, rows))
|
|
|
|
return desired
|
|
|
|
|
|
def build_endpoint_plans(
|
|
adapters: Iterable[PlanningAdapter],
|
|
desired_objects: Iterable[NormalizedObject],
|
|
observations: Iterable[Observation],
|
|
policies: Mapping[str, ReconciliationPolicy] | None = None,
|
|
) -> tuple[EndpointPlan, ...]:
|
|
desired_by_identity: dict[
|
|
tuple[str, str],
|
|
NormalizedObject,
|
|
] = {}
|
|
|
|
for desired in desired_objects:
|
|
if desired.local_key is not None:
|
|
raise RuntimeError(
|
|
"Desired objects in plans cannot contain "
|
|
"an endpoint-local key"
|
|
)
|
|
|
|
identity = (
|
|
desired.object_type,
|
|
desired.logical_key,
|
|
)
|
|
if identity in desired_by_identity:
|
|
raise RuntimeError(
|
|
"Duplicate desired object identity: "
|
|
f"{desired.object_type} {desired.logical_key}"
|
|
)
|
|
desired_by_identity[identity] = desired
|
|
|
|
selected_policies = (
|
|
DEFAULT_POLICIES
|
|
if policies is None
|
|
else policies
|
|
)
|
|
|
|
current_by_identity: dict[
|
|
tuple[str, str, str],
|
|
list[Observation],
|
|
] = defaultdict(list)
|
|
|
|
logicalized_observations = (
|
|
logicalize_observation_references(observations)
|
|
)
|
|
|
|
for observation in logicalized_observations:
|
|
current_by_identity[
|
|
(
|
|
observation.endpoint_id,
|
|
observation.object.object_type,
|
|
observation.object.logical_key,
|
|
)
|
|
].append(observation)
|
|
|
|
adapters_by_endpoint: dict[str, PlanningAdapter] = {}
|
|
for adapter in adapters:
|
|
endpoint_id = adapter.endpoint.identity
|
|
if endpoint_id in adapters_by_endpoint:
|
|
raise RuntimeError(
|
|
f"Duplicate adapter endpoint: {endpoint_id}"
|
|
)
|
|
adapters_by_endpoint[endpoint_id] = adapter
|
|
|
|
plans: list[EndpointPlan] = []
|
|
for endpoint_id, adapter in sorted(
|
|
adapters_by_endpoint.items()
|
|
):
|
|
apply_types = {
|
|
capability.object_type
|
|
for capability in adapter.capabilities
|
|
if capability.operation == CapabilityOperation.APPLY
|
|
}
|
|
if not apply_types:
|
|
continue
|
|
|
|
actions: list[PlanAction] = []
|
|
for identity, desired in sorted(
|
|
desired_by_identity.items()
|
|
):
|
|
object_type, logical_key = identity
|
|
if object_type not in apply_types:
|
|
continue
|
|
|
|
current_rows = current_by_identity.get(
|
|
(endpoint_id, object_type, logical_key),
|
|
[],
|
|
)
|
|
if len(current_rows) > 1:
|
|
raise PlanningConflict(
|
|
f"Multiple endpoint objects for {endpoint_id} "
|
|
f"{object_type} {logical_key!r}"
|
|
)
|
|
|
|
current = (
|
|
current_rows[0].object
|
|
if current_rows
|
|
else None
|
|
)
|
|
if current is None:
|
|
operation = PlanOperation.CREATE
|
|
else:
|
|
policy = selected_policies.get(object_type)
|
|
|
|
if policy is None:
|
|
equivalent = (
|
|
current.payload == desired.payload
|
|
and current.references == desired.references
|
|
)
|
|
else:
|
|
equivalent = policy.equivalent(
|
|
current,
|
|
desired,
|
|
)
|
|
|
|
operation = (
|
|
PlanOperation.NOOP
|
|
if equivalent
|
|
else PlanOperation.UPDATE
|
|
)
|
|
|
|
actions.append(
|
|
PlanAction(
|
|
endpoint_id=endpoint_id,
|
|
operation=operation,
|
|
desired=desired,
|
|
current=current,
|
|
)
|
|
)
|
|
|
|
plans.append(
|
|
EndpointPlan(
|
|
endpoint_id=endpoint_id,
|
|
actions=order_plan_actions(actions),
|
|
)
|
|
)
|
|
|
|
return tuple(plans)
|