Gitea-Tools/gitea_audit.py

"""Audit logging for Gitea MCP mutating actions (issue #18).

Emits one structured JSON record per mutating action so an operator can see
*which execution profile and authenticated Gitea user* performed (or was
blocked from / failed) each mutation.

Design constraints:

- **Off by default.** Records are written only when ``GITEA_AUDIT_LOG`` names a
  file path. With it unset, ``write_event`` is a no-op and callers add zero
  behaviour — existing tool behaviour and API call patterns are unchanged.
- **Never raises.** Auditing must never break the action it records; all sink
  I/O is best-effort and swallows errors.
- **No secrets.** Tokens / Authorization material are redacted from request
  metadata and reason strings before a record is written.
- **No network.** This module performs no HTTP; the caller supplies identity
  and profile metadata it already resolved.
"""
import os
import json
import datetime
import re
import urllib.parse

# Result states for an audited action.
ALLOWED = "allowed"
BLOCKED = "blocked"
FAILED = "failed"
SUCCEEDED = "succeeded"

REDACTED = "[REDACTED]"

# A dict key containing any of these (case-insensitive) has its value redacted.
_SECRET_KEY_HINTS = ("token", "password", "secret", "authorization", "auth")
# A string value starting with one of these has the following run redacted.
_SECRET_VALUE_PREFIXES = ("token ", "Basic ", "Bearer ")

# Known synthetic test-only domains/hostnames to preserve
_SYNTHETIC_HOSTS = {
    "example.com",
    "example.test",
    "example.invalid",
    "internal.example",
    "localhost",
    "gitea.example.com",
    "x"
}

# Known real service hostnames to redact even if not part of a full URL
_REAL_HOSTS = {"gitea.prgs.cc", "gitea.dadeschools.net"}


def redact_urls(text: str) -> str:
    """Redact raw URLs, query-string secrets, and URL credentials.

    Synthetic example/test-only domains are preserved to keep test fixtures functional
    except for any embedded credentials or query parameters containing secrets.
    """
    if not isinstance(text, str) or not text:
        return text

    url_pattern = re.compile(r'(https?://[^\s)>\]}]+)', re.IGNORECASE)

    def replace_url(match):
        url_str = match.group(1)
        try:
            parsed = urllib.parse.urlsplit(url_str)
            host = (parsed.hostname or "").lower()

            is_synthetic = False
            for sh in _SYNTHETIC_HOSTS:
                if host == sh or host.endswith("." + sh):
                    is_synthetic = True
                    break

            if is_synthetic:
                # Rebuild synthetic URL to redact any credentials or query secrets
                new_netloc = parsed.netloc
                if parsed.username or parsed.password:
                    netloc_clean = parsed.hostname
                    if parsed.port:
                        netloc_clean = f"{netloc_clean}:{parsed.port}"
                    new_netloc = f"[REDACTED_USER]:[REDACTED_PASS]@{netloc_clean}"

                new_query = parsed.query
                if parsed.query:
                    params = urllib.parse.parse_qsl(parsed.query)
                    clean_params = []
                    for k, v in params:
                        if any(hint in k.lower() for hint in ("token", "password", "secret", "auth", "key")):
                            clean_params.append((k, "[REDACTED]"))
                        else:
                            clean_params.append((k, v))
                    new_query = urllib.parse.urlencode(clean_params)

                return urllib.parse.urlunsplit((
                    parsed.scheme,
                    new_netloc,
                    parsed.path,
                    new_query,
                    parsed.fragment
                ))
            else:
                return "[REDACTED_URL]"
        except Exception:
            return "[REDACTED_URL]"

    out = url_pattern.sub(replace_url, text)

    # Redact raw occurrences of real service hostnames afterward
    for host in _REAL_HOSTS:
        out = out.replace(host, "[REDACTED_HOST]")

    return out


def _redact_str(text):
    """Redact anything that looks like an Authorization credential or raw URL in *text*."""
    if not isinstance(text, str) or not text:
        return text
    out = text
    for prefix in _SECRET_VALUE_PREFIXES:
        idx = 0
        while True:
            i = out.find(prefix, idx)
            if i == -1:
                break
            j = i + len(prefix)
            while j < len(out) and not out[j].isspace():
                j += 1
            out = out[:i] + prefix + REDACTED + out[j:]
            idx = i + len(prefix) + len(REDACTED)
    return redact_urls(out)


def redact(value):
    """Recursively redact secret-looking keys/values from a JSON-able value."""
    if isinstance(value, dict):
        result = {}
        for k, v in value.items():
            if isinstance(k, str) and any(h in k.lower() for h in _SECRET_KEY_HINTS):
                result[k] = REDACTED
            else:
                result[k] = redact(v)
        return result
    if isinstance(value, (list, tuple)):
        return [redact(v) for v in value]
    if isinstance(value, str):
        return _redact_str(value)
    return value


def audit_log_path():
    """Return the configured audit log file path, or None if auditing is off."""
    return (os.environ.get("GITEA_AUDIT_LOG") or "").strip() or None


def audit_enabled():
    """True when a sink is configured. When False, callers should skip auditing."""
    return audit_log_path() is not None


def build_event(*, action, result, remote=None, server=None, repository=None,
                issue_number=None, pr_number=None, profile_name=None,
                audit_label=None, authenticated_username=None, target_branch=None,
                head_sha=None, reason=None, request_metadata=None, now=None,
                mcp_namespace=None, task_role=None, operation=None):
    """Build a redacted, JSON-able audit record for a mutating action."""
    ts = now or datetime.datetime.now(datetime.timezone.utc)
    if isinstance(ts, datetime.datetime):
        ts = ts.isoformat()
    event = {
        "timestamp": ts,
        "action": action,
        "action_type": "mutating",
        "result": result,
        "remote": remote,
        "server": server,
        "repository": repository,
        "issue_number": issue_number,
        "pr_number": pr_number,
        "profile_name": profile_name,
        "audit_label": audit_label,
        "authenticated_username": authenticated_username,
        "target_branch": target_branch,
        "head_sha": head_sha,
        "reason": _redact_str(reason) if reason else reason,
        "request_metadata": redact(request_metadata) if request_metadata is not None else None,
    }
    if mcp_namespace is not None:
        event["mcp_namespace"] = mcp_namespace
    if task_role is not None:
        event["task_role"] = task_role
    if operation is not None:
        event["operation"] = operation
    return event


def write_event(event, path=None):
    """Append *event* as one JSON line to the audit sink. Never raises.

    Returns True if a line was written, False if auditing is off or the write
    failed (auditing is best-effort and must not break the caller).
    """
    path = path or audit_log_path()
    if not path:
        return False
    try:
        line = json.dumps(event, default=str, sort_keys=True)
        with open(path, "a", encoding="utf-8") as fh:
            fh.write(line + "\n")
        return True
    except Exception:
        return False