Gitea-Tools/sentry_observability.py

"""Optional self-hosted Sentry observability for the Gitea MCP server (#606).

Adds env-var-gated Sentry SDK instrumentation so runtime errors, fail-closed
workflow blockers, lease/terminal-lock/stale-runtime collisions, and recurring
watchdog check-ins are visible in a *self-hosted* Sentry at
``https://sentry.prgs.cc/`` — never Sentry Cloud, and never as the workflow
source of truth (Gitea stays canonical).

Design constraints (mirror ``gitea_audit`` and the #612 incident bridge):

- **Off by default.** With ``MCP_SENTRY_ENABLED`` false/unset *or* ``SENTRY_DSN``
  empty, ``init_sentry`` is a no-op and no events are ever sent — existing tool
  behaviour and API-call patterns are unchanged (acceptance criterion 1).
- **Fail *open* for observability.** A Sentry outage, a missing ``sentry_sdk``
  package, or any capture error must never break an MCP tool success path. Every
  public entry point swallows its own exceptions.
- **Fail *closed* for redaction.** If a field cannot be proven safe it is dropped
  rather than sent. Tokens, passwords, keychain IDs, DSNs, private config, raw
  session-state, full prompt bodies, and full filesystem paths never leave here.
- **No hard dependency.** ``sentry_sdk`` is imported lazily; the module is fully
  importable and testable without it installed.

Sentry is observe-only: it must not approve, merge, close, or otherwise mutate
Gitea workflow state, nor bypass leases, #332, or MCP gates. Alerts may only feed
the sanctioned Gitea issue/comment path via the #612 incident bridge.
"""

from __future__ import annotations

import hashlib
import os
import re
from dataclasses import dataclass
from typing import Any

# Reuse the most comprehensive existing scrubber so redaction stays consistent
# with the #612 incident bridge (tokens, DSNs, cookies, bearer/basic, keychain
# ids, session ids, user:pass@host).
from incident_bridge import redact_text as _redact_text

# Second, complementary scrubber: catches bare ``token <value>`` /
# ``Bearer <value>`` / ``Basic <value>`` prefixes and raw URLs that the
# incident-bridge delimiter patterns miss.
from gitea_audit import _redact_str as _redact_prefixes

# ── Optional SDK (lazy, never a hard dependency) ────────────────────────────
try:  # pragma: no cover - trivial import guard
    import sentry_sdk  # type: ignore
except Exception:  # pragma: no cover - absence is a supported state
    sentry_sdk = None  # type: ignore


# ── Env var names (single source of truth) ──────────────────────────────────
ENV_ENABLED = "MCP_SENTRY_ENABLED"
ENV_DSN = "SENTRY_DSN"
ENV_ENVIRONMENT = "SENTRY_ENVIRONMENT"
ENV_RELEASE = "SENTRY_RELEASE"
ENV_TRACES_SAMPLE_RATE = "MCP_SENTRY_TRACES_SAMPLE_RATE"
ENV_ENABLE_LOGS = "MCP_SENTRY_ENABLE_LOGS"

_TRUTHY = frozenset({"1", "true", "yes", "on"})

REDACTED = "[REDACTED]"
REDACTED_PATH = "[REDACTED_PATH]"


# ── Cron / watchdog monitor slugs (acceptance criterion 6) ──────────────────
# Stable slugs for the recurring/watchdog jobs #606 wants check-ins for. The
# slug is the durable monitor identity in Sentry; the wiring call sites pass one
# of these keys (or an explicit slug) to ``monitor_checkin``.
MONITOR_SLUGS: dict[str, str] = {
    "stale_lease_scan": "gitea-mcp-stale-lease-scan",
    "terminal_lock_scan": "gitea-mcp-terminal-lock-scan",
    "allocator_health": "gitea-mcp-allocator-health",
    "namespace_health": "gitea-mcp-namespace-health",
    "dashboard_freshness": "gitea-mcp-dashboard-freshness",
    "reconciler_cleanup": "gitea-mcp-reconciler-cleanup",
}

_CHECKIN_STATUSES = frozenset({"in_progress", "ok", "error"})


# ── Tag allowlist (issue "Suggested Sentry tags/context") ───────────────────
# Only these keys are ever attached as Sentry tags. Anything else is dropped so
# a caller cannot accidentally leak a sensitive value through a tag.
ALLOWED_TAG_KEYS = frozenset({
    "role",
    "profile",
    "namespace",
    "repo",
    "org",
    "issue_number",
    "pr_number",
    "blocker_type",
    "workflow_hash",
    "session_id_hash",  # hash only — never the raw session id
    "pid",
    "worktree_category",  # category, never the full sensitive path
    "lease_comment_id",
    "expected_head_sha",
    "current_head_sha",
    "terminal_lock_state",
    "capability",
    "mutation_tool",
})

# Absolute-path shapes that must never be sent verbatim (macOS/Linux + temp).
_PATH_RE = re.compile(r"(?:/private)?/(?:Users|home|tmp|var|opt|Volumes)/[^\s\"']*")

# ``extra`` keys whose *full* contents are forbidden by the redaction rules
# (raw session-state, full prompt/comment bodies, private config blobs, raw
# headers).
_FORBIDDEN_EXTRA_KEYS = frozenset({
    "prompt",
    "prompt_body",
    "next_prompt",
    "body",
    "raw_body",
    "session_state",
    "session_state_contents",
    "config",
    "config_contents",
    "private_config",
    "headers",
    "authorization",
})


# ── Configuration ───────────────────────────────────────────────────────────
@dataclass(frozen=True)
class SentryConfig:
    """Immutable snapshot of the Sentry env configuration."""

    enabled: bool = False
    dsn: str | None = None
    environment: str = "development"
    release: str | None = None
    traces_sample_rate: float = 0.0
    enable_logs: bool = False

    @property
    def active(self) -> bool:
        """True only when the operator both opted in *and* supplied a DSN.

        This is the single gate that keeps the feature off by default: enabling
        the flag without a DSN (or vice versa) sends nothing.
        """
        return bool(self.enabled and self.dsn)

    def safe_summary(self) -> dict[str, Any]:
        """Operator-facing status with **no** DSN value (only presence)."""
        return {
            "enabled": self.enabled,
            "dsn_present": bool(self.dsn),
            "environment": self.environment,
            "release": self.release,
            "traces_sample_rate": self.traces_sample_rate,
            "enable_logs": self.enable_logs,
            "active": self.active,
        }


def _env_bool(name: str, env: dict[str, str]) -> bool:
    return (env.get(name) or "").strip().lower() in _TRUTHY


def _env_float(name: str, default: float, env: dict[str, str]) -> float:
    raw = (env.get(name) or "").strip()
    if not raw:
        return default
    try:
        val = float(raw)
    except (TypeError, ValueError):
        return default
    # Clamp to Sentry's valid [0.0, 1.0] sample-rate range.
    if val < 0.0:
        return 0.0
    if val > 1.0:
        return 1.0
    return val


def load_config(env: dict[str, str] | None = None) -> SentryConfig:
    """Build a :class:`SentryConfig` from the environment (read at call time)."""
    env = dict(os.environ if env is None else env)
    dsn = (env.get(ENV_DSN) or "").strip() or None
    return SentryConfig(
        enabled=_env_bool(ENV_ENABLED, env),
        dsn=dsn,
        environment=(env.get(ENV_ENVIRONMENT) or "").strip() or "development",
        release=(env.get(ENV_RELEASE) or "").strip() or None,
        traces_sample_rate=_env_float(ENV_TRACES_SAMPLE_RATE, 0.0, env),
        enable_logs=_env_bool(ENV_ENABLE_LOGS, env),
    )


def sdk_available() -> bool:
    """True when the optional ``sentry_sdk`` package is importable."""
    return sentry_sdk is not None


# ── Redaction (fail closed) ─────────────────────────────────────────────────
def sanitize_path(value: Any) -> str:
    """Reduce a filesystem path to a non-sensitive *category* token.

    Full local paths must never be sent. We keep only a coarse worktree
    category derived from the path shape (author/reviewer/merger/reconciler/
    branches/root/other).
    """
    text = "" if value is None else str(value)
    low = text.lower()
    if not text:
        return "unknown"
    # Order matters: more specific role markers before the generic "branches".
    if "reconcile" in low:
        return "reconciler"
    if "review" in low:
        return "reviewer"
    if "merge" in low or "merger" in low:
        return "merger"
    if "author" in low or re.search(r"/branches/(?:feat|fix|docs|chore|issue)", low):
        return "author"
    if "/branches/" in low:
        return "branches"
    if low.rstrip("/").endswith("gitea-tools"):
        return "root"
    return "other"


def redact_value(value: Any) -> Any:
    """Recursively redact a JSON-able value: secret text, absolute paths, and
    known-sensitive dict keys are removed. Fail closed — any error drops the
    value entirely rather than risk leaking it."""
    try:
        if isinstance(value, dict):
            out: dict[str, Any] = {}
            for k, v in value.items():
                key = str(k)
                low = key.lower()
                if low in _FORBIDDEN_EXTRA_KEYS or any(
                    s in low
                    for s in ("token", "secret", "password", "cookie", "auth", "dsn", "keychain")
                ):
                    out[key] = REDACTED
                    continue
                out[key] = redact_value(v)
            return out
        if isinstance(value, (list, tuple)):
            return [redact_value(v) for v in value]
        if isinstance(value, str):
            scrubbed = _redact_text(value)
            scrubbed = _redact_prefixes(scrubbed)
            scrubbed = _PATH_RE.sub(REDACTED_PATH, scrubbed)
            return scrubbed
        return value
    except Exception:
        return REDACTED


def hash_session_id(session_id: Any) -> str:
    """Short, stable, non-reversible fingerprint of a session id."""
    digest = hashlib.sha256(str(session_id).encode("utf-8", "replace")).hexdigest()
    return digest[:12]


def build_tags(**kwargs: Any) -> dict[str, str]:
    """Return a scrubbed, allowlisted tag dict.

    ``session_id`` is accepted but only ever surfaced as ``session_id_hash``.
    ``worktree_path`` collapses to ``worktree_category``. Any non-allowlisted
    key, or a value that still contains redacted material after scrubbing, is
    dropped.
    """
    raw: dict[str, Any] = dict(kwargs)

    # Hash the session id — never emit it raw.
    session_id = raw.pop("session_id", None)
    if session_id and "session_id_hash" not in raw:
        raw["session_id_hash"] = hash_session_id(session_id)

    # A full worktree path collapses to a category tag.
    wt = raw.pop("worktree_path", None)
    if wt and "worktree_category" not in raw:
        raw["worktree_category"] = sanitize_path(wt)

    out: dict[str, str] = {}
    for key, val in raw.items():
        if key not in ALLOWED_TAG_KEYS:
            continue
        if val is None:
            continue
        scrubbed = redact_value(val)
        text = str(scrubbed)
        if not text or REDACTED in text or REDACTED_PATH in text:
            continue
        if len(text) > 200:
            text = text[:200] + "…"
        out[key] = text
    return out


def scrub_event(event: Any, hint: Any = None) -> dict[str, Any] | None:
    """Sentry ``before_send`` / ``before_send_log`` hook.

    Recursively redacts the outgoing event. On *any* failure it returns ``None``
    so the event is dropped rather than sent unscrubbed (fail closed for
    redaction).
    """
    try:
        if not isinstance(event, dict):
            return None
        scrubbed = redact_value(event)
        # Drop server_name if it leaked a hostname/path; PID is kept via tags.
        scrubbed.pop("server_name", None)
        return scrubbed
    except Exception:
        return None


# ── Event builders (pure, independently testable) ───────────────────────────
def build_blocker_event(
    blocker_type: str,
    *,
    message: str | None = None,
    next_action: str | None = None,
    level: str = "warning",
    tags: dict[str, Any] | None = None,
    extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a redacted, structured Sentry event for a workflow blocker.

    ``next_action`` maps the issue's "canonical next action when available"
    requirement (acceptance criterion 7).
    """
    merged_tags = dict(tags or {})
    merged_tags.setdefault("blocker_type", blocker_type)
    safe_tags = build_tags(**merged_tags)

    safe_extra = redact_value(dict(extra or {}))
    if next_action:
        # A short canonical next action is allowed (it is not a full prompt).
        safe_extra["canonical_next_action"] = redact_value(str(next_action)[:500])

    event: dict[str, Any] = {
        "message": redact_value(message or blocker_type),
        "level": level if level in ("debug", "info", "warning", "error", "fatal") else "warning",
        "logger": "gitea-mcp.workflow",
        "tags": safe_tags,
        "extra": safe_extra,
        "fingerprint": ["workflow-blocker", blocker_type],
    }
    return event


def build_checkin_payload(
    monitor: str,
    status: str,
    *,
    check_in_id: str | None = None,
    duration: float | None = None,
) -> dict[str, Any]:
    """Build a Sentry cron check-in payload for one of :data:`MONITOR_SLUGS`.

    ``monitor`` may be a registry key (e.g. ``"stale_lease_scan"``) or an
    explicit slug. Raises ``ValueError`` on an unknown status so callers cannot
    silently send a malformed check-in.
    """
    if status not in _CHECKIN_STATUSES:
        raise ValueError(
            f"invalid check-in status {status!r}; expected one of {sorted(_CHECKIN_STATUSES)}"
        )
    slug = MONITOR_SLUGS.get(monitor, monitor)
    payload: dict[str, Any] = {"monitor_slug": slug, "status": status}
    if check_in_id:
        payload["check_in_id"] = str(check_in_id)
    if duration is not None:
        try:
            payload["duration"] = float(duration)
        except (TypeError, ValueError):
            pass
    return payload


# ── Runtime init + capture (fail open) ──────────────────────────────────────
_STATE: dict[str, Any] = {"initialized": False, "config": None}


def is_initialized() -> bool:
    return bool(_STATE.get("initialized"))


def active_config() -> SentryConfig | None:
    return _STATE.get("config")


def reset_for_tests() -> None:
    """Clear module init state. Test-only helper (never called in production)."""
    _STATE["initialized"] = False
    _STATE["config"] = None


def init_sentry(config: SentryConfig | None = None) -> dict[str, Any]:
    """Initialise the Sentry SDK if (and only if) enabled + DSN + SDK present.

    Idempotent and never raises. Returns an operator-safe status dict (no DSN
    value). Behaviour is unchanged when the feature is off.
    """
    cfg = config or load_config()
    status: dict[str, Any] = {"initialized": False, **cfg.safe_summary()}
    try:
        if not cfg.active:
            status["reason"] = "disabled (MCP_SENTRY_ENABLED false or SENTRY_DSN empty)"
            _STATE["config"] = cfg
            return status
        if not sdk_available():
            status["reason"] = "sentry_sdk not installed"
            _STATE["config"] = cfg
            return status

        init_kwargs: dict[str, Any] = {
            "dsn": cfg.dsn,
            "environment": cfg.environment,
            "release": cfg.release,
            "traces_sample_rate": cfg.traces_sample_rate,
            "before_send": scrub_event,
            "send_default_pii": False,
        }
        if cfg.enable_logs:
            # sentry-sdk 2.x captures Python logs as structured logs when the
            # experimental logs feature is enabled; scrub those too.
            init_kwargs["_experiments"] = {
                "enable_logs": True,
                "before_send_log": scrub_event,
            }
        sentry_sdk.init(**init_kwargs)  # type: ignore[union-attr]
        _STATE["initialized"] = True
        _STATE["config"] = cfg
        status["initialized"] = True
        status["reason"] = "sentry initialised"
    except Exception as exc:  # fail open: observability must not block startup
        status["reason"] = f"init failed (ignored): {type(exc).__name__}"
        _STATE["initialized"] = False
    return status


def _set_scope_tags(scope: Any, tags: dict[str, str]) -> None:
    for key, val in tags.items():
        try:
            scope.set_tag(key, val)
        except Exception:
            pass


def capture_workflow_blocker(
    blocker_type: str,
    *,
    message: str | None = None,
    next_action: str | None = None,
    level: str = "warning",
    tags: dict[str, Any] | None = None,
    extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Capture a fail-closed workflow blocker as a structured Sentry event.

    Always returns the redacted event dict (so callers/tests can inspect it),
    and sends it to Sentry only when initialised. Fail open.
    """
    event = build_blocker_event(
        blocker_type,
        message=message,
        next_action=next_action,
        level=level,
        tags=tags,
        extra=extra,
    )
    try:
        if is_initialized() and sdk_available():
            sentry_sdk.capture_event(event)  # type: ignore[union-attr]
    except Exception:
        pass
    return event


def capture_exception(
    exc: BaseException,
    *,
    tags: dict[str, Any] | None = None,
    extra: dict[str, Any] | None = None,
) -> bool:
    """Capture a runtime exception with scrubbed tags. Fail open.

    Returns True only when the event was handed to an initialised SDK.
    """
    try:
        if not (is_initialized() and sdk_available()):
            return False
        safe_tags = build_tags(**(tags or {}))
        safe_extra = redact_value(dict(extra or {}))
        with sentry_sdk.push_scope() as scope:  # type: ignore[union-attr]
            _set_scope_tags(scope, safe_tags)
            for key, val in safe_extra.items():
                try:
                    scope.set_extra(key, val)
                except Exception:
                    pass
            sentry_sdk.capture_exception(exc)  # type: ignore[union-attr]
        return True
    except Exception:
        return False


def monitor_checkin(
    monitor: str,
    status: str,
    *,
    check_in_id: str | None = None,
    duration: float | None = None,
) -> dict[str, Any] | None:
    """Send a Sentry cron check-in for a watchdog job. Fail open.

    Returns the payload (for inspection/tests), or ``None`` if the status was
    invalid. Only transmits when initialised.
    """
    try:
        payload = build_checkin_payload(
            monitor, status, check_in_id=check_in_id, duration=duration
        )
    except ValueError:
        return None
    try:
        if is_initialized() and sdk_available() and hasattr(sentry_sdk, "capture_checkin"):
            sentry_sdk.capture_checkin(**payload)  # type: ignore[union-attr]
    except Exception:
        pass
    return payload