From ad7a672a488b31d3fc23db10ccac349af49d588e Mon Sep 17 00:00:00 2001 From: aidankhogg Date: Tue, 30 Jun 2026 16:43:02 +0100 Subject: [PATCH] Add resolver policy derivation --- netengine/spec/__init__.py | 4 ++ netengine/spec/authority.py | 60 ++++++++++++++++++++ tests/test_authority_spec.py | 104 ++++++++++++++++++++++++++++++++++- 3 files changed, 167 insertions(+), 1 deletion(-) diff --git a/netengine/spec/__init__.py b/netengine/spec/__init__.py index 3234d3d..3fa4710 100644 --- a/netengine/spec/__init__.py +++ b/netengine/spec/__init__.py @@ -6,7 +6,9 @@ AuthorityScope, AuthoritySource, BoundaryPolicy, + ResolverPolicy, default_authorities_for_spec, + resolver_policy_from_boundary, ) __all__ = [ @@ -15,5 +17,7 @@ "AuthorityScope", "AuthoritySource", "BoundaryPolicy", + "ResolverPolicy", "default_authorities_for_spec", + "resolver_policy_from_boundary", ] diff --git a/netengine/spec/authority.py b/netengine/spec/authority.py index 2818f79..aa99adb 100644 --- a/netengine/spec/authority.py +++ b/netengine/spec/authority.py @@ -61,6 +61,17 @@ class Authority(SpecModel): source: AuthoritySource = Field(default=AuthoritySource.LOCAL) +class ResolverPolicy(SpecModel): + """Derived resolver capabilities for a boundary authority posture.""" + + local_root: bool = Field(default=True) + upstream: bool = Field(default=False) + mirror_table: bool = Field(default=False) + peer_suffix_delegation: bool = Field(default=False) + imported_trust_bundle: bool = Field(default=False) + notes: list[str] = Field(default_factory=list) + + class BoundaryPolicy(SpecModel): """Authority-layer policy for traffic and trust at the world boundary.""" @@ -105,6 +116,55 @@ def validate_boundary_policy(self) -> "BoundaryPolicy": raise ValueError("isolated mode should not enable upstream resolution") return self + + +def resolver_policy_from_boundary(policy: BoundaryPolicy) -> ResolverPolicy: + """Derive resolver behavior from a boundary policy. + + The result intentionally models resolver capabilities rather than concrete + DNS server ordering. Notes capture the intended precedence for modes where + ordering matters. + """ + + notes: list[str] = [] + upstream = False + mirror_table = False + + if policy.real_internet == GatewayRealInternetMode.ISOLATED: + notes.append("isolated: local root only") + elif policy.real_internet == GatewayRealInternetMode.SHADOWED: + upstream = True + notes.append("shadowed: local root first, upstream second") + elif policy.real_internet == GatewayRealInternetMode.MIRRORED: + mirror_table = True + notes.append("mirrored: mirror table first, local root second; upstream disabled") + else: + upstream = policy.upstream_resolver_enabled + notes.append( + f"{policy.real_internet.value}: no minimum resolver semantics beyond configured upstream" + ) + + peer_suffix_delegation = policy.cross_world in { + GatewayCrossWorldMode.PEERED, + GatewayCrossWorldMode.FEDERATED, + } + imported_trust_bundle = policy.cross_world == GatewayCrossWorldMode.FEDERATED + + if policy.cross_world == GatewayCrossWorldMode.PEERED: + notes.append("peered: peer suffix delegation without shared trust") + elif policy.cross_world == GatewayCrossWorldMode.FEDERATED: + notes.append("federated: peer suffix delegation with imported trust bundle") + + return ResolverPolicy( + local_root=True, + upstream=upstream, + mirror_table=mirror_table, + peer_suffix_delegation=peer_suffix_delegation, + imported_trust_bundle=imported_trust_bundle, + notes=notes, + ) + + def default_authorities_for_spec(spec: "NetEngineSpec") -> list[Authority]: """Return stable default authorities for the control surfaces in ``spec``. diff --git a/tests/test_authority_spec.py b/tests/test_authority_spec.py index 9eae4a2..00875c3 100644 --- a/tests/test_authority_spec.py +++ b/tests/test_authority_spec.py @@ -3,7 +3,15 @@ import pytest from pydantic import ValidationError -from netengine.spec import Authority, AuthorityKind, AuthorityScope, AuthoritySource, BoundaryPolicy +from netengine.spec import ( + Authority, + AuthorityKind, + AuthorityScope, + AuthoritySource, + BoundaryPolicy, + ResolverPolicy, + resolver_policy_from_boundary, +) from netengine.spec.models import CrossWorldPeer, ServiceMirror from netengine.spec.types import GatewayCrossWorldMode, GatewayRealInternetMode @@ -105,6 +113,100 @@ def test_boundary_policy_federated_accepts_peer_trust_bundle() -> None: def test_boundary_policy_isolated_disallows_upstream_resolution() -> None: with pytest.raises(ValidationError, match="isolated mode should not enable upstream"): BoundaryPolicy(upstream_resolver_enabled=True, upstream_resolver_ip="8.8.8.8") + + +def test_resolver_policy_defaults_to_local_root_only() -> None: + policy = ResolverPolicy() + + assert policy.local_root is True + assert policy.upstream is False + assert policy.mirror_table is False + assert policy.peer_suffix_delegation is False + assert policy.imported_trust_bundle is False + assert policy.notes == [] + + +def test_resolver_policy_from_isolated_boundary_uses_local_root_only() -> None: + resolver_policy = resolver_policy_from_boundary(BoundaryPolicy()) + + assert resolver_policy.local_root is True + assert resolver_policy.upstream is False + assert resolver_policy.mirror_table is False + assert resolver_policy.peer_suffix_delegation is False + assert resolver_policy.imported_trust_bundle is False + assert resolver_policy.notes == ["isolated: local root only"] + + +def test_resolver_policy_from_shadowed_boundary_enables_upstream() -> None: + resolver_policy = resolver_policy_from_boundary( + BoundaryPolicy(real_internet=GatewayRealInternetMode.SHADOWED) + ) + + assert resolver_policy.local_root is True + assert resolver_policy.upstream is True + assert resolver_policy.mirror_table is False + assert resolver_policy.notes == ["shadowed: local root first, upstream second"] + + +def test_resolver_policy_from_mirrored_boundary_uses_mirror_table_without_upstream() -> None: + resolver_policy = resolver_policy_from_boundary( + BoundaryPolicy( + real_internet=GatewayRealInternetMode.MIRRORED, + service_mirrors=[ + ServiceMirror(real_hostname="api.example.com", in_world_service="10.1.2.3") + ], + upstream_resolver_enabled=True, + ) + ) + + assert resolver_policy.local_root is True + assert resolver_policy.upstream is False + assert resolver_policy.mirror_table is True + assert resolver_policy.notes == [ + "mirrored: mirror table first, local root second; upstream disabled" + ] + + +def test_resolver_policy_from_peered_boundary_delegates_peer_suffixes_without_trust() -> None: + resolver_policy = resolver_policy_from_boundary( + BoundaryPolicy( + cross_world=GatewayCrossWorldMode.PEERED, + peers=[CrossWorldPeer(name="world-b", endpoint="10.99.0.1:8443")], + ) + ) + + assert resolver_policy.local_root is True + assert resolver_policy.peer_suffix_delegation is True + assert resolver_policy.imported_trust_bundle is False + assert resolver_policy.notes == [ + "isolated: local root only", + "peered: peer suffix delegation without shared trust", + ] + + +def test_resolver_policy_from_federated_boundary_imports_trust_bundle() -> None: + resolver_policy = resolver_policy_from_boundary( + BoundaryPolicy( + cross_world=GatewayCrossWorldMode.FEDERATED, + peers=[ + CrossWorldPeer( + name="world-b", + endpoint="10.99.0.1:8443", + trust_bundle="spiffe://world-b.example/bundle", + ) + ], + ) + ) + + assert resolver_policy.local_root is True + assert resolver_policy.peer_suffix_delegation is True + assert resolver_policy.imported_trust_bundle is True + assert resolver_policy.notes == [ + "isolated: local root only", + "federated: peer suffix delegation with imported trust bundle", + ] + + def test_default_authorities_for_spec_maps_spec_sections(minimal_spec) -> None: from netengine.spec import default_authorities_for_spec