Add BoundaryPolicy to authority spec with cross-field validation

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",
 ]

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``.
 

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
 
 

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