Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions netengine/spec/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
AuthoritySource,
BoundaryPolicy,
ResolverPolicy,
TrustBundle,
default_authorities_for_spec,
resolver_policy_from_boundary,
)
Expand All @@ -18,6 +19,7 @@
"AuthoritySource",
"BoundaryPolicy",
"ResolverPolicy",
"TrustBundle",
"default_authorities_for_spec",
"resolver_policy_from_boundary",
]
33 changes: 33 additions & 0 deletions netengine/spec/authority.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down
61 changes: 61 additions & 0 deletions tests/test_authority_spec.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
AuthoritySource,
BoundaryPolicy,
ResolverPolicy,
TrustBundle,
resolver_policy_from_boundary,
)
from netengine.spec.models import CrossWorldPeer, ServiceMirror
Expand Down Expand Up @@ -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()

Expand Down