Gitea-Tools/session_context_binding.py

"""Session-immutable MCP mutation context (#714).

Pins profile, remote, host, repository, identity, and role for the life of an
MCP process (or until an explicit ``gitea_activate_profile`` re-bind).

Capability resolution and mutation gates must evaluate only the active
profile for the requested remote. Silent cross-host / cross-profile
substitution is forbidden.
"""

from __future__ import annotations

import os
import re
import threading
import urllib.parse
from dataclasses import dataclass
from typing import Any, Mapping


@dataclass(frozen=True, slots=True)
class _SessionContext:
    """One atomic, immutable process-session binding."""

    profile_name: str | None
    remote: str | None
    host: str | None
    identity: str | None
    repository: str | None
    org: str | None
    role_kind: str | None
    expected_username: str | None
    source: str
    pid: int

    def as_dict(self) -> dict[str, Any]:
        return {
            "profile_name": self.profile_name,
            "remote": self.remote,
            "host": self.host,
            "identity": self.identity,
            "repository": self.repository,
            "org": self.org,
            "role_kind": self.role_kind,
            "expected_username": self.expected_username,
            "source": self.source,
            "pid": self.pid,
        }


# Process-local only — never a shared file (same rationale as mutation authority).
# The frozen value prevents partial mutation, while the lock makes first-bind and
# sanctioned rebind atomic across concurrent MCP calls.
_SESSION_CONTEXT: _SessionContext | None = None
_SESSION_CONTEXT_LOCK = threading.RLock()


def _reset_session_context_for_testing() -> None:
    """Reset at a pytest test boundary; unavailable to production callers.

    Production sessions transition only through process startup/PID change or
    the explicit profile-activation rebind.  Keeping this helper private and
    requiring pytest's per-test marker prevents it from becoming an MCP/runtime
    bypass.
    """
    if "PYTEST_CURRENT_TEST" not in os.environ:
        raise RuntimeError("session context reset is restricted to pytest boundaries")
    global _SESSION_CONTEXT
    with _SESSION_CONTEXT_LOCK:
        _SESSION_CONTEXT = None


def get_session_context() -> dict[str, Any] | None:
    """Return a detached snapshot of the bound context, or None if unbound."""
    with _SESSION_CONTEXT_LOCK:
        if _SESSION_CONTEXT is None:
            return None
        return _SESSION_CONTEXT.as_dict()


def profile_host(profile: dict | None) -> str | None:
    """Hostname from profile base_url, lowercased, or None."""
    if not profile:
        return None
    base = (profile.get("base_url") or "").strip()
    if not base:
        return None
    try:
        parsed = urllib.parse.urlparse(base)
        host = (parsed.netloc or parsed.path or "").strip().lower()
        return host or None
    except Exception:
        return None


def remote_host(remote: str | None, remotes: dict | None) -> str | None:
    """Hostname for a known remote key."""
    if not remote or not remotes:
        return None
    entry = remotes.get(remote) or {}
    return (entry.get("host") or "").strip().lower() or None


def profile_matches_remote(
    profile: dict | None,
    remote: str | None,
    remotes: dict | None,
    *,
    contexts: dict | None = None,
) -> bool:
    """True when *profile* is bound to the same host/context as *remote*."""
    if not profile or not remote:
        return False
    r_host = remote_host(remote, remotes)
    p_host = profile_host(profile)
    if r_host and p_host:
        return r_host == p_host
    # Fall back to context name heuristics when base_url missing.
    ctx = (profile.get("context") or "").strip().lower()
    if not ctx or not contexts:
        return False
    ctx_data = contexts.get(ctx) or {}
    gitea = ctx_data.get("gitea") or {}
    base = (gitea.get("base_url") or "").strip()
    if not base or not r_host:
        return False
    try:
        parsed = urllib.parse.urlparse(base)
        c_host = (parsed.netloc or parsed.path or "").strip().lower()
    except Exception:
        return False
    return bool(c_host) and c_host == r_host


def bind_session_context(
    *,
    profile_name: str,
    remote: str | None,
    host: str | None,
    identity: str | None,
    repository: str | None = None,
    org: str | None = None,
    role_kind: str | None = None,
    expected_username: str | None = None,
    source: str = "bind",
) -> dict[str, Any]:
    """Atomically bind/re-bind context (the explicit activation path)."""
    with _SESSION_CONTEXT_LOCK:
        return _bind_session_context_unlocked(
            profile_name=profile_name,
            remote=remote,
            host=host,
            identity=identity,
            repository=repository,
            org=org,
            role_kind=role_kind,
            expected_username=expected_username,
            source=source,
        )


def _bind_session_context_unlocked(
    *,
    profile_name: str,
    remote: str | None,
    host: str | None,
    identity: str | None,
    repository: str | None,
    org: str | None,
    role_kind: str | None,
    expected_username: str | None,
    source: str,
) -> dict[str, Any]:
    """Store a complete immutable context while the caller holds the lock."""
    global _SESSION_CONTEXT
    _SESSION_CONTEXT = _SessionContext(
        profile_name=(profile_name or "").strip() or None,
        remote=(remote or "").strip() or None,
        host=(host or "").strip().lower() or None,
        identity=(identity or "").strip() or None,
        repository=(repository or "").strip() or None,
        org=(org or "").strip() or None,
        role_kind=(role_kind or "").strip() or None,
        expected_username=(expected_username or "").strip() or None,
        source=source,
        pid=os.getpid(),
    )
    return _SESSION_CONTEXT.as_dict()


def seed_session_context_if_unbound(
    *,
    profile_name: str,
    remote: str | None,
    host: str | None,
    identity: str | None,
    repository: str | None = None,
    org: str | None = None,
    role_kind: str | None = None,
    expected_username: str | None = None,
    source: str = "seed",
) -> dict[str, Any]:
    """Atomically bind only when this process has no current context.

    A changed environment or an interleaved call is not a session boundary and
    therefore cannot replace an established binding.  A newly started/forked
    process is recognized by PID; explicit ``gitea_activate_profile`` uses
    :func:`bind_session_context` as its sanctioned logical-session transition.
    """
    with _SESSION_CONTEXT_LOCK:
        if _SESSION_CONTEXT is None or _SESSION_CONTEXT.pid != os.getpid():
            return _bind_session_context_unlocked(
                profile_name=profile_name,
                remote=remote,
                host=host,
                identity=identity,
                repository=repository,
                org=org,
                role_kind=role_kind,
                expected_username=expected_username,
                source=source,
            )
        return _SESSION_CONTEXT.as_dict()


def assess_session_context(
    *,
    profile_name: str | None,
    remote: str | None,
    host: str | None = None,
    identity: str | None = None,
    repository: str | None = None,
    org: str | None = None,
    expected_username: str | None = None,
    require_bound: bool = False,
    require_complete: bool = False,
) -> dict[str, Any]:
    """Compare live values against the bound session context.

    Returns ``proven`` / ``block`` / ``reasons``. When unbound and
    ``require_bound`` is false, does not block (caller may seed). When
    unbound and ``require_bound`` is true, fails closed. ``require_complete``
    additionally fails closed when the binding carries no verified
    repository/organization identity, so a mutation can never run against an
    unknown repository (#714).
    """
    reasons: list[str] = []
    with _SESSION_CONTEXT_LOCK:
        bound = _SESSION_CONTEXT
        ctx = bound.as_dict() if bound is not None else None
    if ctx is None or ctx.get("pid") != os.getpid():
        if require_bound:
            reasons.append(
                "session mutation context is unbound; call gitea_whoami or "
                "gitea_activate_profile before mutating (fail closed)"
            )
            return _assessment(False, reasons, ctx)
        return _assessment(True, reasons, ctx)

    if require_complete and not (ctx.get("repository") and ctx.get("org")):
        reasons.append(
            "session repository/organization identity is unverified "
            f"(repository={ctx.get('repository')!r}, org={ctx.get('org')!r}); "
            "a mutation cannot proceed without a verified workspace "
            "repository (fail closed)"
        )
        return _assessment(False, reasons, ctx)

    live_profile = (profile_name or "").strip() or None
    live_remote = (remote or "").strip() or None
    live_host = (host or "").strip().lower() or None
    live_identity = (identity or "").strip() or None
    live_repo = (repository or "").strip() or None
    live_org = (org or "").strip() or None

    if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"):
        reasons.append(
            f"profile drift: live '{live_profile}' != bound "
            f"'{ctx.get('profile_name')}' (fail closed)"
        )
    if ctx.get("remote") and live_remote and live_remote != ctx.get("remote"):
        reasons.append(
            f"remote drift: live '{live_remote}' != bound "
            f"'{ctx.get('remote')}' (fail closed)"
        )
    if ctx.get("host") and live_host and live_host != ctx.get("host"):
        reasons.append(
            f"host drift: live '{live_host}' != bound "
            f"'{ctx.get('host')}' (fail closed)"
        )
    if (
        ctx.get("identity")
        and live_identity
        and live_identity != ctx.get("identity")
    ):
        reasons.append(
            f"identity drift: live '{live_identity}' != bound "
            f"'{ctx.get('identity')}' (fail closed)"
        )
    if ctx.get("repository") and live_repo and live_repo != ctx.get("repository"):
        reasons.append(
            f"repository drift: live '{live_repo}' != bound "
            f"'{ctx.get('repository')}' (fail closed)"
        )
    if ctx.get("org") and live_org and live_org != ctx.get("org"):
        reasons.append(
            f"org drift: live '{live_org}' != bound "
            f"'{ctx.get('org')}' (fail closed)"
        )

    expected = expected_username or ctx.get("expected_username")
    if expected and live_identity and live_identity != expected:
        reasons.append(
            f"identity mismatch: authenticated '{live_identity}' != "
            f"profile expected '{expected}' (fail closed)"
        )

    return _assessment(not reasons, reasons, ctx)


def assess_identity_match(
    *,
    authenticated: str | None,
    expected_username: str | None,
) -> dict[str, Any]:
    """Fail closed when profile declares a username that does not match live auth."""
    reasons: list[str] = []
    auth = (authenticated or "").strip() or None
    expected = (expected_username or "").strip() or None
    if expected and auth and auth != expected:
        reasons.append(
            f"identity mismatch: authenticated '{auth}' != "
            f"profile expected '{expected}' (fail closed)"
        )
    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "authenticated": auth,
        "expected_username": expected,
    }


_REPO_SLUG_RE = re.compile(r"^\s*(?P<org>[^/\s]+)\s*/\s*(?P<repo>[^/\s]+?)(?:\.git)?\s*$")


def parse_repository_slug(slug: str | None) -> tuple[str, str] | None:
    """Split a canonical ``owner/repository`` slug, or None when unparseable."""
    match = _REPO_SLUG_RE.match(slug or "")
    if not match:
        return None
    return match.group("org"), match.group("repo")


def format_repository_slug(org: str | None, repo: str | None) -> str | None:
    """``owner/repository`` from parts, or None when either side is missing."""
    left = (org or "").strip()
    right = (repo or "").strip()
    if not left or not right:
        return None
    return f"{left}/{right}"


def declared_allowed_repositories(
    profile: Mapping[str, Any] | None,
    *,
    strict: bool = False,
) -> list[str]:
    """Canonical ``owner/repository`` authorization boundary declared by *profile*.

    This list is an authorization boundary, never the session binding itself:
    the verified workspace selects exactly one entry (see
    :func:`assess_repository_scope`).

    When *strict* is true (mutation path), missing profile, non-list values, or
    any malformed entry raise ``ValueError`` instead of silently collapsing to
    an empty scope (which would fail open). Read-only diagnostics may use
    strict=False.
    """
    if not profile:
        if strict:
            raise ValueError(
                "profile unresolved; cannot declare allowed_repositories "
                "(fail closed)"
            )
        return []
    raw = profile.get("allowed_repositories")
    if raw is None:
        return []
    if not isinstance(raw, (list, tuple)):
        if strict:
            raise ValueError(
                "allowed_repositories must be a list of owner/repository slugs "
                "(fail closed)"
            )
        return []
    slugs: list[str] = []
    errors: list[str] = []
    for entry in raw:
        if not isinstance(entry, str):
            errors.append(f"non-string allowed_repositories entry {entry!r}")
            continue
        parsed = parse_repository_slug(entry)
        if not parsed:
            errors.append(
                f"malformed allowed_repositories entry {entry!r} "
                "(expected owner/repository)"
            )
            continue
        slugs.append(f"{parsed[0]}/{parsed[1]}")
    if strict and errors:
        raise ValueError("; ".join(errors) + " (fail closed)")
    return slugs


def assess_repository_scope(
    *,
    workspace_slug: str | None,
    allowed: list[str] | None,
    profile_name: str | None = None,
    require_scope: bool = False,
) -> dict[str, Any]:
    """Authorize the verified workspace repository against the profile allowlist.

    The workspace-derived slug is the only candidate: a profile that authorizes
    several repositories still binds to the single verified one, and never to
    the list as a whole.

    *require_scope* (mutation path): missing/empty allowlist fails closed. The
    workspace remote cannot self-authorize without a configured boundary.
    """
    scope = list(allowed or [])
    reasons: list[str] = []
    name = profile_name or "(active profile)"
    if not scope:
        if require_scope:
            reasons.append(
                f"mutation denied: profile '{name}' has no non-empty "
                "allowed_repositories authorization boundary (fail closed)"
            )
            return {
                "proven": False,
                "block": True,
                "reasons": reasons,
                "scope_enforced": True,
                "workspace_slug": workspace_slug,
                "allowed_repositories": [],
            }
        return {
            "proven": True,
            "block": False,
            "reasons": reasons,
            "scope_enforced": False,
            "workspace_slug": workspace_slug,
            "allowed_repositories": [],
        }
    if not workspace_slug:
        reasons.append(
            f"no verified workspace repository could be established for profile "
            f"'{name}'; repository scope cannot be authorized (fail closed)"
        )
    elif workspace_slug.lower() not in {entry.lower() for entry in scope}:
        reasons.append(
            f"repository scope denial: workspace repository '{workspace_slug}' "
            f"is not authorized by profile '{name}' allowed_repositories "
            f"{sorted(scope)} (fail closed)"
        )
    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "scope_enforced": True,
        "workspace_slug": workspace_slug,
        "allowed_repositories": sorted(scope),
    }


def assess_repository_override(
    *,
    requested_org: str | None,
    requested_repo: str | None,
    bound_org: str | None,
    bound_repo: str | None,
) -> dict[str, Any]:
    """Caller-supplied org/repo must agree with the immutable binding.

    A mutation request is never allowed to establish, complete, or replace the
    binding — it may only be checked against it.
    """
    reasons: list[str] = []
    req_org = (requested_org or "").strip() or None
    req_repo = (requested_repo or "").strip() or None
    if req_repo and bound_repo and req_repo.lower() != bound_repo.lower():
        reasons.append(
            f"repository override denial: request targets '{req_repo}' but the "
            f"session is bound to '{bound_repo}' (fail closed)"
        )
    if req_org and bound_org and req_org.lower() != bound_org.lower():
        reasons.append(
            f"organization override denial: request targets '{req_org}' but the "
            f"session is bound to '{bound_org}' (fail closed)"
        )
    return {"proven": not reasons, "block": bool(reasons), "reasons": reasons}


def filter_profiles_for_remote(
    config: dict | None,
    remote: str | None,
    remotes: dict | None,
) -> list[str]:
    """Profile names whose host/context matches *remote* (enabled only)."""
    if not config or not remote:
        return []
    profiles = config.get("profiles") or {}
    contexts = config.get("contexts") or {}
    names: list[str] = []
    for name, data in profiles.items():
        if not isinstance(data, dict):
            continue
        if not data.get("enabled", True):
            continue
        if profile_matches_remote(data, remote, remotes, contexts=contexts):
            names.append(name)
    return sorted(names)


def profile_allowed_for_remote(
    profile: dict | None,
    remote: str | None,
    remotes: dict | None,
    *,
    contexts: dict | None = None,
) -> dict[str, Any]:
    """Assess whether the active profile may serve *remote*."""
    reasons: list[str] = []
    if not profile:
        reasons.append("active profile unresolved (fail closed)")
        return {"proven": False, "block": True, "reasons": reasons}
    if not remote:
        return {"proven": True, "block": False, "reasons": reasons}
    # Legacy env-only profiles have no configured base URL or v2 context. Their
    # first call may establish the process remote/host pin; after that,
    # assess_session_context rejects any drift. Configured v2 profiles still
    # require positive host/context alignment here.
    if not profile_host(profile) and not (profile.get("context") or "").strip():
        return {"proven": True, "block": False, "reasons": reasons}
    if not profile_matches_remote(profile, remote, remotes, contexts=contexts):
        p_name = profile.get("profile_name") or profile.get("name") or "(unknown)"
        p_host = profile_host(profile) or "(none)"
        r_host = remote_host(remote, remotes) or "(none)"
        reasons.append(
            f"cross-host profile denial: profile '{p_name}' (host '{p_host}') "
            f"cannot serve remote '{remote}' (host '{r_host}') (fail closed)"
        )
    return {"proven": not reasons, "block": bool(reasons), "reasons": reasons}


def mutation_context_audit_fields(
    ctx: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Fields to include in pre-mutation audit records."""
    data = ctx if ctx is not None else get_session_context()
    if not data:
        return {
            "session_context_bound": False,
            "session_profile": None,
            "session_remote": None,
            "session_host": None,
            "session_identity": None,
            "session_repository": None,
            "session_org": None,
        }
    return {
        "session_context_bound": True,
        "session_profile": data.get("profile_name"),
        "session_remote": data.get("remote"),
        "session_host": data.get("host"),
        "session_identity": data.get("identity"),
        "session_repository": data.get("repository"),
        "session_org": data.get("org"),
        "session_role_kind": data.get("role_kind"),
        "session_context_source": data.get("source"),
    }


def _assessment(
    proven: bool, reasons: list[str], ctx: Mapping[str, Any] | None
) -> dict[str, Any]:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": list(reasons),
        "bound_context": dict(ctx) if ctx else None,
        "audit": mutation_context_audit_fields(ctx),
    }