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
4 changes: 4 additions & 0 deletions netengine/spec/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,9 @@
AuthorityScope,
AuthoritySource,
BoundaryPolicy,
ResolverPolicy,
default_authorities_for_spec,
resolver_policy_from_boundary,
)

__all__ = [
Expand All @@ -15,5 +17,7 @@
"AuthorityScope",
"AuthoritySource",
"BoundaryPolicy",
"ResolverPolicy",
"default_authorities_for_spec",
"resolver_policy_from_boundary",
]
60 changes: 60 additions & 0 deletions netengine/spec/authority.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,17 @@ class Authority(SpecModel):
source: AuthoritySource = Field(default=AuthoritySource.LOCAL)


class ResolverPolicy(SpecModel):
"""Derived resolver capabilities for a boundary authority posture."""

local_root: bool = Field(default=True)
upstream: bool = Field(default=False)
mirror_table: bool = Field(default=False)
peer_suffix_delegation: bool = Field(default=False)
imported_trust_bundle: bool = Field(default=False)
notes: list[str] = Field(default_factory=list)


class BoundaryPolicy(SpecModel):
"""Authority-layer policy for traffic and trust at the world boundary."""

Expand Down Expand Up @@ -105,6 +116,55 @@ def validate_boundary_policy(self) -> "BoundaryPolicy":
raise ValueError("isolated mode should not enable upstream resolution")

return self


def resolver_policy_from_boundary(policy: BoundaryPolicy) -> ResolverPolicy:
"""Derive resolver behavior from a boundary policy.

The result intentionally models resolver capabilities rather than concrete
DNS server ordering. Notes capture the intended precedence for modes where
ordering matters.
"""

notes: list[str] = []
upstream = False
mirror_table = False

if policy.real_internet == GatewayRealInternetMode.ISOLATED:
notes.append("isolated: local root only")
elif policy.real_internet == GatewayRealInternetMode.SHADOWED:
upstream = True
notes.append("shadowed: local root first, upstream second")
elif policy.real_internet == GatewayRealInternetMode.MIRRORED:
mirror_table = True
notes.append("mirrored: mirror table first, local root second; upstream disabled")
else:
upstream = policy.upstream_resolver_enabled
notes.append(
f"{policy.real_internet.value}: no minimum resolver semantics beyond configured upstream"
)

peer_suffix_delegation = policy.cross_world in {
GatewayCrossWorldMode.PEERED,
GatewayCrossWorldMode.FEDERATED,
}
imported_trust_bundle = policy.cross_world == GatewayCrossWorldMode.FEDERATED

if policy.cross_world == GatewayCrossWorldMode.PEERED:
notes.append("peered: peer suffix delegation without shared trust")
elif policy.cross_world == GatewayCrossWorldMode.FEDERATED:
notes.append("federated: peer suffix delegation with imported trust bundle")

return ResolverPolicy(
local_root=True,
upstream=upstream,
mirror_table=mirror_table,
peer_suffix_delegation=peer_suffix_delegation,
imported_trust_bundle=imported_trust_bundle,
notes=notes,
)


def default_authorities_for_spec(spec: "NetEngineSpec") -> list[Authority]:
"""Return stable default authorities for the control surfaces in ``spec``.

Expand Down
104 changes: 103 additions & 1 deletion tests/test_authority_spec.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,15 @@
import pytest
from pydantic import ValidationError

from netengine.spec import Authority, AuthorityKind, AuthorityScope, AuthoritySource, BoundaryPolicy
from netengine.spec import (
Authority,
AuthorityKind,
AuthorityScope,
AuthoritySource,
BoundaryPolicy,
ResolverPolicy,
resolver_policy_from_boundary,
)
from netengine.spec.models import CrossWorldPeer, ServiceMirror
from netengine.spec.types import GatewayCrossWorldMode, GatewayRealInternetMode

Expand Down Expand Up @@ -105,6 +113,100 @@ def test_boundary_policy_federated_accepts_peer_trust_bundle() -> None:
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_resolver_policy_defaults_to_local_root_only() -> None:
policy = ResolverPolicy()

assert policy.local_root is True
assert policy.upstream is False
assert policy.mirror_table is False
assert policy.peer_suffix_delegation is False
assert policy.imported_trust_bundle is False
assert policy.notes == []


def test_resolver_policy_from_isolated_boundary_uses_local_root_only() -> None:
resolver_policy = resolver_policy_from_boundary(BoundaryPolicy())

assert resolver_policy.local_root is True
assert resolver_policy.upstream is False
assert resolver_policy.mirror_table is False
assert resolver_policy.peer_suffix_delegation is False
assert resolver_policy.imported_trust_bundle is False
assert resolver_policy.notes == ["isolated: local root only"]


def test_resolver_policy_from_shadowed_boundary_enables_upstream() -> None:
resolver_policy = resolver_policy_from_boundary(
BoundaryPolicy(real_internet=GatewayRealInternetMode.SHADOWED)
)

assert resolver_policy.local_root is True
assert resolver_policy.upstream is True
assert resolver_policy.mirror_table is False
assert resolver_policy.notes == ["shadowed: local root first, upstream second"]


def test_resolver_policy_from_mirrored_boundary_uses_mirror_table_without_upstream() -> None:
resolver_policy = resolver_policy_from_boundary(
BoundaryPolicy(
real_internet=GatewayRealInternetMode.MIRRORED,
service_mirrors=[
ServiceMirror(real_hostname="api.example.com", in_world_service="10.1.2.3")
],
upstream_resolver_enabled=True,
)
)

assert resolver_policy.local_root is True
assert resolver_policy.upstream is False
assert resolver_policy.mirror_table is True
assert resolver_policy.notes == [
"mirrored: mirror table first, local root second; upstream disabled"
]


def test_resolver_policy_from_peered_boundary_delegates_peer_suffixes_without_trust() -> None:
resolver_policy = resolver_policy_from_boundary(
BoundaryPolicy(
cross_world=GatewayCrossWorldMode.PEERED,
peers=[CrossWorldPeer(name="world-b", endpoint="10.99.0.1:8443")],
)
)

assert resolver_policy.local_root is True
assert resolver_policy.peer_suffix_delegation is True
assert resolver_policy.imported_trust_bundle is False
assert resolver_policy.notes == [
"isolated: local root only",
"peered: peer suffix delegation without shared trust",
]


def test_resolver_policy_from_federated_boundary_imports_trust_bundle() -> None:
resolver_policy = resolver_policy_from_boundary(
BoundaryPolicy(
cross_world=GatewayCrossWorldMode.FEDERATED,
peers=[
CrossWorldPeer(
name="world-b",
endpoint="10.99.0.1:8443",
trust_bundle="spiffe://world-b.example/bundle",
)
],
)
)

assert resolver_policy.local_root is True
assert resolver_policy.peer_suffix_delegation is True
assert resolver_policy.imported_trust_bundle is True
assert resolver_policy.notes == [
"isolated: local root only",
"federated: peer suffix delegation with imported trust bundle",
]


def test_default_authorities_for_spec_maps_spec_sections(minimal_spec) -> None:
from netengine.spec import default_authorities_for_spec

Expand Down