"""Reference solution — the resource graph, the admission gate, the reachability prover. The phase's one question: **how do you make a topology claim you can prove?** "We configured a private endpoint" and "no traffic leaves the region" are different statements. The first is a configuration; the second is a property of the whole topology, and it is only true if *every* path is checked — including the one through a peering into a VNet that has its own internet egress. Three mechanisms: * a **resource graph** with plan/apply/drift, because the graph is the part that matters and dependency ordering is what makes a partial failure safe; * an **admission gate** that denies by default and names its rule; * a **reachability prover** that searches the topology and returns a **counter-example path** when the answer is yes and should not be. A control that says "denied" is an opinion. A control that says "denied, and here is the path" is a finding somebody can fix. Deterministic: no clock beyond an injected one, sorted traversal, derived identifiers. ``python solution.py`` runs the worked example. """ from __future__ import annotations import hashlib import ipaddress import json from collections import deque from dataclasses import dataclass, field, replace from enum import Enum from typing import (Any, Callable, Dict, FrozenSet, Iterable, List, Mapping, Optional, Sequence, Set, Tuple) # ====================================================================================== # 1. The resource graph # ====================================================================================== class GraphError(Exception): pass @dataclass(frozen=True) class Resource: """One declared resource. ``address`` is Terraform's ``type.name`` form. ``depends_on`` is *explicit* dependency. Real Terraform also infers dependencies from interpolated references, which is most of them — the explicit list is the escape hatch for ordering that the data flow does not express. """ address: str resource_type: str attributes: Mapping[str, Any] = field(default_factory=dict) depends_on: Tuple[str, ...] = () def digest(self) -> str: return hashlib.sha256(json.dumps( {"type": self.resource_type, "attributes": self.attributes}, sort_keys=True, separators=(",", ":"), default=str).encode()).hexdigest() class ResourceGraph: """The declared configuration as a DAG. Terraform's real contribution is not the language — it is that infrastructure is a **graph**, so ordering, parallelism and blast radius all fall out of one structure rather than being managed by hand in a script. """ def __init__(self) -> None: self._resources: Dict[str, Resource] = {} def add(self, resource: Resource) -> Resource: if resource.address in self._resources: raise GraphError(f"duplicate resource address: {resource.address}") self._resources[resource.address] = resource return resource def get(self, address: str) -> Resource: try: return self._resources[address] except KeyError: raise GraphError(f"unknown resource: {address}") from None def addresses(self) -> List[str]: return sorted(self._resources) def validate(self) -> None: """Every dependency must exist, and there must be no cycle. Both are checked at construction rather than during apply. A cycle discovered halfway through an apply leaves the estate in a state neither the config nor the previous state describes. """ for resource in self._resources.values(): for dependency in resource.depends_on: if dependency not in self._resources: raise GraphError( f"{resource.address} depends on unknown {dependency}") self.order() # raises on a cycle def order(self) -> List[str]: """A deterministic topological order: Kahn's algorithm with a sorted frontier. The sort is what makes it deterministic. Without it, two runs over the same graph produce different orders, and a plan that reorders between runs is a plan nobody can review. """ indegree = {a: 0 for a in self._resources} dependents: Dict[str, List[str]] = {a: [] for a in self._resources} for address, resource in self._resources.items(): for dependency in resource.depends_on: indegree[address] += 1 dependents[dependency].append(address) ready = sorted(a for a, d in indegree.items() if d == 0) out: List[str] = [] while ready: address = ready.pop(0) out.append(address) for dependent in sorted(dependents[address]): indegree[dependent] -= 1 if indegree[dependent] == 0: ready.append(dependent) ready.sort() if len(out) != len(self._resources): remaining = sorted(set(self._resources) - set(out)) raise GraphError(f"dependency cycle among: {remaining}") return out def transitive_dependents(self, address: str) -> List[str]: """Everything downstream — the blast radius of a change, and the set that must not be applied when this resource fails.""" self.get(address) dependents: Dict[str, List[str]] = {a: [] for a in self._resources} for a, resource in self._resources.items(): for dependency in resource.depends_on: dependents[dependency].append(a) seen: Set[str] = set() queue = deque(dependents[address]) while queue: current = queue.popleft() if current in seen: continue seen.add(current) queue.extend(dependents[current]) return sorted(seen) # ====================================================================================== # 2. State, plan and apply # ====================================================================================== class Action(str, Enum): NOOP = "noop" CREATE = "create" UPDATE = "update" DELETE = "delete" REPLACE = "replace" #: Attributes that cannot be changed in place. Changing one forces a **replace** — #: destroy then create — which for a stateful resource is data loss, and it is the single #: most dangerous thing a plan can silently contain. FORCE_NEW: Mapping[str, FrozenSet[str]] = { "azurerm_subnet": frozenset({"address_prefix", "virtual_network"}), "azurerm_kubernetes_cluster": frozenset({"location", "resource_group"}), "azurerm_storage_account": frozenset({"location", "account_kind"}), "azurerm_private_endpoint": frozenset({"subnet", "target"}), } @dataclass(frozen=True) class Change: address: str action: Action before: Optional[Mapping[str, Any]] after: Optional[Mapping[str, Any]] changed_attributes: Tuple[str, ...] = () force_new_because: Tuple[str, ...] = () @property def destructive(self) -> bool: return self.action in (Action.DELETE, Action.REPLACE) @dataclass(frozen=True) class Plan: changes: Tuple[Change, ...] @property def empty(self) -> bool: return not any(c.action is not Action.NOOP for c in self.changes) def by_action(self, action: Action) -> List[Change]: return [c for c in self.changes if c.action is action] @property def destructive_changes(self) -> List[Change]: return [c for c in self.changes if c.destructive] def summary(self) -> str: counts = {a: len(self.by_action(a)) for a in Action} return (f"{counts[Action.CREATE]} to add, {counts[Action.UPDATE]} to change, " f"{counts[Action.REPLACE]} to replace, {counts[Action.DELETE]} to " f"destroy") class State: """What we believe exists. Terraform's state file, in a dict. State is the source of most Terraform pain, and the reason is structural: it is a *third* thing, alongside the config and reality, and any two of them can disagree. """ def __init__(self) -> None: self._resources: Dict[str, Mapping[str, Any]] = {} def get(self, address: str) -> Optional[Mapping[str, Any]]: return self._resources.get(address) def set(self, address: str, attributes: Mapping[str, Any]) -> None: self._resources[address] = dict(attributes) def remove(self, address: str) -> None: self._resources.pop(address, None) def addresses(self) -> List[str]: return sorted(self._resources) def snapshot(self) -> Dict[str, Mapping[str, Any]]: return {a: dict(v) for a, v in self._resources.items()} def diff_attributes(before: Mapping[str, Any], after: Mapping[str, Any]) -> List[str]: return sorted({k for k in set(before) | set(after) if before.get(k) != after.get(k)}) def plan(graph: ResourceGraph, state: State) -> Plan: """Diff desired against believed-actual, in dependency order. Note what a plan is *not*: it is not a diff against reality. It is a diff against **state**, which is why drift detection (§3) is a separate operation — and why a plan can be empty while the estate is wrong. """ graph.validate() changes: List[Change] = [] for address in graph.order(): resource = graph.get(address) before = state.get(address) after = dict(resource.attributes) if before is None: changes.append(Change(address, Action.CREATE, None, after)) continue changed = diff_attributes(before, after) if not changed: changes.append(Change(address, Action.NOOP, before, after)) continue forcing = sorted(set(changed) & FORCE_NEW.get(resource.resource_type, frozenset())) action = Action.REPLACE if forcing else Action.UPDATE changes.append(Change(address, action, before, after, tuple(changed), tuple(forcing))) # Anything in state and no longer in the config is destroyed — in REVERSE dependency # order, so a dependent is removed before the thing it depends on. orphans = [a for a in state.addresses() if a not in set(graph.addresses())] for address in reversed(sorted(orphans)): changes.append(Change(address, Action.DELETE, state.get(address), None)) return Plan(tuple(changes)) @dataclass(frozen=True) class ApplyResult: applied: Tuple[str, ...] failed: Tuple[str, ...] skipped: Tuple[str, ...] errors: Mapping[str, str] @property def ok(self) -> bool: return not self.failed #: A provider takes (address, action, attributes) and either returns the realized #: attributes or raises. Injected so a failure is scriptable and the tests are #: deterministic. Provider = Callable[[str, Action, Optional[Mapping[str, Any]]], Mapping[str, Any]] def apply(plan_: Plan, graph: ResourceGraph, state: State, provider: Provider) -> ApplyResult: """Apply in dependency order; a failure **skips every dependent**. That skip is the whole safety property. Creating a private endpoint whose subnet failed to create produces a resource in an undefined state and a state file that disagrees with reality — which is much worse than stopping. Note that state is written **per resource, as it succeeds**. A crash mid-apply leaves a state file describing what actually got built, which is the only recoverable outcome. """ applied: List[str] = [] failed: List[str] = [] skipped: List[str] = [] errors: Dict[str, str] = {} blocked: Set[str] = set() by_address = {c.address: c for c in plan_.changes} for address in graph.order(): change = by_address.get(address) if change is None or change.action is Action.NOOP: continue resource = graph.get(address) if any(d in blocked or d in failed for d in resource.depends_on): skipped.append(address) blocked.add(address) continue try: realized = provider(address, change.action, change.after) except Exception as exc: # noqa: BLE001 - provider boundary failed.append(address) blocked.add(address) errors[address] = str(exc) continue state.set(address, realized) applied.append(address) for change in plan_.by_action(Action.DELETE): try: provider(change.address, Action.DELETE, None) except Exception as exc: # noqa: BLE001 failed.append(change.address) errors[change.address] = str(exc) continue state.remove(change.address) applied.append(change.address) return ApplyResult(tuple(applied), tuple(failed), tuple(skipped), errors) # ====================================================================================== # 3. Drift # ====================================================================================== @dataclass(frozen=True) class Drift: address: str attribute: str in_state: Any in_reality: Any def __str__(self) -> str: return (f"{self.address}.{self.attribute}: state={self.in_state!r} " f"actual={self.in_reality!r}") @dataclass(frozen=True) class DriftReport: drifted: Tuple[Drift, ...] missing: Tuple[str, ...] # in state, not in reality unmanaged: Tuple[str, ...] # in reality, not in state @property def clean(self) -> bool: return not (self.drifted or self.missing or self.unmanaged) def detect_drift(state: State, reality: Mapping[str, Mapping[str, Any]]) -> DriftReport: """Compare state against reality, **per attribute**. Three categories, and they call for three different responses: * **drifted** — somebody changed it out of band. Revert, or absorb it into the config; the choice is a judgment, and doing neither is how drift accumulates. * **missing** — state believes it exists and it does not. Usually a manual deletion; the next apply will recreate it, which may or may not be wanted. * **unmanaged** — it exists and nothing manages it. The most dangerous category, because it will be destroyed by nothing and audited by nobody. """ drifted: List[Drift] = [] missing: List[str] = [] for address in state.addresses(): believed = state.get(address) or {} actual = reality.get(address) if actual is None: missing.append(address) continue for attribute in diff_attributes(believed, actual): drifted.append(Drift(address, attribute, believed.get(attribute), actual.get(attribute))) unmanaged = sorted(set(reality) - set(state.addresses())) return DriftReport(tuple(drifted), tuple(sorted(missing)), tuple(unmanaged)) # ====================================================================================== # 4. The admission gate # ====================================================================================== class Effect(str, Enum): ALLOW = "allow" DENY = "deny" @dataclass(frozen=True) class AdmissionDecision: effect: Effect address: str rule: str reason: str @property def allowed(self) -> bool: return self.effect is Effect.ALLOW #: A rule reads a resource and returns a reason to deny, or None. AdmissionRule = Callable[[Resource], Optional[str]] @dataclass(frozen=True) class Policy: name: str applies_to: Tuple[str, ...] # resource types; empty = all check: AdmissionRule description: str = "" def matches(self, resource: Resource) -> bool: return not self.applies_to or resource.resource_type in self.applies_to def requires_private_endpoint(resource: Resource) -> Optional[str]: """Restricted data must not be reachable from the public network. Note the two separate conditions: a private endpoint present AND public access disabled. A private endpoint alone does not close the public path — the service keeps its public FQDN and its public listener, and traffic simply *may* use the private route. That distinction is the single most common misconfiguration in this area, and it is invisible in a diagram. """ if resource.attributes.get("data_classification") != "restricted": return None if not resource.attributes.get("private_endpoint"): return "restricted data with no private endpoint" if resource.attributes.get("public_network_access", "enabled") != "disabled": return ("restricted data with a private endpoint but public network access " "still enabled — the private endpoint does not close the public path") return None def forbids_public_ip(resource: Resource) -> Optional[str]: if resource.attributes.get("public_ip"): return "a public IP is not permitted on this resource type" return None def requires_signed_image(resource: Resource) -> Optional[str]: image = resource.attributes.get("image", "") if not resource.attributes.get("image_signature"): return f"image {image!r} is unsigned" if not resource.attributes.get("image_signer") in ALLOWED_SIGNERS: return (f"image {image!r} is signed by " f"{resource.attributes.get('image_signer')!r}, which is not trusted") return None def requires_digest_pin(resource: Resource) -> Optional[str]: """A tag is mutable. ``:latest`` today is not ``:latest`` tomorrow, and the signature you verified was for a different artifact.""" image = resource.attributes.get("image", "") if image and "@sha256:" not in image: return f"image {image!r} is not pinned to a digest" return None def requires_resource_limits(resource: Resource) -> Optional[str]: missing = [k for k in ("cpu_limit", "memory_limit") if not resource.attributes.get(k)] if missing: return f"missing {', '.join(missing)}; one workload can starve the node" return None def requires_owner_tag(resource: Resource) -> Optional[str]: tags = resource.attributes.get("tags", {}) if not tags.get("owner"): return "no owner tag; an unowned resource is an unowned cost and an unowned risk" return None def forbids_wildcard_egress(resource: Resource) -> Optional[str]: destinations = resource.attributes.get("egress_destinations", ()) if "*" in destinations or "0.0.0.0/0" in destinations: return "wildcard egress; the exfiltration channel is the network" return None ALLOWED_SIGNERS = frozenset({"platform-ci", "security-team"}) DEFAULT_POLICIES: Tuple[Policy, ...] = ( Policy("restricted-data-private-endpoint", ("azurerm_storage_account", "azurerm_cognitive_account", "azurerm_key_vault", "azurerm_search_service"), requires_private_endpoint, "restricted data is reachable only over a private endpoint"), Policy("no-public-ip", ("azurerm_kubernetes_cluster_node_pool", "azurerm_linux_virtual_machine"), forbids_public_ip, "no public IPs on compute"), Policy("signed-images", ("kubernetes_deployment",), requires_signed_image, "container images are signed by a trusted signer"), Policy("digest-pinned-images", ("kubernetes_deployment",), requires_digest_pin, "images are pinned to a digest, not a mutable tag"), Policy("resource-limits", ("kubernetes_deployment",), requires_resource_limits, "every workload declares CPU and memory limits"), Policy("owner-tag", (), requires_owner_tag, "every resource has a named owner"), Policy("no-wildcard-egress", ("azurerm_firewall_policy_rule_collection",), forbids_wildcard_egress, "egress is allow-listed"), ) class AdmissionGate: """Deny by default. Every deny names its rule. Deny-by-default here means: a resource **type** nobody has written a policy for is still evaluated against the type-agnostic rules, and a rule that raises is a deny. The second is the important one — a policy engine that fails open when its own code breaks is a policy engine that will fail open during an incident. """ def __init__(self, policies: Sequence[Policy] = DEFAULT_POLICIES) -> None: self.policies = tuple(policies) self.decisions: List[AdmissionDecision] = [] def evaluate(self, resource: Resource) -> List[AdmissionDecision]: out: List[AdmissionDecision] = [] for policy in self.policies: if not policy.matches(resource): continue try: reason = policy.check(resource) except Exception as exc: # noqa: BLE001 - fail closed out.append(AdmissionDecision( Effect.DENY, resource.address, policy.name, f"policy evaluation failed: {exc}")) continue if reason: out.append(AdmissionDecision(Effect.DENY, resource.address, policy.name, reason)) if not out: out.append(AdmissionDecision(Effect.ALLOW, resource.address, "-", "no policy objected")) self.decisions.extend(out) return out def admit(self, graph: ResourceGraph) -> Tuple[bool, List[AdmissionDecision]]: """Evaluate the whole graph. Returns (ok, every denial).""" denials: List[AdmissionDecision] = [] for address in graph.addresses(): denials.extend(d for d in self.evaluate(graph.get(address)) if not d.allowed) return not denials, denials # ====================================================================================== # 5. The network topology # ====================================================================================== @dataclass(frozen=True) class Subnet: name: str vnet: str cidr: str nat_gateway: bool = False route_table: Optional[str] = None @property def network(self) -> ipaddress.IPv4Network: return ipaddress.ip_network(self.cidr) @dataclass(frozen=True) class VNet: name: str region: str cidr: str @dataclass(frozen=True) class Peering: """Peerings are **directional** in Azure, and they are not transitive. Both facts cause real outages. A peering configured one way carries no traffic; and A↔B plus B↔C does *not* give A↔C — you need a hub with routing, or a third peering. ``allow_forwarded_traffic`` is what makes hub-and-spoke work, and it is off by default. """ source: str target: str allow_forwarded_traffic: bool = False class Direction(str, Enum): INBOUND = "inbound" OUTBOUND = "outbound" @dataclass(frozen=True) class NsgRule: """Priority-ordered, first-match-wins — like a firewall, unlike a policy engine. Which is a genuine trap for anyone arriving from [Phase 09](../../phase-09-control-plane-kya-zero-trust/index.md): there, deny beats allow regardless of order. Here, a low-priority allow at 100 beats a deny at 200, and the rule set's meaning depends entirely on the numbers. """ name: str priority: int direction: Direction action: Effect source: str # a CIDR, a subnet name, or "*" destination: str ports: Tuple[int, ...] = () # empty = any def matches(self, source: str, destination: str, port: int) -> bool: return (_endpoint_matches(self.source, source) and _endpoint_matches(self.destination, destination) and (not self.ports or port in self.ports)) def _endpoint_matches(pattern: str, value: str) -> bool: if pattern in ("*", "Any"): return True if pattern == value: return True if "/" in pattern: try: network = ipaddress.ip_network(pattern) except ValueError: return False try: return ipaddress.ip_address(value) in network except ValueError: return False return False @dataclass(frozen=True) class PrivateEndpoint: """A private endpoint is a NIC in your subnet with a private IP for a PaaS service. The half that breaks: **DNS**. Without a private DNS zone linked to the VNet, the service's FQDN still resolves to its public IP, so traffic takes the public path and everything appears to work — until an auditor asks, or until public access is disabled and it stops working with no obvious cause. """ name: str subnet: str target_service: str private_ip: str dns_zone_linked: bool = True @dataclass(frozen=True) class Service: """A PaaS service. ``public_network_access`` is the field that decides whether the private endpoint actually closed anything.""" name: str region: str public_network_access: bool = True data_classification: str = "internal" @dataclass(frozen=True) class Workload: name: str subnet: str identity: str = "" @dataclass(frozen=True) class EgressPolicy: """The firewall/NAT allow-list on the way out of a subnet.""" subnet: str allowed_fqdns: FrozenSet[str] = frozenset() allow_internet: bool = False class Topology: """VNets, subnets, peerings, NSGs, private endpoints, services, egress. Enough of a model to answer the question that matters: *can this workload reach that service, and does the traffic leave the region?* """ def __init__(self) -> None: self.vnets: Dict[str, VNet] = {} self.subnets: Dict[str, Subnet] = {} self.peerings: List[Peering] = [] self.nsg_rules: Dict[str, List[NsgRule]] = {} # subnet -> rules self.private_endpoints: Dict[str, PrivateEndpoint] = {} self.services: Dict[str, Service] = {} self.workloads: Dict[str, Workload] = {} self.egress: Dict[str, EgressPolicy] = {} # -- construction ---------------------------------------------------------------- def add_vnet(self, vnet: VNet) -> None: self.vnets[vnet.name] = vnet def add_subnet(self, subnet: Subnet) -> None: if subnet.vnet not in self.vnets: raise GraphError(f"subnet {subnet.name} references unknown vnet " f"{subnet.vnet}") self.subnets[subnet.name] = subnet def add_peering(self, peering: Peering) -> None: for name in (peering.source, peering.target): if name not in self.vnets: raise GraphError(f"peering references unknown vnet {name}") self.peerings.append(peering) def add_nsg_rule(self, rule: NsgRule, subnet: str) -> None: self.nsg_rules.setdefault(subnet, []).append(rule) def add_private_endpoint(self, endpoint: PrivateEndpoint) -> None: self.private_endpoints[endpoint.name] = endpoint def add_service(self, service: Service) -> None: self.services[service.name] = service def add_workload(self, workload: Workload) -> None: self.workloads[workload.name] = workload def set_egress(self, policy: EgressPolicy) -> None: self.egress[policy.subnet] = policy # -- queries --------------------------------------------------------------------- def peered(self, source_vnet: str) -> List[str]: """DIRECTLY peered targets only. Peering is directional and NOT transitive.""" return sorted({p.target for p in self.peerings if p.source == source_vnet}) def vnet_of(self, subnet: str) -> str: return self.subnets[subnet].vnet def region_of_subnet(self, subnet: str) -> str: return self.vnets[self.vnet_of(subnet)].region def nsg_verdict(self, subnet: str, direction: Direction, source: str, destination: str, port: int) -> Tuple[Effect, str]: """**First match by priority wins.** No default-deny fallback here beyond the implicit one — which is why the returned rule name matters when debugging.""" rules = sorted(self.nsg_rules.get(subnet, []), key=lambda r: r.priority) for rule in rules: if rule.direction is direction and rule.matches(source, destination, port): return rule.action, rule.name return Effect.DENY, "implicit-deny" # ====================================================================================== # 6. The reachability prover # ====================================================================================== @dataclass(frozen=True) class Hop: kind: str # "subnet" | "peering" | "private-endpoint" | "egress" | "service" name: str detail: str = "" def __str__(self) -> str: return f"{self.kind}:{self.name}" + (f" ({self.detail})" if self.detail else "") @dataclass(frozen=True) class Path: hops: Tuple[Hop, ...] def render(self) -> str: return " -> ".join(str(h) for h in self.hops) @property def leaves_region(self) -> bool: return any(h.kind == "egress" for h in self.hops) @dataclass(frozen=True) class ReachabilityResult: """A result is not a boolean. A control that says "denied" is an opinion. A control that says "denied, and here is the path" is a finding somebody can fix — and one that says "allowed, via this path, which leaves the region" is the counter-example that makes a residency claim falsifiable. """ reachable: bool paths: Tuple[Path, ...] blocked_by: Tuple[str, ...] = () @property def counter_example(self) -> Optional[Path]: return self.paths[0] if self.paths else None @property def any_path_leaves_region(self) -> bool: return any(p.leaves_region for p in self.paths) class ReachabilityProver: """Search the topology for every path from a workload to a service. The naive check — "is there a private endpoint in my VNet?" — misses two things, and both are how residency claims turn out to be false: * a **peering** into a VNet that has the endpoint (or has internet egress); * a **DNS** gap, where the private endpoint exists but the FQDN still resolves publicly, so traffic takes the public path regardless. Both are found by searching rather than by inspecting one hop. """ def __init__(self, topology: Topology, *, max_depth: int = 6) -> None: self.topology = topology self.max_depth = max_depth def reach(self, workload: str, service: str, *, port: int = 443) -> ReachabilityResult: topo = self.topology if workload not in topo.workloads: raise GraphError(f"unknown workload: {workload}") if service not in topo.services: raise GraphError(f"unknown service: {service}") start = topo.workloads[workload] target = topo.services[service] paths: List[Path] = [] blocked: List[str] = [] # BFS over (subnet, hops), visiting each subnet once. Visiting once is a # deliberate simplification: it finds A path per subnet rather than every path, # which is what a counter-example needs and is a great deal cheaper. queue: deque = deque([(start.subnet, (Hop("subnet", start.subnet),))]) seen: Set[str] = {start.subnet} while queue: subnet, hops = queue.popleft() if len(hops) > self.max_depth: continue # 1. Is there a private endpoint for the target in this subnet? for endpoint in sorted(topo.private_endpoints.values(), key=lambda e: e.name): if endpoint.subnet != subnet or endpoint.target_service != service: continue effect, rule = topo.nsg_verdict( subnet, Direction.OUTBOUND, subnet, endpoint.private_ip, port) if effect is Effect.DENY: blocked.append(f"NSG on {subnet} denies the private endpoint " f"({rule})") continue if not endpoint.dns_zone_linked: # The endpoint exists and DNS does not point at it. Traffic will use # the public path — which is exactly the misconfiguration that makes # a residency claim false while every diagram looks correct. blocked.append( f"private endpoint {endpoint.name} has no linked DNS zone; " f"the FQDN still resolves publicly") continue paths.append(Path(hops + ( Hop("private-endpoint", endpoint.name, endpoint.private_ip), Hop("service", service, target.region)))) # 2. Can we get out to the internet and reach it publicly? policy = topo.egress.get(subnet) if target.public_network_access and policy is not None: fqdn = f"{service}.public" if policy.allow_internet or fqdn in policy.allowed_fqdns or \ service in policy.allowed_fqdns: effect, rule = topo.nsg_verdict( subnet, Direction.OUTBOUND, subnet, "0.0.0.0/0", port) if effect is Effect.ALLOW: paths.append(Path(hops + ( Hop("egress", subnet, "internet" if policy.allow_internet else "allow-listed FQDN"), Hop("service", service, target.region)))) else: blocked.append(f"NSG on {subnet} denies egress ({rule})") elif not policy.allowed_fqdns: blocked.append(f"egress policy on {subnet} allows nothing") else: blocked.append(f"egress policy on {subnet} does not allow {service}") # 3. Other subnets in the SAME VNet. Azure routes these implicitly — there # is no peering and no route table involved — which is why "is the # private endpoint in my subnet?" is the wrong question. It only has to # be somewhere in the VNet. vnet = topo.vnet_of(subnet) for other in sorted(topo.subnets.values(), key=lambda s: s.name): if other.vnet != vnet or other.name == subnet or other.name in seen: continue effect, rule = topo.nsg_verdict( subnet, Direction.OUTBOUND, subnet, other.name, port) if effect is Effect.DENY: blocked.append(f"NSG on {subnet} denies {other.name} ({rule})") continue inbound, in_rule = topo.nsg_verdict( other.name, Direction.INBOUND, subnet, other.name, port) if inbound is Effect.DENY: blocked.append(f"NSG on {other.name} denies inbound from {subnet} " f"({in_rule})") continue seen.add(other.name) queue.append((other.name, hops + ( Hop("intra-vnet", vnet), Hop("subnet", other.name)))) # 4. Peerings — the hop a naive per-VNet check misses entirely. for peer in topo.peered(vnet): for other in sorted(topo.subnets.values(), key=lambda s: s.name): if other.vnet != peer or other.name in seen: continue effect, rule = topo.nsg_verdict( subnet, Direction.OUTBOUND, subnet, other.name, port) if effect is Effect.DENY: blocked.append(f"NSG on {subnet} denies {other.name} ({rule})") continue inbound, in_rule = topo.nsg_verdict( other.name, Direction.INBOUND, subnet, other.name, port) if inbound is Effect.DENY: blocked.append(f"NSG on {other.name} denies inbound from " f"{subnet} ({in_rule})") continue seen.add(other.name) queue.append((other.name, hops + ( Hop("peering", f"{vnet}->{peer}"), Hop("subnet", other.name)))) paths.sort(key=lambda p: (len(p.hops), p.render())) return ReachabilityResult(bool(paths), tuple(paths), tuple(sorted(set(blocked)))) def prove_no_egress(self, workload: str, service: str) -> ReachabilityResult: """The residency question: is there ANY path that leaves the region? Stated as a proof obligation rather than a configuration check, which is the point of the whole phase. """ result = self.reach(workload, service) offending = tuple(p for p in result.paths if p.leaves_region) return ReachabilityResult(bool(offending), offending, result.blocked_by) # ====================================================================================== # 7. GPU scheduling # ====================================================================================== @dataclass(frozen=True) class NodePool: """A GPU node pool. Every field here breaks a normal autoscaling assumption. ``startup_seconds`` is the one people underestimate: a GPU node is an image pull of several gigabytes plus driver initialization plus engine warm-up. Reactive autoscaling against a 6-minute start time is not autoscaling, it is a 6-minute outage with a graph. """ name: str gpu_type: str gpus_per_node: int node_count: int max_nodes: int startup_seconds: int = 360 taints: Tuple[str, ...] = () mig_profile: Optional[str] = None # e.g. "1g.10gb" @property def slices_per_gpu(self) -> int: """MIG partitions one physical GPU into isolated instances. The trade: more, smaller, *hardware-isolated* instances — good for many small models, useless for one large one, because a slice's memory is a hard ceiling and a 70B model does not fit in a 10 GB slice however many you have. """ if not self.mig_profile: return 1 return MIG_PROFILES.get(self.mig_profile, (1, 0))[0] @property def total_slices(self) -> int: return self.node_count * self.gpus_per_node * self.slices_per_gpu #: profile -> (instances per GPU, GiB per instance) for an 80 GiB A100/H100. MIG_PROFILES: Mapping[str, Tuple[int, int]] = { "1g.10gb": (7, 10), "2g.20gb": (3, 20), "3g.40gb": (2, 40), "7g.80gb": (1, 80), } @dataclass(frozen=True) class Workload3D: name: str gpus_required: int gpu_type: str gang: bool = False # all-or-nothing (tensor parallelism) toleration: Tuple[str, ...] = () @dataclass(frozen=True) class Placement: workload: str node_pool: Optional[str] nodes_used: int reason: str @property def placed(self) -> bool: return self.node_pool is not None def schedule(workloads: Sequence[Workload3D], pools: Sequence[NodePool]) -> List[Placement]: """Place workloads onto pools, honouring taints and **gang scheduling**. Gang scheduling is the property that breaks the normal model: a tensor-parallel deployment across 8 GPUs needs all 8 *simultaneously*, or it makes no progress at all. Kubernetes' default scheduler places pods one at a time and will happily place 6 of 8 and leave them idle — holding the GPUs — while waiting for two more that may never arrive. That is a deadlock, and it is why Volcano, Kueue and the like exist. """ remaining = {p.name: p.node_count * p.gpus_per_node for p in pools} by_name = {p.name: p for p in pools} placements: List[Placement] = [] # Largest first: a gang of 8 placed after four gangs of 2 may find no contiguous # capacity, even though the total was sufficient. for workload in sorted(workloads, key=lambda w: (-w.gpus_required, w.name)): candidates = [p for p in pools if p.gpu_type == workload.gpu_type] tolerated = [p for p in candidates if all(t in workload.toleration for t in p.taints)] if not candidates: placements.append(Placement(workload.name, None, 0, f"no pool with {workload.gpu_type}")) continue if not tolerated: placements.append(Placement( workload.name, None, 0, f"pools are tainted {candidates[0].taints} and the workload does not " f"tolerate them")) continue for pool in sorted(tolerated, key=lambda p: p.name): available = remaining[pool.name] if available < workload.gpus_required: continue if workload.gang: nodes_needed = -(-workload.gpus_required // pool.gpus_per_node) whole_nodes_free = available // pool.gpus_per_node if whole_nodes_free < nodes_needed: continue remaining[pool.name] -= nodes_needed * pool.gpus_per_node placements.append(Placement( workload.name, pool.name, nodes_needed, f"gang of {workload.gpus_required} across {nodes_needed} node(s)")) break remaining[pool.name] -= workload.gpus_required placements.append(Placement( workload.name, pool.name, 1, f"{workload.gpus_required} GPU(s)")) break else: placements.append(Placement( workload.name, None, 0, f"insufficient capacity for " f"{'a gang of ' if workload.gang else ''}" f"{workload.gpus_required}")) return sorted(placements, key=lambda p: p.workload) def scale_out_delay(pool: NodePool, *, engine_warmup_seconds: int = 120) -> int: """Total time from "we need a node" to "it is serving". Node start plus engine warm-up. This is the number that makes reactive autoscaling wrong for GPU inference and makes a **warm pool** the answer. """ return pool.startup_seconds + engine_warmup_seconds # ====================================================================================== # Worked example # ====================================================================================== def _provider(failing: FrozenSet[str] = frozenset()) -> Provider: def provider(address: str, action: Action, attributes: Optional[Mapping[str, Any]]) -> Mapping[str, Any]: if address in failing: raise RuntimeError(f"provider error creating {address}") return dict(attributes or {}) return provider def build_graph() -> ResourceGraph: graph = ResourceGraph() owner = {"tags": {"owner": "layla.almansouri"}} graph.add(Resource("azurerm_virtual_network.platform", "azurerm_virtual_network", {"cidr": "10.10.0.0/16", "region": "uaenorth", **owner})) graph.add(Resource("azurerm_subnet.aks", "azurerm_subnet", {"address_prefix": "10.10.1.0/24", "virtual_network": "platform", **owner}, depends_on=("azurerm_virtual_network.platform",))) graph.add(Resource("azurerm_subnet.endpoints", "azurerm_subnet", {"address_prefix": "10.10.2.0/24", "virtual_network": "platform", **owner}, depends_on=("azurerm_virtual_network.platform",))) graph.add(Resource("azurerm_kubernetes_cluster.aks", "azurerm_kubernetes_cluster", {"location": "uaenorth", "resource_group": "rg-ai-platform", "node_count": 3, **owner}, depends_on=("azurerm_subnet.aks",))) graph.add(Resource("azurerm_cognitive_account.openai", "azurerm_cognitive_account", {"location": "uaenorth", "data_classification": "restricted", "private_endpoint": True, "public_network_access": "disabled", **owner})) graph.add(Resource("azurerm_private_endpoint.openai", "azurerm_private_endpoint", {"subnet": "endpoints", "target": "openai", **owner}, depends_on=("azurerm_subnet.endpoints", "azurerm_cognitive_account.openai"))) graph.add(Resource("kubernetes_deployment.agent_kernel", "kubernetes_deployment", {"image": "registry.bank.ae/agent-kernel@sha256:ab12cd34", "image_signature": "sig-1", "image_signer": "platform-ci", "cpu_limit": "2", "memory_limit": "4Gi", **owner}, depends_on=("azurerm_kubernetes_cluster.aks",))) return graph def main() -> None: # pragma: no cover - narrative output print("=" * 78) print("1. THE RESOURCE GRAPH IS THE POINT") print("=" * 78) graph = build_graph() graph.validate() for i, address in enumerate(graph.order(), 1): deps = graph.get(address).depends_on print(f" {i}. {address:<44} after {list(deps) or '-'}") print(f" blast radius of the vnet: " f"{len(graph.transitive_dependents('azurerm_virtual_network.platform'))} " f"resources") cyclic = ResourceGraph() cyclic.add(Resource("a.x", "t", depends_on=("b.y",))) cyclic.add(Resource("b.y", "t", depends_on=("a.x",))) try: cyclic.validate() except GraphError as exc: print(f" a cycle is rejected at construction: {exc}") print(" -> ordering, parallelism and blast radius all fall out of ONE structure.") print() print("=" * 78) print("2. PLAN IS A DIFF AGAINST STATE — NOT AGAINST REALITY") print("=" * 78) state = State() first = plan(graph, state) print(f" first plan : {first.summary()}") result = apply(first, graph, state, _provider()) print(f" apply : {len(result.applied)} applied, ok={result.ok}") print(f" second plan: {plan(graph, state).summary()} empty={plan(graph, state).empty}") changed = build_graph() changed._resources["azurerm_kubernetes_cluster.aks"] = replace( changed.get("azurerm_kubernetes_cluster.aks"), attributes={**changed.get("azurerm_kubernetes_cluster.aks").attributes, "node_count": 6}) update = plan(changed, state) for change in update.changes: if change.action is not Action.NOOP: print(f" {change.action.value:<8} {change.address:<44} " f"{list(change.changed_attributes)}") replaced = build_graph() replaced._resources["azurerm_kubernetes_cluster.aks"] = replace( replaced.get("azurerm_kubernetes_cluster.aks"), attributes={**replaced.get("azurerm_kubernetes_cluster.aks").attributes, "location": "uaecentral"}) danger = plan(replaced, state) for change in danger.destructive_changes: print(f" {change.action.value.upper():<8} {change.address:<44} " f"forced by {list(change.force_new_because)}") print(" -> changing `location` DESTROYS and recreates the cluster. That line in a") print(" plan is the one nobody reads, and it is the one that matters.") print() print("=" * 78) print("3. A FAILURE SKIPS ITS DEPENDENTS") print("=" * 78) fresh = State() p = plan(graph, fresh) result = apply(p, graph, fresh, _provider(frozenset({"azurerm_subnet.aks"}))) print(f" applied: {list(result.applied)}") print(f" failed : {list(result.failed)}") print(f" skipped: {list(result.skipped)}") print(" -> the cluster was not attempted, because its subnet does not exist.") print(" Creating it anyway would produce a resource in an undefined state and") print(" a state file that disagrees with reality.") print() print("=" * 78) print("4. DRIFT") print("=" * 78) reality = state.snapshot() reality["azurerm_kubernetes_cluster.aks"] = { **reality["azurerm_kubernetes_cluster.aks"], "node_count": 9} reality["azurerm_storage_account.someones_test"] = {"location": "westeurope"} del reality["azurerm_private_endpoint.openai"] report = detect_drift(state, reality) for d in report.drifted: print(f" drifted : {d}") for m in report.missing: print(f" missing : {m} (in state, gone from reality)") for u in report.unmanaged: print(f" unmanaged: {u} (exists, managed by nothing)") print(" -> three categories, three different responses. 'unmanaged' is the") print(" dangerous one: destroyed by nothing, audited by nobody.") print() print("=" * 78) print("5. ADMISSION: DENY BY DEFAULT, AND NAME THE RULE") print("=" * 78) gate = AdmissionGate() ok, denials = gate.admit(graph) print(f" the good graph: admitted={ok}") bad = ResourceGraph() bad.add(Resource("azurerm_storage_account.docs", "azurerm_storage_account", {"data_classification": "restricted", "private_endpoint": False, "tags": {"owner": "layla"}})) bad.add(Resource("azurerm_key_vault.secrets", "azurerm_key_vault", {"data_classification": "restricted", "private_endpoint": True, "public_network_access": "enabled", "tags": {"owner": "layla"}})) bad.add(Resource("kubernetes_deployment.tool", "kubernetes_deployment", {"image": "docker.io/somebody/tool:latest", "tags": {"owner": "omar"}})) bad.add(Resource("azurerm_linux_virtual_machine.jump", "azurerm_linux_virtual_machine", {"public_ip": "20.1.2.3"})) ok, denials = AdmissionGate().admit(bad) for d in denials: print(f" DENY {d.address:<44} [{d.rule}]") print(f" {d.reason}") print(" -> the key vault is the interesting one: it HAS a private endpoint and is") print(" still denied, because public access is enabled. A private endpoint does") print(" not close the public path, and that distinction is invisible in a") print(" diagram.") print() print("=" * 78) print("6. REACHABILITY — WITH A COUNTER-EXAMPLE") print("=" * 78) topo = Topology() topo.add_vnet(VNet("platform", "uaenorth", "10.10.0.0/16")) topo.add_vnet(VNet("shared", "uaenorth", "10.20.0.0/16")) topo.add_vnet(VNet("legacy", "westeurope", "10.30.0.0/16")) topo.add_subnet(Subnet("aks", "platform", "10.10.1.0/24")) topo.add_subnet(Subnet("endpoints", "platform", "10.10.2.0/24")) topo.add_subnet(Subnet("shared-svc", "shared", "10.20.1.0/24")) topo.add_subnet(Subnet("legacy-app", "legacy", "10.30.1.0/24", nat_gateway=True)) topo.add_peering(Peering("platform", "shared")) topo.add_peering(Peering("shared", "legacy", allow_forwarded_traffic=True)) topo.add_nsg_rule(NsgRule("allow-all-out", 100, Direction.OUTBOUND, Effect.ALLOW, "*", "*"), "aks") topo.add_nsg_rule(NsgRule("allow-all-in", 100, Direction.INBOUND, Effect.ALLOW, "*", "*"), "endpoints") topo.add_nsg_rule(NsgRule("allow-all-out", 100, Direction.OUTBOUND, Effect.ALLOW, "*", "*"), "endpoints") topo.add_nsg_rule(NsgRule("allow-all-in", 100, Direction.INBOUND, Effect.ALLOW, "*", "*"), "shared-svc") topo.add_nsg_rule(NsgRule("allow-all-out", 100, Direction.OUTBOUND, Effect.ALLOW, "*", "*"), "shared-svc") topo.add_nsg_rule(NsgRule("allow-all-in", 100, Direction.INBOUND, Effect.ALLOW, "*", "*"), "legacy-app") topo.add_nsg_rule(NsgRule("allow-all-out", 100, Direction.OUTBOUND, Effect.ALLOW, "*", "*"), "legacy-app") topo.add_service(Service("openai", "uaenorth", public_network_access=False, data_classification="restricted")) topo.add_private_endpoint(PrivateEndpoint("pe-openai", "endpoints", "openai", "10.10.2.4")) topo.add_workload(Workload("agent-kernel", "aks")) topo.set_egress(EgressPolicy("aks", frozenset())) topo.set_egress(EgressPolicy("legacy-app", allow_internet=True)) prover = ReachabilityProver(topo) result = prover.reach("agent-kernel", "openai") print(f" agent-kernel -> openai: reachable={result.reachable}") print(f" path: {result.counter_example.render()}") print(" note the private endpoint is in 'endpoints', NOT in the workload's own") print(" subnet. Azure routes intra-VNet implicitly, so 'is the PE in my subnet?'") print(" is the wrong question — it only has to be somewhere in the VNet.") print() blocked_topo = Topology() for v in topo.vnets.values(): blocked_topo.add_vnet(v) for s in topo.subnets.values(): blocked_topo.add_subnet(s) blocked_topo.add_service(topo.services["openai"]) blocked_topo.add_private_endpoint(topo.private_endpoints["pe-openai"]) blocked_topo.add_workload(Workload("agent-kernel", "aks")) blocked_topo.add_nsg_rule(NsgRule("deny-endpoints", 100, Direction.OUTBOUND, Effect.DENY, "*", "endpoints"), "aks") blocked_topo.add_nsg_rule(NsgRule("allow-rest", 200, Direction.OUTBOUND, Effect.ALLOW, "*", "*"), "aks") denied = ReachabilityProver(blocked_topo).reach("agent-kernel", "openai") print(f" with an NSG deny at priority 100: reachable={denied.reachable}") for reason in denied.blocked_by: print(f" {reason}") print(" NSG rules are FIRST-MATCH-BY-PRIORITY, unlike the policy engine in") print(" Phase 09 where deny beats allow regardless of order. Swap the priorities") print(" here and the deny never fires.") print() print(" now the residency question. A public model endpoint, and a peering chain") print(" into a VNet in ANOTHER REGION with open internet egress:") topo.add_service(Service("openai-global", "westeurope", public_network_access=True)) topo.set_egress(EgressPolicy("shared-svc", allow_internet=False, allowed_fqdns=frozenset())) residency = prover.prove_no_egress("agent-kernel", "openai-global") print(f" any path leaves the region: {residency.reachable}") if residency.counter_example: print(f" COUNTER-EXAMPLE: {residency.counter_example.render()}") print(" -> two peering hops and a NAT gateway somebody configured in 2021. A") print(" per-VNet check says 'no internet egress from aks' and is correct and") print(" useless. THIS is the difference between a configuration and a control.") print() print(" the DNS gap — the misconfiguration that looks entirely correct:") topo.add_service(Service("search", "uaenorth", public_network_access=True)) topo.add_private_endpoint(PrivateEndpoint("pe-search", "aks", "search", "10.10.1.10", dns_zone_linked=False)) result = prover.reach("agent-kernel", "search") for reason in result.blocked_by: if "DNS" in reason: print(f" {reason}") print(" -> the endpoint exists, the diagram is right, and every packet takes") print(" the public path. Nothing errors.") print() print("=" * 78) print("7. GPUs BREAK EVERY AUTOSCALING ASSUMPTION") print("=" * 78) pools = [ NodePool("gpu-a100", "A100-80", gpus_per_node=8, node_count=2, max_nodes=4, startup_seconds=420, taints=("nvidia.com/gpu",)), NodePool("gpu-mig", "A100-80", gpus_per_node=8, node_count=1, max_nodes=2, startup_seconds=420, taints=("nvidia.com/gpu",), mig_profile="1g.10gb"), ] print(f" {'pool':<10} {'nodes':<6} {'GPUs':<6} {'MIG':<10} {'slices':<8} startup") for pool in pools: print(f" {pool.name:<10} {pool.node_count:<6} " f"{pool.node_count * pool.gpus_per_node:<6} " f"{pool.mig_profile or '-':<10} {pool.total_slices:<8} " f"{scale_out_delay(pool)}s to serving") minutes = scale_out_delay(pools[0]) // 60 print(f" -> {minutes} minutes from 'we need a node' to 'it is serving' — a " f"{pools[0].startup_seconds}s node") print(" start plus 120s of engine warm-up. Reactive autoscaling against that is") print(f" a {minutes}-minute outage with a graph, which is why the answer is a " f"warm pool.") print() workloads = [ Workload3D("llama-70b-tp8", 8, "A100-80", gang=True, toleration=("nvidia.com/gpu",)), Workload3D("llama-70b-tp8-b", 8, "A100-80", gang=True, toleration=("nvidia.com/gpu",)), Workload3D("embedder", 1, "A100-80", toleration=("nvidia.com/gpu",)), Workload3D("reranker", 2, "A100-80", toleration=("nvidia.com/gpu",)), Workload3D("untolerating", 1, "A100-80"), Workload3D("wrong-gpu", 1, "H100-80", toleration=("nvidia.com/gpu",)), ] for placement in schedule(workloads, pools): where = placement.node_pool or "UNPLACED" print(f" {placement.workload:<18} {where:<10} {placement.reason}") print(" -> the gangs are placed FIRST, deliberately. Kubernetes' default scheduler") print(" places pods one at a time: it will happily give 6 of 8 GPUs to a") print(" tensor-parallel deployment, which then makes no progress while HOLDING") print(" them. That is a deadlock, and it is why Volcano and Kueue exist.") if __name__ == "__main__": # pragma: no cover main()