diff --git a/netengine/spec/__init__.py b/netengine/spec/__init__.py index 3fa4710..0f64019 100644 --- a/netengine/spec/__init__.py +++ b/netengine/spec/__init__.py @@ -7,6 +7,7 @@ AuthoritySource, BoundaryPolicy, ResolverPolicy, + TrustBundle, default_authorities_for_spec, resolver_policy_from_boundary, ) @@ -18,6 +19,7 @@ "AuthoritySource", "BoundaryPolicy", "ResolverPolicy", + "TrustBundle", "default_authorities_for_spec", "resolver_policy_from_boundary", ] diff --git a/netengine/spec/authority.py b/netengine/spec/authority.py index aa99adb..f8c501b 100644 --- a/netengine/spec/authority.py +++ b/netengine/spec/authority.py @@ -72,6 +72,39 @@ class ResolverPolicy(SpecModel): notes: list[str] = Field(default_factory=list) +class TrustBundle(SpecModel): + """Trust metadata imported for a peered or federated world boundary.""" + + peer_id: str = Field(...) + peer_name: str | None = Field(default=None) + mode: GatewayCrossWorldMode = Field(...) + dns_suffixes: list[str] = Field(default_factory=list) + peer_root_ca: str | None = Field(default=None) + oidc_issuer: str | None = Field(default=None) + accepted_audiences: list[str] = Field(default_factory=list) + mail_domains: list[str] = Field(default_factory=list) + dkim_policy: str | None = Field(default=None) + + @model_validator(mode="after") + def validate_trust_bundle(self) -> "TrustBundle": + """Validate cross-world trust metadata requirements.""" + if self.mode == GatewayCrossWorldMode.NONE: + raise ValueError("trust bundle mode must be peered or federated, not none") + + if not self.dns_suffixes: + raise ValueError("trust bundle dns_suffixes must not be empty") + + if self.mode == GatewayCrossWorldMode.FEDERATED and not ( + self.peer_root_ca or self.oidc_issuer + ): + raise ValueError( + "federated trust bundle requires at least one trust-bearing field " + "such as peer_root_ca or oidc_issuer" + ) + + return self + + class BoundaryPolicy(SpecModel): """Authority-layer policy for traffic and trust at the world boundary.""" diff --git a/tests/test_authority_spec.py b/tests/test_authority_spec.py index 00875c3..9abc1ac 100644 --- a/tests/test_authority_spec.py +++ b/tests/test_authority_spec.py @@ -10,6 +10,7 @@ AuthoritySource, BoundaryPolicy, ResolverPolicy, + TrustBundle, resolver_policy_from_boundary, ) from netengine.spec.models import CrossWorldPeer, ServiceMirror @@ -57,6 +58,66 @@ def test_authority_model_is_frozen() -> None: authority.operator = "other-operator" +def test_trust_bundle_accepts_peered_dns_suffixes_without_trust_metadata() -> None: + bundle = TrustBundle( + peer_id="world-b", + peer_name="World B", + mode=GatewayCrossWorldMode.PEERED, + dns_suffixes=["world-b.test"], + ) + + assert bundle.peer_id == "world-b" + assert bundle.peer_name == "World B" + assert bundle.mode == GatewayCrossWorldMode.PEERED + assert bundle.dns_suffixes == ["world-b.test"] + assert bundle.accepted_audiences == [] + assert bundle.mail_domains == [] + + +def test_trust_bundle_rejects_none_mode() -> None: + with pytest.raises(ValidationError, match="must be peered or federated"): + TrustBundle( + peer_id="world-b", + mode=GatewayCrossWorldMode.NONE, + dns_suffixes=["world-b.test"], + ) + + +def test_trust_bundle_requires_dns_suffixes() -> None: + with pytest.raises(ValidationError, match="dns_suffixes must not be empty"): + TrustBundle(peer_id="world-b", mode=GatewayCrossWorldMode.PEERED, dns_suffixes=[]) + + +def test_trust_bundle_federated_requires_trust_bearing_metadata() -> None: + with pytest.raises(ValidationError, match="trust-bearing field"): + TrustBundle( + peer_id="world-b", + mode=GatewayCrossWorldMode.FEDERATED, + dns_suffixes=["world-b.test"], + ) + + +def test_trust_bundle_federated_accepts_peer_root_ca_or_oidc_issuer() -> None: + ca_bundle = TrustBundle( + peer_id="world-b", + mode=GatewayCrossWorldMode.FEDERATED, + dns_suffixes=["world-b.test"], + peer_root_ca="-----BEGIN CERTIFICATE-----...", + ) + oidc_bundle = TrustBundle( + peer_id="world-c", + mode="federated", + dns_suffixes=["world-c.test"], + oidc_issuer="https://issuer.world-c.test", + accepted_audiences=["netengine"], + ) + + assert ca_bundle.peer_root_ca == "-----BEGIN CERTIFICATE-----..." + assert oidc_bundle.mode == GatewayCrossWorldMode.FEDERATED + assert oidc_bundle.oidc_issuer == "https://issuer.world-c.test" + assert oidc_bundle.accepted_audiences == ["netengine"] + + def test_boundary_policy_defaults_to_isolated_no_cross_world() -> None: policy = BoundaryPolicy()