mail-address-semantics/mas_core/statute.py

"""MAS v0.1 statute model, schema validation, semantic validation, inheritance.

Everything here is pure: documents come in as already-decoded JSON objects,
and inheritance resolution takes a caller-supplied `fetch` function. Network
access lives in `mas_core.discovery` only.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from importlib import resources
from types import MappingProxyType
from typing import Callable, Mapping, Optional

from jsonschema import Draft202012Validator, FormatChecker

from .parser import ATOM_CHARS, RESOURCE_TYPE_CHARS, MasSyntaxError, parse_local_part

ERROR, WARNING = "error", "warning"
SCHEMA, SEMANTIC = "schema", "semantic"


@dataclass(frozen=True)
class Finding:
    layer: str      # "schema" | "semantic"
    severity: str   # "error" | "warning"
    code: str
    path: str
    message: str

    def __str__(self) -> str:
        return f"{self.layer} {self.severity} [{self.code}] {self.path or '/'}: {self.message}"


class StatuteInvalid(ValueError):
    def __init__(self, findings: list[Finding]):
        self.findings = [f for f in findings if f.severity == ERROR]
        super().__init__("; ".join(str(f) for f in self.findings))

    @property
    def layer(self) -> str:
        return SCHEMA if any(f.layer == SCHEMA for f in self.findings) else SEMANTIC

    @property
    def codes(self) -> set[str]:
        return {f.code for f in self.findings}


@dataclass(frozen=True)
class Statute:
    namespace: str
    p2: bool
    resource_types: Mapping[str, Mapping]
    interaction_modes: Mapping[str, Mapping]
    local_part_case: str
    dot_folding: bool
    addresses: Mapping[str, Mapping]
    inherits: Optional[str] = None
    unicode_local_part: bool = False
    inherited_from: tuple[str, ...] = ()
    document: Mapping = field(default_factory=dict, compare=False, repr=False)

    def resource_type_enabled(self, name: str) -> bool:
        spec = self.resource_types.get(name)
        return bool(spec and spec.get("enabled"))

    def interaction_mode_enabled(self, name: str) -> bool:
        spec = self.interaction_modes.get(name)
        return bool(spec and spec.get("enabled"))

    def to_dict(self) -> dict:
        return {
            "namespace": self.namespace,
            "profiles": {"p1": True, "p2": self.p2},
            "resource_types": {k: dict(v) for k, v in self.resource_types.items()},
            "interaction_modes": {k: dict(v) for k, v in self.interaction_modes.items()},
            "equivalence": {"local_part_case": self.local_part_case, "dot_folding": self.dot_folding},
            "addresses": {k: dict(v) for k, v in self.addresses.items()},
            "inherits": self.inherits,
            "inherited_from": list(self.inherited_from),
        }


_SCHEMA = json.loads(resources.files("mas_core").joinpath(
    "schema/mas-statute-v0.1.schema.json").read_text(encoding="utf-8"))
_VALIDATOR = Draft202012Validator(_SCHEMA, format_checker=FormatChecker())


def canonical_host(host: str) -> str:
    """Normal domain-name canonicalization: case, trailing dot, IDNA A-label."""
    h = host.strip().rstrip(".").lower()
    try:
        return h.encode("idna").decode("ascii")
    except UnicodeError:
        return h


def schema_findings(doc) -> list[Finding]:
    out = []
    for err in sorted(_VALIDATOR.iter_errors(doc), key=lambda e: list(e.absolute_path)):
        path = "/" + "/".join(str(p) for p in err.absolute_path)
        out.append(Finding(SCHEMA, ERROR, "schema-" + str(err.validator), path, err.message))
    return out


def _fold(key: str, case: str, dots: bool) -> str:
    if case == "ascii-insensitive":
        key = key.lower()
    if dots:
        key = key.replace(".", "")
    return key


def semantic_findings(doc: Mapping, host: Optional[str] = None,
                      statute_url: Optional[str] = None) -> list[Finding]:
    """Checks beyond the JSON schema. Assumes `doc` is schema-valid."""
    out: list[Finding] = []

    def add(sev, code, path, msg):
        out.append(Finding(SEMANTIC, sev, code, path, msg))

    ns = doc["namespace"]
    if canonical_host(ns) != ns:
        add(WARNING, "namespace-not-canonical", "/namespace",
            f"namespace should be in canonical form {canonical_host(ns)!r}")
    if host is not None and canonical_host(ns) != canonical_host(host):
        add(ERROR, "namespace-mismatch", "/namespace",
            f"statute declares {ns!r} but was retrieved for host {host!r}")

    p2 = doc["profiles"]["p2"]
    modes = doc.get("interaction_modes", {})
    enabled_modes = [m for m, spec in modes.items() if spec["enabled"]]
    if not p2 and enabled_modes:
        add(ERROR, "p2-disabled-modes-enabled", "/interaction_modes",
            f"profiles.p2 is false but interaction modes are enabled: {', '.join(enabled_modes)}")
    if p2 and not enabled_modes and not doc.get("inherits"):
        add(WARNING, "p2-without-modes", "/profiles/p2",
            "P2 is enabled but no interaction mode is enabled")

    for name, spec in doc.get("resource_types", {}).items():
        path = f"/resource_types/{name}"
        if not name or not all(c in RESOURCE_TYPE_CHARS for c in name):
            add(ERROR, "invalid-resource-type-name", path, "name violates MAS resource-type syntax")
        if spec.get("identifier_policy") == "sequential-allowed":
            add(WARNING, "enumerable-identifiers", path,
                "sequential identifiers are enumerable; identifiers are not credentials")

    case = doc["equivalence"]["local_part_case"]
    dots = doc["equivalence"]["dot_folding"]
    seen: dict[str, str] = {}
    for key in doc.get("addresses", {}):
        path = f"/addresses/{key}"
        bad = sorted(set(key) & set("+!$^"))
        if bad:
            add(ERROR, "registry-key-operator", path,
                f"registry keys name base opaque targets; {' '.join(bad)} not allowed")
            continue
        try:
            parsed = parse_local_part(key)
        except MasSyntaxError as e:
            add(ERROR, "registry-key-syntax", path, str(e))
            continue
        if parsed.target.kind != "opaque":
            add(ERROR, "registry-key-resource", path,
                "registry keys must be opaque targets; resources are declared via resource_types")
            continue
        folded = _fold(key, case, dots)
        if folded in seen:
            add(ERROR, "registry-key-collision", path,
                f"collides with {seen[folded]!r} under the declared equivalence policy")
        seen[folded] = key

    inherits = doc.get("inherits")
    own_url = statute_url or f"https://{canonical_host(ns)}/.well-known/mas"
    if inherits and inherits.rstrip("/") == own_url.rstrip("/"):
        add(ERROR, "inheritance-cycle", "/inherits", "statute inherits from itself")

    if doc.get("unicode_local_part"):
        add(WARNING, "unicode-local-part", "/unicode_local_part",
            "experimental: Unicode local-parts are not canonical MAS v0.1 syntax")
    return out


def lint(doc, host: Optional[str] = None) -> list[Finding]:
    """All findings: schema errors, or (if schema-valid) semantic errors and warnings."""
    found = schema_findings(doc)
    if found:
        return found
    return semantic_findings(doc, host)


def build_statute(doc: Mapping, host: Optional[str] = None,
                  statute_url: Optional[str] = None) -> Statute:
    """Validate (schema + semantics) and build a Statute. Raises StatuteInvalid."""
    found = schema_findings(doc) or semantic_findings(doc, host, statute_url)
    if any(f.severity == ERROR for f in found):
        raise StatuteInvalid(found)
    ro = lambda d: MappingProxyType({k: MappingProxyType(dict(v)) for k, v in d.items()})
    return Statute(
        namespace=canonical_host(doc["namespace"]),
        p2=doc["profiles"]["p2"],
        resource_types=ro(doc.get("resource_types", {})),
        interaction_modes=ro(doc.get("interaction_modes", {})),
        local_part_case=doc["equivalence"]["local_part_case"],
        dot_folding=doc["equivalence"]["dot_folding"],
        addresses=ro(doc.get("addresses", {})),
        inherits=doc.get("inherits"),
        unicode_local_part=doc.get("unicode_local_part", False),
        document=MappingProxyType(dict(doc)),
    )


def merge_defaults(child: Statute, parent: Statute, parent_url: str) -> Statute:
    """Defaults-only inheritance: child wins; address registry never inherits."""
    rt = {**parent.resource_types, **child.resource_types}
    im = {**parent.interaction_modes, **child.interaction_modes}
    return Statute(
        namespace=child.namespace, p2=child.p2,
        resource_types=MappingProxyType(rt), interaction_modes=MappingProxyType(im),
        local_part_case=child.local_part_case, dot_folding=child.dot_folding,
        addresses=child.addresses, inherits=child.inherits,
        unicode_local_part=child.unicode_local_part,
        inherited_from=(parent_url,) + parent.inherited_from,
        document=child.document,
    )


class InheritanceCycle(ValueError):
    def __init__(self, chain: list[str]):
        super().__init__(" -> ".join(chain))
        self.chain = chain


def resolve_inheritance(statute: Statute, own_url: str,
                        fetch: Callable[[str], Statute], max_depth: int = 8) -> Statute:
    """Follow explicit `inherits` URIs only, detecting cycles (A -> B -> A, ...).

    `fetch(url)` must return an independently validated parent Statute, or
    raise. Parents are fetched for their own namespace; their registry is
    discarded by `merge_defaults`.
    """
    chain = [own_url.rstrip("/")]
    statutes = [statute]
    cur = statute
    while cur.inherits:
        url = cur.inherits.rstrip("/")
        if url in chain:
            raise InheritanceCycle(chain + [url])
        if len(chain) > max_depth:
            raise InheritanceCycle(chain + [url + " (depth limit)"])
        chain.append(url)
        cur = fetch(cur.inherits)
        statutes.append(cur)
    merged = statutes[-1]
    for i in range(len(statutes) - 2, -1, -1):
        merged = merge_defaults(statutes[i], merged, chain[i + 1])
    return merged