From e422fae81f89a898e93180e8398cd7892153265f Mon Sep 17 00:00:00 2001 From: aidankhogg Date: Tue, 30 Jun 2026 16:35:29 +0100 Subject: [PATCH] Add authority boundary policy model --- netengine/spec/__init__.py | 9 +++++- netengine/spec/authority.py | 51 +++++++++++++++++++++++++++-- netengine/spec/models.py | 1 + tests/test_authority_spec.py | 62 +++++++++++++++++++++++++++++++++++- 4 files changed, 119 insertions(+), 4 deletions(-) diff --git a/netengine/spec/__init__.py b/netengine/spec/__init__.py index 812d590..5c96c41 100644 --- a/netengine/spec/__init__.py +++ b/netengine/spec/__init__.py @@ -1,10 +1,17 @@ """Spec parsing and models for NetEngine declarative specifications.""" -from netengine.spec.authority import Authority, AuthorityKind, AuthorityScope, AuthoritySource +from netengine.spec.authority import ( + Authority, + AuthorityKind, + AuthorityScope, + AuthoritySource, + BoundaryPolicy, +) __all__ = [ "Authority", "AuthorityKind", "AuthorityScope", "AuthoritySource", + "BoundaryPolicy", ] diff --git a/netengine/spec/authority.py b/netengine/spec/authority.py index d76dcbf..ea5bcc5 100644 --- a/netengine/spec/authority.py +++ b/netengine/spec/authority.py @@ -1,9 +1,10 @@ """Foundational authority primitives for NetEngine specifications.""" from enum import Enum -from pydantic import Field +from pydantic import Field, model_validator -from netengine.spec.models import SpecModel +from netengine.spec.models import CrossWorldPeer, ServiceMirror, SpecModel +from netengine.spec.types import GatewayCrossWorldMode, GatewayRealInternetMode class AuthorityKind(str, Enum): @@ -53,3 +54,49 @@ class Authority(SpecModel): controls: list[str] = Field(...) description: str | None = Field(default=None) source: AuthoritySource = Field(default=AuthoritySource.LOCAL) + + +class BoundaryPolicy(SpecModel): + """Authority-layer policy for traffic and trust at the world boundary.""" + + real_internet: GatewayRealInternetMode = Field(default=GatewayRealInternetMode.ISOLATED) + cross_world: GatewayCrossWorldMode = Field(default=GatewayCrossWorldMode.NONE) + service_mirrors: list[ServiceMirror] = Field(default_factory=list) + upstream_resolver_enabled: bool = Field(default=False) + upstream_resolver_ip: str | None = Field(default=None) + peers: list[CrossWorldPeer] = Field(default_factory=list) + + @model_validator(mode="after") + def validate_boundary_policy(self) -> "BoundaryPolicy": + """Validate boundary posture combinations and required trust metadata.""" + if self.real_internet == GatewayRealInternetMode.MIRRORED and not self.service_mirrors: + raise ValueError("mirrored real-internet posture requires at least one service mirror") + + if self.real_internet != GatewayRealInternetMode.MIRRORED and self.service_mirrors: + raise ValueError("service mirrors require real_internet to be 'mirrored'") + + if ( + self.cross_world in {GatewayCrossWorldMode.PEERED, GatewayCrossWorldMode.FEDERATED} + and not self.peers + ): + raise ValueError("cross-world peered or federated mode requires peers") + + if self.cross_world == GatewayCrossWorldMode.FEDERATED: + peers_missing_trust = [ + peer.name + for peer in self.peers + if not (peer.trust_bundle or peer.trust_anchor_cert) + ] + if peers_missing_trust: + missing = ", ".join(peers_missing_trust) + raise ValueError( + "federated peers must provide sufficient trust metadata through " + f"a trust bundle or peer trust anchor: {missing}" + ) + + if self.real_internet == GatewayRealInternetMode.ISOLATED and ( + self.upstream_resolver_enabled or self.upstream_resolver_ip is not None + ): + raise ValueError("isolated mode should not enable upstream resolution") + + return self diff --git a/netengine/spec/models.py b/netengine/spec/models.py index 06123b7..66dfc40 100644 --- a/netengine/spec/models.py +++ b/netengine/spec/models.py @@ -615,6 +615,7 @@ class CrossWorldPeer(SpecModel): name: str = Field(...) endpoint: str = Field(...) mode: GatewayCrossWorldMode = Field(default=GatewayCrossWorldMode.PEERED) + trust_bundle: Optional[str] = None trust_anchor_cert: Optional[str] = None diff --git a/tests/test_authority_spec.py b/tests/test_authority_spec.py index 63ef75e..d4b891d 100644 --- a/tests/test_authority_spec.py +++ b/tests/test_authority_spec.py @@ -3,7 +3,9 @@ import pytest from pydantic import ValidationError -from netengine.spec import Authority, AuthorityKind, AuthorityScope, AuthoritySource +from netengine.spec import Authority, AuthorityKind, AuthorityScope, AuthoritySource, BoundaryPolicy +from netengine.spec.models import CrossWorldPeer, ServiceMirror +from netengine.spec.types import GatewayCrossWorldMode, GatewayRealInternetMode def test_authority_defaults_to_local_source() -> None: @@ -45,3 +47,61 @@ def test_authority_model_is_frozen() -> None: with pytest.raises(ValidationError): authority.operator = "other-operator" + + +def test_boundary_policy_defaults_to_isolated_no_cross_world() -> None: + policy = BoundaryPolicy() + + assert policy.real_internet == GatewayRealInternetMode.ISOLATED + assert policy.cross_world == GatewayCrossWorldMode.NONE + assert policy.service_mirrors == [] + assert policy.upstream_resolver_enabled is False + assert policy.upstream_resolver_ip is None + assert policy.peers == [] + + +def test_boundary_policy_mirrored_requires_service_mirror() -> None: + with pytest.raises(ValidationError, match="requires at least one service mirror"): + BoundaryPolicy(real_internet=GatewayRealInternetMode.MIRRORED) + + +def test_boundary_policy_service_mirrors_require_mirrored() -> None: + with pytest.raises(ValidationError, match="service mirrors require"): + BoundaryPolicy( + service_mirrors=[ + ServiceMirror(real_hostname="api.example.com", in_world_service="10.1.2.3") + ] + ) + + +def test_boundary_policy_peered_requires_peers() -> None: + with pytest.raises(ValidationError, match="requires peers"): + BoundaryPolicy(cross_world=GatewayCrossWorldMode.PEERED) + + +def test_boundary_policy_federated_requires_peer_trust_metadata() -> None: + with pytest.raises(ValidationError, match="trust bundle or peer trust anchor"): + BoundaryPolicy( + cross_world=GatewayCrossWorldMode.FEDERATED, + peers=[CrossWorldPeer(name="world-b", endpoint="10.99.0.1:8443")], + ) + + +def test_boundary_policy_federated_accepts_peer_trust_bundle() -> None: + policy = BoundaryPolicy( + cross_world=GatewayCrossWorldMode.FEDERATED, + peers=[ + CrossWorldPeer( + name="world-b", + endpoint="10.99.0.1:8443", + trust_bundle="spiffe://world-b.example/bundle", + ) + ], + ) + + assert policy.peers[0].trust_bundle == "spiffe://world-b.example/bundle" + + +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")