src/ail_ref/anchor.py

Read-only source from the unchanged 0.3.4 kit. Existing license terms apply.

"""Public-key anchor checks. A supplied file is not proof of HTTPS provenance."""
from __future__ import annotations

import json
import ssl
import urllib.error
import urllib.request
from urllib.parse import urlsplit

from .packet import verify_identity_packet

ANCHOR_FORMAT = "AIL-PUBLIC-KEY/0.3"
ANCHOR_KEYS = {"format", "principal", "display_name", "role", "language", "identity_anchor", "public_key", "statement"}
ANCHOR_PUBLIC_KEY_KEYS = {"algorithm", "encoding", "value", "fingerprint"}
ANCHOR_STATEMENT = "This public key is designated to sign AIL identity packets for the named principal."
MAX_ANCHOR_BYTES = 65536


def validate_trusted_url(url: str) -> str:
    if not isinstance(url, str):
        raise ValueError("trusted anchor URL must be a string")
    parsed = urlsplit(url)
    if (parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password
            or parsed.query or parsed.fragment or parsed.path != "/.well-known/ail-owner.json"
            or parsed.port not in (None, 443)):
        raise ValueError("trusted anchor URL must be an HTTPS /.well-known/ail-owner.json URL")
    return url


def compare_public_anchor(packet: dict, anchor: dict, trusted_url: str) -> dict:
    """Compare keys after signature verification; caller establishes anchor provenance."""
    signed = verify_identity_packet(packet)
    result = {"valid": False, "signature_status": signed.signature_status,
              "anchor_status": "ail:rejected", "identity_assurance": "ail:self_asserted_key",
              "key_id": signed.key_id, "reason": ""}
    if not signed.valid:
        result["reason"] = signed.reason
        return result
    try:
        validate_trusted_url(trusted_url)
        if packet["subject"]["identity_anchor"] != trusted_url:
            raise ValueError("packet anchor URL differs from independently selected trusted URL")
        if not isinstance(anchor, dict) or set(anchor) != ANCHOR_KEYS:
            raise ValueError("public anchor has missing or unexpected fields")
        public = anchor.get("public_key")
        if not isinstance(public, dict) or set(public) != ANCHOR_PUBLIC_KEY_KEYS:
            raise ValueError("anchor public key has missing or unexpected fields")
        subject = packet["subject"]
        for field in ("principal", "display_name", "role", "language", "identity_anchor"):
            if anchor[field] != subject[field]:
                raise ValueError("anchor " + field + " differs from signed packet")
        if anchor["format"] != ANCHOR_FORMAT or anchor["statement"] != ANCHOR_STATEMENT:
            raise ValueError("unsupported public anchor format or statement")
        if public != packet["public_key"]:
            raise ValueError("anchor public key differs from signed packet")
        result.update(valid=True, anchor_status="ail:key_matches_supplied_anchor",
                      reason="Signed packet and supplied public anchor match; source provenance is separate.")
    except (ValueError, TypeError, KeyError) as exc:
        result["reason"] = str(exc)
    return result


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, request, fp, code, msg, headers, newurl):
        return None


def fetch_trusted_anchor(trusted_url: str) -> dict:
    """Fetch an explicitly selected HTTPS origin without following redirects."""
    validate_trusted_url(trusted_url)
    opener = urllib.request.build_opener(
        urllib.request.HTTPSHandler(context=ssl.create_default_context()), _NoRedirect()
    )
    request = urllib.request.Request(trusted_url, headers={"Accept": "application/json", "User-Agent": "AIL-Reference/0.3.3"})
    with opener.open(request, timeout=8) as response:
        if response.status != 200 or response.geturl() != trusted_url:
            raise ValueError("trusted anchor response changed origin or status")
        raw = response.read(MAX_ANCHOR_BYTES + 1)
        if len(raw) > MAX_ANCHOR_BYTES:
            raise ValueError("public anchor exceeds size limit")
        return json.loads(raw.decode("utf-8"))


def verify_with_trusted_https_anchor(packet: dict, trusted_url: str) -> dict:
    """Only a successful TLS fetch from caller-selected URL upgrades domain assurance."""
    signed = verify_identity_packet(packet)
    if not signed.valid:
        return {"valid": False, "signature_status": signed.signature_status,
                "anchor_status": "ail:rejected", "identity_assurance": "ail:self_asserted_key",
                "key_id": signed.key_id, "reason": signed.reason}
    try:
        validate_trusted_url(trusted_url)
        if packet["subject"]["identity_anchor"] != trusted_url:
            raise ValueError("packet anchor URL differs from independently selected trusted URL")
        anchor = fetch_trusted_anchor(trusted_url)
        result = compare_public_anchor(packet, anchor, trusted_url)
        if result["valid"]:
            result.update(anchor_status="ail:https_domain_key_matched",
                          identity_assurance="ail:domain_anchored_key",
                          reason="Key matched the public anchor fetched from the selected HTTPS origin; personal identity remains unverified.")
        return result
    except (ValueError, urllib.error.URLError, OSError, UnicodeError, json.JSONDecodeError) as exc:
        return {"valid": False, "signature_status": signed.signature_status,
                "anchor_status": "ail:rejected", "identity_assurance": "ail:self_asserted_key",
                "key_id": signed.key_id, "reason": str(exc)}