# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-w3c-baggage-allowlist-and-pii-sanitization-gates-must-an-instance-propagate-cross-service-context-to-prevent-header-amplification-and-external-data-exfiltration.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-w3c-baggage-allowlist-and-pii-sanitization-gates-must-an-instance-propagate-cross-service-context-to-prevent-header-amplification-and-external-data-exfiltration.warrant.json
# Markdown SHA-256:   3763c51e87c4188e0255729c5d0162531af686a124fe1e8f46316aae661ed2ee
# Code AST SHA-256:   e0e24b589827aeceac37ae8a3930599eb10d95c63a2709f6b134eedfe2e9bcd2
# Verified Classes:   BaggagePolicyViolationError, SafeBaggageContext
# Defensive Guards:   7 fail-closed invariants (zero external deps)

import re
from dataclasses import dataclass
from typing import Mapping
from urllib.parse import quote, unquote, urlsplit

# Strict lowercase RFC-7230-safe key token (1..64 chars)
_BAGGAGE_KEY_RE = re.compile(r"^[a-z0-9_.-]{1,64}$")

# Forbidden substrings in any baggage key to prevent accidental PII/secret injection
_FORBIDDEN_KEY_SUBSTRINGS = (
    "email", "mail", "token", "secret", "password", "passwd", "auth",
    "jwt", "bearer", "cookie", "session", "ssn", "phone", "card", "pii",
)

# Default operational allowlist for cross-service context propagation
DEFAULT_ALLOWED_BAGGAGE_KEYS: frozenset[str] = frozenset({
    "tenant_id",
    "tenant_tier",
    "canary_cohort",
    "synthetic_run_id",
    "region_affinity",
    "priority_class",
})


class BaggagePolicyViolationError(ValueError):
    '''Raised when a baggage key/value violates allowlist, PII, or byte-budget gates.'''
    pass


@dataclass(frozen=True)
class SafeBaggageContext:
    '''
    Immutable, size-bounded, allowlisted W3C Baggage container.
    Enforces:
    1. Perimeter stripping on untrusted ingress.
    2. Strict operational key allowlist + PII substring block.
    3. Percent-encoded values (`quote(val, safe="")`) with <= 2048B total wire size.
    4. Egress domain fencing so external third-party APIs never receive baggage.
    '''
    entries: Mapping[str, str]

    MAX_ENTRIES: int = 16
    MAX_VALUE_CHARS: int = 128
    MAX_WIRE_BYTES: int = 2048

    @classmethod
    def from_inbound_header(
        cls,
        raw_baggage_header: str | None,
        *,
        is_trusted_internal_ingress: bool = False,
        allowed_keys: frozenset[str] = DEFAULT_ALLOWED_BAGGAGE_KEYS,
    ) -> "SafeBaggageContext":
        '''
        Parses an incoming W3C `baggage` header.
        If `is_trusted_internal_ingress` is False (public edge), strips 100% of
        client-supplied baggage to prevent privilege/cohort spoofing.
        '''
        if not is_trusted_internal_ingress or not raw_baggage_header:
            return cls(entries={})

        if len(raw_baggage_header) > cls.MAX_WIRE_BYTES * 2:
            # Drop oversized incoming header rather than risking CPU/memory stalls
            return cls(entries={})

        parsed: dict[str, str] = {}
        for item in raw_baggage_header.split(","):
            if len(parsed) >= cls.MAX_ENTRIES:
                break
            # Strip optional W3C `;prop=val` metadata parameters per member
            member = item.split(";", 1)[0].strip()
            if "=" not in member:
                continue
            raw_k, raw_v = member.split("=", 1)
            k = raw_k.strip().lower()
            if k not in allowed_keys or not _BAGGAGE_KEY_RE.match(k):
                continue
            val = unquote(raw_v.strip())
            if 0 < len(val) <= cls.MAX_VALUE_CHARS and all(32 <= ord(c) <= 126 for c in val):
                parsed[k] = val

        return cls(entries=parsed)

    def with_updated_entry(
        self,
        key: str,
        value: str,
        *,
        allowed_keys: frozenset[str] = DEFAULT_ALLOWED_BAGGAGE_KEYS,
    ) -> "SafeBaggageContext":
        '''
        Adds or updates a single baggage entry after validating allowlist,
        PII key restrictions, character bounds, and total wire byte budget.
        '''
        norm_key = key.strip().lower()
        if not _BAGGAGE_KEY_RE.match(norm_key):
            raise BaggagePolicyViolationError(f"Malformed baggage key: {key!r}")

        if any(bad in norm_key for bad in _FORBIDDEN_KEY_SUBSTRINGS):
            raise BaggagePolicyViolationError(
                f"Forbidden PII/credential key name in W3C baggage: {norm_key!r}"
            )

        if norm_key not in allowed_keys:
            raise BaggagePolicyViolationError(
                f"Baggage key {norm_key!r} is not in the operational allowlist"
            )

        clean_val = value.strip()
        if not clean_val or len(clean_val) > self.MAX_VALUE_CHARS:
            raise BaggagePolicyViolationError(
                f"Baggage value for {norm_key!r} must be 1..{self.MAX_VALUE_CHARS} chars"
            )
        if not all(32 <= ord(c) <= 126 for c in clean_val):
            raise BaggagePolicyViolationError(
                f"Baggage value for {norm_key!r} contains non-printable ASCII characters"
            )

        updated = dict(self.entries)
        if norm_key not in updated and len(updated) >= self.MAX_ENTRIES:
            raise BaggagePolicyViolationError(
                f"Baggage entry count exceeds cap of {self.MAX_ENTRIES}"
            )
        updated[norm_key] = clean_val

        candidate = SafeBaggageContext(entries=updated)
        wire_str = candidate._serialize_unchecked()
        if len(wire_str.encode("ascii")) > self.MAX_WIRE_BYTES:
            raise BaggagePolicyViolationError(
                f"Serialized baggage exceeds {self.MAX_WIRE_BYTES}-byte proxy safety ceiling"
            )
        return candidate

    def _serialize_unchecked(self) -> str:
        return ",".join(
            f"{k}={quote(v, safe='')}" for k, v in sorted(self.entries.items())
        )

    def to_outbound_headers(
        self,
        target_url: str,
        *,
        internal_domain_suffixes: tuple[str, ...] = (".internal", ".svc.cluster.local"),
    ) -> dict[str, str]:
        '''
        Emits the `baggage` header ONLY if `target_url` resolves to an explicitly
        trusted internal domain suffix. Strips baggage completely on external calls.
        '''
        if not self.entries:
            return {}

        hostname = (urlsplit(target_url).hostname or "").strip().lower().rstrip(".")
        is_internal = any(
            hostname == suffix.lstrip(".") or hostname.endswith(suffix)
            for suffix in internal_domain_suffixes
        )
        if not is_internal:
            return {}

        return {"baggage": self._serialize_unchecked()}