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
9 changes: 8 additions & 1 deletion netengine/spec/__init__.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,17 @@
"""Spec parsing and models for NetEngine declarative specifications."""

from netengine.spec.authority import Authority, AuthorityKind, AuthorityScope, AuthoritySource
from netengine.spec.authority import (
Authority,
AuthorityKind,
AuthorityScope,
AuthoritySource,
default_authorities_for_spec,
)

__all__ = [
"Authority",
"AuthorityKind",
"AuthorityScope",
"AuthoritySource",
"default_authorities_for_spec",
]
116 changes: 116 additions & 0 deletions netengine/spec/authority.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,15 @@
"""Foundational authority primitives for NetEngine specifications."""

from enum import Enum
from typing import TYPE_CHECKING

from pydantic import Field

from netengine.spec.models import SpecModel

if TYPE_CHECKING:
from netengine.spec.models import NetEngineSpec


class AuthorityKind(str, Enum):
"""Kinds of authority recognized by NetEngine specs."""
Expand Down Expand Up @@ -53,3 +58,114 @@ class Authority(SpecModel):
controls: list[str] = Field(...)
description: str | None = Field(default=None)
source: AuthoritySource = Field(default=AuthoritySource.LOCAL)


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

The default authority set is derived from the existing top-level spec
sections. Each returned authority uses a stable id so downstream runtime
state, audit records, and generated artifacts can refer to authority
surfaces consistently across loads of the same spec.
"""

core_network_controls = [
f"substrate.networks.{name}"
for name in sorted(spec.substrate.networks)
if name == "core" or name.startswith("core-") or name.startswith("core_")
]

return [
Authority(
id="world-root",
kind=AuthorityKind.WORLD_ROOT,
scope=AuthorityScope.WORLD,
operator="world_registry",
controls=["world_registry"],
description="Root governance authority for world admission and policy.",
),
Authority(
id="root-naming",
kind=AuthorityKind.ROOT_NAMING,
scope=AuthorityScope.WORLD,
operator="dns",
controls=["dns.root", "dns.tlds"],
description="Authority over the in-world DNS root and TLD delegations.",
),
Authority(
id="numbering",
kind=AuthorityKind.NUMBERING,
scope=AuthorityScope.WORLD,
operator="domain_registry",
controls=["domain_registry.address_space", *core_network_controls],
description="Authority over address-space allocation and core network pools.",
),
Authority(
id="domain-registry",
kind=AuthorityKind.DOMAIN_REGISTRY,
scope=AuthorityScope.INWORLD,
operator="domain_registry",
controls=["domain_registry"],
description="Authority over in-world domain resources and registry records.",
),
Authority(
id="default-registrar",
kind=AuthorityKind.REGISTRAR,
scope=AuthorityScope.INWORLD,
operator="domain_registry.registrar",
controls=["domain_registry.registrar"],
description="Default registrar authority for domain registration workflows.",
),
Authority(
id="trust-root",
kind=AuthorityKind.TRUST,
scope=AuthorityScope.WORLD,
operator="pki",
controls=["pki.root_ca", "pki.acme"],
description="Authority over root trust and certificate issuance.",
),
Authority(
id="platform-identity",
kind=AuthorityKind.PLATFORM_IDENTITY,
scope=AuthorityScope.PLATFORM,
operator="identity_platform",
controls=["identity_platform"],
description="Authority over platform operator identity.",
),
Authority(
id="inworld-identity",
kind=AuthorityKind.INWORLD_IDENTITY,
scope=AuthorityScope.INWORLD,
operator="identity_inworld",
controls=["identity_inworld"],
description="Authority over in-world inhabitants and organization users.",
),
Authority(
id="transit-boundary",
kind=AuthorityKind.TRANSIT,
scope=AuthorityScope.BOUNDARY,
operator="gateway_portal",
controls=[
"gateway_portal",
"gateway_portal.real_internet",
"gateway_portal.cross_world",
],
description="Authority over gateway portal transit and boundary posture.",
),
Authority(
id="mail-authority",
kind=AuthorityKind.MAIL,
scope=AuthorityScope.WORLD,
operator="world_services.mail",
controls=["world_services.mail"],
description="Authority over world mail service configuration.",
),
Authority(
id="service-catalog",
kind=AuthorityKind.SERVICE_CATALOG,
scope=AuthorityScope.WORLD,
operator="org_apps.catalog",
controls=["org_apps.catalog"],
description="Authority over the application service catalog.",
),
]
41 changes: 41 additions & 0 deletions tests/test_authority_spec.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,3 +45,44 @@ def test_authority_model_is_frozen() -> None:

with pytest.raises(ValidationError):
authority.operator = "other-operator"


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

authorities = default_authorities_for_spec(minimal_spec)
by_id = {authority.id: authority for authority in authorities}

assert list(by_id) == [
"world-root",
"root-naming",
"numbering",
"domain-registry",
"default-registrar",
"trust-root",
"platform-identity",
"inworld-identity",
"transit-boundary",
"mail-authority",
"service-catalog",
]
assert by_id["root-naming"].kind == AuthorityKind.ROOT_NAMING
assert by_id["root-naming"].controls == ["dns.root", "dns.tlds"]
assert by_id["numbering"].kind == AuthorityKind.NUMBERING
assert by_id["numbering"].controls == [
"domain_registry.address_space",
"substrate.networks.core",
]
assert by_id["world-root"].controls == ["world_registry"]
assert by_id["domain-registry"].controls == ["domain_registry"]
assert by_id["default-registrar"].controls == ["domain_registry.registrar"]
assert by_id["trust-root"].controls == ["pki.root_ca", "pki.acme"]
assert by_id["platform-identity"].controls == ["identity_platform"]
assert by_id["inworld-identity"].controls == ["identity_inworld"]
assert by_id["transit-boundary"].controls == [
"gateway_portal",
"gateway_portal.real_internet",
"gateway_portal.cross_world",
]
assert by_id["mail-authority"].controls == ["world_services.mail"]
assert by_id["service-catalog"].controls == ["org_apps.catalog"]