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"
REQUESTED = "requested"
ACCEPTED = "accepted"
PENDING = "pending"

REDACTED = "[REDACTED]"

# A dict key containing any of these (case-insensitive) has its value redacted.
_SECRET_KEY_HINTS = (
    "token",
    "password",
    "passwd",
    "pwd",
    "secret",
    "authorization",
    "auth",
    "api_key",
    "apikey",
    "access_key",
    "private_key",
    "client_secret",
    "credential",
)
# A string value starting with one of these has the following run redacted.
# Space-terminated scheme prefixes only. Colon forms (``token:`` / ``api_key:``)
# and ``Authorization: Bearer …`` are handled by ``_ASSIGNMENT_SECRET_PATTERN``
# so policy/docs text that merely *names* a scheme is not itself flagged as a
# live secret by console detectors.
_SECRET_VALUE_PREFIXES = (
    "token ",
    "Basic ",
    "Bearer ",
)

# Bare token-shaped values only — never a broad ``sec-`` prefix that erases
# ordinary words (#664 B8 over-redaction).
_BARE_SECRET_PATTERN = re.compile(
    r'(?i)\b(?:'
    r'ghp_[A-Za-z0-9_]{16,}'
    r'|gho_[A-Za-z0-9_]{16,}'
    r'|ghu_[A-Za-z0-9_]{16,}'
    r'|ghs_[A-Za-z0-9_]{16,}'
    r'|ghr_[A-Za-z0-9_]{16,}'
    r'|sk-live-[A-Za-z0-9_-]{16,}'
    r'|sk-proj-[A-Za-z0-9_-]{16,}'
    r'|sk-[A-Za-z0-9_-]{20,}'
    r'|glpat-[A-Za-z0-9_-]{16,}'
    r')\b'
)

# Key/value credentials embedded in free text (password=..., api_key: ..., GITEA_TOKEN=..., etc.).
# Group 1 captures the key name (e.g. GITEA_TOKEN, password, api_key).
# Group 2 captures delimiter/whitespace (=, : ).
# Group 3 captures the secret value, stopping at whitespace or non-secret delimiters (&, ;, ,, quotes, closing brackets).
_ASSIGNMENT_SECRET_PATTERN = re.compile(
    r'(?i)\b([A-Za-z0-9_]*?(?:token|password|passwd|pwd|secret|api[_-]?key|access[_-]?key|'
    r'client[_-]?secret|private[_-]?key|authorization|credential))\b(\s*[:=]\s*)('
    r'"[^"]*"|\'[^\']*\'|'
    r'(?:Bearer|Basic|Token)\s+[^\s;&,"\'\)\}\]\>]+|'
    r'[^\s;&,"\'\)\}\]\>]+'
    r')'
)

# Connection-string style credentials: Password=...; User ID=...; etc.
_CONN_STRING_SECRET_PATTERN = re.compile(
    r'(?i)\b((?:password|pwd|user\s*id|uid|username|account)\s*=\s*)([^\s;\'"]+)'
)


# 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

    # Match any URI scheme (http, https, postgres, mysql, mongodb, redis, etc.)
    url_pattern = re.compile(r'([a-z0-9\+\.\-]+://[^\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 or (parsed.username or parsed.password) or parsed.scheme.lower() not in ("http", "https"):
                # Rebuild URL to redact any credentials or query secrets
                new_netloc = parsed.netloc
                if parsed.username or parsed.password:
                    netloc_clean = parsed.hostname or ""
                    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 _mask_assignment(match: re.Match) -> str:
    """Keep the key and separator; replace only the secret value."""
    val = match.group(3)
    if val.startswith(REDACTED) or val.startswith("%5BREDACTED") or val.startswith("[REDACTED"):
        return f"{match.group(1)}{match.group(2)}{val}"
    return f"{match.group(1)}{match.group(2)}{REDACTED}"


def _mask_conn_secret(match: re.Match) -> str:
    """Keep the connection-string key; replace only the credential value."""
    val = match.group(2)
    if val.startswith(REDACTED) or val.startswith("%5BREDACTED") or val.startswith("[REDACTED"):
        return f"{match.group(1)}{val}"
    return f"{match.group(1)}{REDACTED}"


def _redact_str(text):
    """Redact credentials, bare token shapes, and raw URLs in *text* (#664 B8).

    Covers key/value credentials, authorization/bearer material, bare
    token-shaped values, connection-string credentials, and secrets embedded
    in larger sentences. Deliberately does **not** erase ordinary words that
    merely begin with a broad ``sec-`` prefix.
    """
    if not isinstance(text, str) or not text:
        return text
    out = redact_urls(text)
    out = _BARE_SECRET_PATTERN.sub(REDACTED, out)
    out = _ASSIGNMENT_SECRET_PATTERN.sub(_mask_assignment, out)
    out = _CONN_STRING_SECRET_PATTERN.sub(_mask_conn_secret, out)
    out_lower = out.lower()
    for prefix in _SECRET_VALUE_PREFIXES:
        prefix_lower = prefix.lower()
        idx = 0
        while True:
            i = out_lower.find(prefix_lower, 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:]
            out_lower = out.lower()
            idx = i + len(prefix) + len(REDACTED)
    return 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