diff --git a/netengine/spec/__init__.py b/netengine/spec/__init__.py index da21a1c..3234d3d 100644 --- a/netengine/spec/__init__.py +++ b/netengine/spec/__init__.py @@ -5,6 +5,7 @@ AuthorityKind, AuthorityScope, AuthoritySource, + BoundaryPolicy, default_authorities_for_spec, ) @@ -13,5 +14,6 @@ "AuthorityKind", "AuthorityScope", "AuthoritySource", + "BoundaryPolicy", "default_authorities_for_spec", ] diff --git a/netengine/spec/authority.py b/netengine/spec/authority.py index 6ab424d..2818f79 100644 --- a/netengine/spec/authority.py +++ b/netengine/spec/authority.py @@ -1,11 +1,12 @@ """Foundational authority primitives for NetEngine specifications.""" from enum import Enum +from pydantic import Field, model_validator from typing import TYPE_CHECKING -from pydantic import Field -from netengine.spec.models import SpecModel +from netengine.spec.models import CrossWorldPeer, ServiceMirror, SpecModel +from netengine.spec.types import GatewayCrossWorldMode, GatewayRealInternetMode if TYPE_CHECKING: from netengine.spec.models import NetEngineSpec @@ -60,6 +61,50 @@ class Authority(SpecModel): 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 def default_authorities_for_spec(spec: "NetEngineSpec") -> list[Authority]: """Return stable default authorities for the control surfaces in ``spec``. 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 f4dc96f..9eae4a2 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: @@ -47,6 +49,62 @@ def test_authority_model_is_frozen() -> None: 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") def test_default_authorities_for_spec_maps_spec_sections(minimal_spec) -> None: from netengine.spec import default_authorities_for_spec