- mas_core: P2-maximal parser with required_profile, generator, statute schema + semantic lint + defaults-only inheritance, exact-host discovery with typed failures and same-origin redirects, pure policy resolver - mas_conformance: vendored spec fixtures (drift-tested) + property and statute cases - mas CLI: parse / explain / discover / lint-statute / test - mas-gateway: generic mail -> MAS event -> HTTP (MTA pipe + webhook) - examples: static great-service.example statute, reference ticket service - docs/experiment-review.md: first-pass findings; live MTA test still open Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 623586@bnt-lap001 Assistant-Session: 10b30928-5b3b-4c11-9934-5ecca07367d8
165 lines
6.4 KiB
Python
165 lines
6.4 KiB
Python
"""Pure MAS policy: parse tree + validated statute (+ local registry) -> semantics.
|
|
|
|
No network access, no clock, no randomness. Without a statute, nothing
|
|
beyond syntax is claimed: opaque targets stay `kind = opaque`, resource
|
|
syntax stays unconfirmed, interaction preferences stay unconfirmed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Mapping, Optional
|
|
|
|
from .model import ParsedAddress, ParsedLocalPart, ResourceTarget
|
|
from .statute import Statute
|
|
|
|
# Kind sources, in the order they are consulted.
|
|
SOURCE_STATUTE = "statute"
|
|
SOURCE_LOCAL = "local-registry"
|
|
SOURCE_NONE = "none"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SemanticFailure:
|
|
code: str # ProfileNotEnabled | InteractionModeNotEnabled | ResourceTypeNotEnabled | UnknownResourceType
|
|
message: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AddressSemantics:
|
|
address: str
|
|
parsed: ParsedLocalPart
|
|
namespace_confirmed: bool
|
|
kind: str
|
|
kind_source: str
|
|
registry_key: Optional[str] = None
|
|
resource_type: Optional[str] = None
|
|
resource_id: Optional[str] = None
|
|
facet: Optional[str] = None
|
|
interaction: Optional[str] = None
|
|
equivalence_key: str = ""
|
|
failures: tuple[SemanticFailure, ...] = ()
|
|
unknowns: tuple[str, ...] = ()
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return not self.failures
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"address": self.address,
|
|
"namespace_confirmed": self.namespace_confirmed,
|
|
"kind": self.kind,
|
|
"kind_source": self.kind_source,
|
|
"registry_key": self.registry_key,
|
|
"resource_type": self.resource_type,
|
|
"resource_id": self.resource_id,
|
|
"facet": self.facet,
|
|
"interaction": self.interaction,
|
|
"required_profile": self.parsed.required_profile.value,
|
|
"equivalence_key": self.equivalence_key,
|
|
"failures": [{"code": f.code, "message": f.message} for f in self.failures],
|
|
"unknowns": list(self.unknowns),
|
|
}
|
|
|
|
|
|
def _fold(s: str, case: str, dots: bool) -> str:
|
|
if case == "ascii-insensitive":
|
|
s = s.lower()
|
|
if dots:
|
|
s = s.replace(".", "")
|
|
return s
|
|
|
|
|
|
def equivalence_key(parsed: ParsedLocalPart, statute: Optional[Statute]) -> str:
|
|
"""Comparison key for the base target. Exact unless a statute declares folding.
|
|
|
|
Dot folding applies to opaque targets only, never to resource identifiers.
|
|
"""
|
|
base = parsed.base
|
|
if statute is None:
|
|
return base
|
|
if isinstance(parsed.target, ResourceTarget):
|
|
return _fold(base, statute.local_part_case, False)
|
|
return _fold(base, statute.local_part_case, statute.dot_folding)
|
|
|
|
|
|
def _lookup(registry: Mapping[str, Mapping], key: str, case: str, dots: bool):
|
|
if key in registry:
|
|
return key, registry[key]
|
|
want = _fold(key, case, dots)
|
|
for k, v in registry.items():
|
|
if _fold(k, case, dots) == want:
|
|
return k, v
|
|
return None, None
|
|
|
|
|
|
def resolve(parsed, statute: Optional[Statute] = None,
|
|
local_registry: Optional[Mapping] = None, *, domain: Optional[str] = None) -> AddressSemantics:
|
|
"""Resolve semantics. `local_registry` has the statute's `addresses` /
|
|
`resource_types` shape and applies only where the statute is silent."""
|
|
if isinstance(parsed, ParsedAddress):
|
|
domain, parsed = parsed.domain, parsed.local
|
|
address = f"{parsed.raw}@{domain}" if domain else parsed.raw
|
|
local = local_registry or {}
|
|
failures: list[SemanticFailure] = []
|
|
unknowns: list[str] = []
|
|
case = statute.local_part_case if statute else "exact"
|
|
dots = statute.dot_folding if statute else False
|
|
|
|
kind, source, reg_key = "opaque", SOURCE_NONE, None
|
|
rtype = rid = None
|
|
target = parsed.target
|
|
|
|
if statute is None:
|
|
unknowns.append("namespace: no validated MAS statute; semantics unconfirmed")
|
|
|
|
if isinstance(target, ResourceTarget):
|
|
rtype, rid = target.resource_type, target.resource_id
|
|
if statute is not None and rtype in statute.resource_types:
|
|
if statute.resource_type_enabled(rtype):
|
|
kind, source = "resource", SOURCE_STATUTE
|
|
else:
|
|
failures.append(SemanticFailure("ResourceTypeNotEnabled",
|
|
f"resource type {rtype!r} is disabled by the statute"))
|
|
elif local.get("resource_types", {}).get(rtype, {}).get("enabled"):
|
|
kind, source = "resource", SOURCE_LOCAL
|
|
elif statute is not None:
|
|
failures.append(SemanticFailure("UnknownResourceType",
|
|
f"statute does not declare resource type {rtype!r}"))
|
|
else:
|
|
unknowns.append(f"resource type {rtype!r}: '=' syntax alone confers no application semantics")
|
|
else:
|
|
if statute is not None:
|
|
reg_key, entry = _lookup(statute.addresses, target.value, case, dots)
|
|
if entry:
|
|
kind, source = entry["kind"], SOURCE_STATUTE
|
|
if source == SOURCE_NONE:
|
|
k, entry = _lookup(local.get("addresses", {}), target.value, case, dots)
|
|
if entry:
|
|
reg_key, kind, source = k, entry["kind"], SOURCE_LOCAL
|
|
if source == SOURCE_NONE:
|
|
unknowns.append("kind: no registry entry; punctuation is not a classifier")
|
|
|
|
facet = parsed.facet.value if parsed.facet else None
|
|
if facet is not None:
|
|
unknowns.append(f"facet {facet!r}: application-defined; may be stripped in transit")
|
|
|
|
mode = parsed.interaction.mode if parsed.interaction else None
|
|
if mode is not None:
|
|
if statute is None:
|
|
unknowns.append(f"interaction {mode!r}: P2 support unconfirmed")
|
|
elif not statute.p2:
|
|
failures.append(SemanticFailure("ProfileNotEnabled",
|
|
"address uses '!' (P2) but the statute does not enable P2"))
|
|
elif not statute.interaction_mode_enabled(mode):
|
|
failures.append(SemanticFailure("InteractionModeNotEnabled",
|
|
f"interaction mode {mode!r} is not enabled by the statute"))
|
|
|
|
return AddressSemantics(
|
|
address=address, parsed=parsed, namespace_confirmed=statute is not None,
|
|
kind=kind, kind_source=source, registry_key=reg_key,
|
|
resource_type=rtype, resource_id=rid, facet=facet, interaction=mode,
|
|
equivalence_key=equivalence_key(parsed, statute),
|
|
failures=tuple(failures), unknowns=tuple(unknowns),
|
|
)
|