gendesign/ops/glitchtip-auth-forwarder/forwarder.py

"""
Caddy access log -> GlitchTip auth event forwarder.

Tails /var/log/caddy/auth_audit.log (CADDY_LOG_FILE env), фильтрует JSON lines с
status==401, шлёт в GlitchTip через Sentry SDK как warning event с тэгами.

Persistent offset в /state/offset.json — не дублируем при restart.
Throttle: при >10 401 events за 60s — однократный digest event
(чтобы не флудить GlitchTip storm'ом); индивидуальные events во время storm пропускаются.

before_send=_drop_basic_auth_noise (glitchtip-noise фикс): все события отсюда
дропаются перед отправкой в GlitchTip — 401 от неаутентифицированного запроса
не ошибка сервиса, это боты сканируют закрытый basic_auth'ом сайт. Раньше это
был крупнейший источник шума в трекере (3 738 issue). Скрипт по-прежнему тэйлит
лог и печатает `[forwarder] 401 event sent: ...` в stdout (docker logs) — просто
больше не шлёт эти события в issue-трекер. Смотри `_drop_basic_auth_noise` docstring.

Реальный Caddy JSON access log (v2) структура:
{
  "level": "info",
  "ts": 1779524882.734,
  "logger": "http.log.access.log0",
  "msg": "handled request",
  "request": {
    "remote_ip": "95.165.147.218",
    "remote_port": "50796",
    "client_ip": "95.165.147.218",
    "proto": "HTTP/2.0",
    "method": "GET",
    "host": "gendsgn.ru",
    "uri": "/",
    "headers": {
      "User-Agent": ["Mozilla/5.0 ..."],
      "Authorization": ["Basic dXNlcjpwYXNz"],  # raw — при включённом log_credentials в Caddyfile
      ...
    },
    "tls": {...}
  },
  "user_id": "",
  "duration": 0.000134,
  "size": 0,
  "status": 401,
  "resp_headers": {...}
}
"""

import base64
import json
import os
import signal
import sys
import tempfile
import time
from collections import deque
from datetime import datetime, timezone
from pathlib import Path

import sentry_sdk

LOG_FILE = Path(os.getenv("CADDY_LOG_FILE", "/var/log/caddy/gendsgn.ru.log"))
STATE_FILE = Path(os.getenv("STATE_FILE", "/state/offset.json"))
DSN = os.environ["GLITCHTIP_DSN"]  # required, fail-fast
ENV = os.getenv("APP_ENV", "production")
POLL_INTERVAL_S = 2
THROTTLE_WINDOW_S = 60
THROTTLE_THRESHOLD = 10  # >10 events за 60s -> digest mode

# Известные боты которые получают 401 от Uptime мониторов —
# логируем их только в digest, не как individual events.
KNOWN_MONITOR_UAS = frozenset(
    [
        "GlitchTip/",
        "SentryUptimeBot/",
    ]
)

_shutdown = False
# Throttle для unhandled exceptions — не более одного Sentry event за 5 минут,
# чтобы persistent ошибка (напр. PermissionError) не спамила GlitchTip.
_last_exc_sent: float = 0.0
_EXC_THROTTLE_S: float = 300.0

# event_type-теги, которыми emit_event/emit_digest помечают КАЖДОЕ отправляемое
# событие (см. scope.set_tag("event_type", ...) ниже) — используются как ключ
# для before_send-фильтра.
_BASIC_AUTH_EVENT_TYPES = frozenset({"basic_auth_failed", "basic_auth_storm"})


def _drop_basic_auth_noise(event: dict, hint: dict) -> dict | None:  # type: ignore[type-arg]
    """before_send-фильтр: 401 неаутентифицированного basic_auth-запроса — НЕ
    ошибка сервиса, а expected-поведение сканеров-ботов, ломящихся в закрытый
    basic_auth'ом gendsgn.ru (`GET /wp-admin/install.php` и подобное). До этого
    фикса emit_event/emit_digest слали КАЖДЫЙ такой 401 individual-событием (или
    storm-digest) в GlitchTip — remote_ip в message/тегах раздувал кардинальность
    (3 738 issue, 2 019 различных заголовков, топ — 222 события на «GET
    /wp-admin/install.p…»), топя содержательные алерты (OperationalError, sweep
    failures) в шуме сканеров.

    Дропаем НА ИСТОЧНИКЕ (before_send), не постфактум-чисткой issue-трекера —
    так шум не появляется вообще, а не изредка удаляется руками. Фильтруем по
    тегу `event_type`, который ставят ТОЛЬКО emit_event/emit_digest — необработанные
    исключения самого форвардера (`capture_exception` в конце `main()`, реальный
    баг скрипта) этот тег не несут и проходят фильтр как есть (см. `except
    Exception` ниже в `main()`).
    """
    tags = event.get("tags")
    event_type = None
    if isinstance(tags, dict):
        event_type = tags.get("event_type")
    elif isinstance(tags, list):
        # sentry_sdk в некоторых версиях сериализует tags как list[tuple[str, str]]
        # вместо dict — на всякий случай поддерживаем обе формы.
        event_type = dict(tags).get("event_type") if tags else None
    if event_type in _BASIC_AUTH_EVENT_TYPES:
        return None
    return event


def _signal_handler(signum: int, frame: object) -> None:
    global _shutdown
    _shutdown = True


def load_offset() -> int:
    if STATE_FILE.exists():
        try:
            data = json.loads(STATE_FILE.read_text())
            return int(data.get("offset", 0))
        except Exception:
            pass
    return 0


def save_offset(offset: int) -> None:
    """Atomic write через temp file + rename — избегаем corrupt state при kill."""
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(
        {"offset": offset, "updated_at": datetime.now(timezone.utc).isoformat()}
    )
    # Пишем в temp рядом с целевым файлом (тот же volume -> атомарный rename)
    fd, tmp_path = tempfile.mkstemp(dir=STATE_FILE.parent, prefix=".offset_tmp_")
    try:
        with os.fdopen(fd, "w") as f:
            f.write(payload)
        os.replace(tmp_path, STATE_FILE)
    except Exception:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise


def is_monitor_ua(ua: str) -> bool:
    """Возвращает True если User-Agent принадлежит известному монитору."""
    return any(ua.startswith(prefix) for prefix in KNOWN_MONITOR_UAS)


def _extract_attempted_username(entry: dict) -> str | None:  # type: ignore[type-arg]
    """
    Извлекает username из Authorization header (Basic auth) из Caddy access log.

    Требования к Caddyfile:
    - global option `log_credentials` (иначе Caddy редактирует Authorization → "REDACTED")
    - этот forwarder ожидает auth_audit.log как primary source (CADDY_LOG_FILE env)

    Возвращает только username (всё до ':' в base64-decoded auth).
    Password нигде не сохраняется, не логируется.

    None если: header отсутствует, не Basic, base64 decode fail, нет ':', > 128 chars.
    """
    req = entry.get("request") or {}
    headers = req.get("headers") or {}
    auth_values = headers.get("Authorization") or headers.get("authorization")
    if not auth_values:
        return None
    if isinstance(auth_values, list):
        if not auth_values:
            return None
        auth = auth_values[0]
    else:
        auth = auth_values
    if not isinstance(auth, str) or not auth.lower().startswith("basic "):
        return None
    try:
        decoded = base64.b64decode(auth[6:], validate=True).decode("utf-8", errors="replace")
    except Exception:
        return None
    if ":" not in decoded:
        return None
    username = decoded.split(":", 1)[0]
    # Sanity check: только разумная длина
    if not username or len(username) > 128:
        return None
    return username


def emit_event(entry: dict) -> None:  # type: ignore[type-arg]
    """Отправляет одиночный 401 event в GlitchTip."""
    req = entry.get("request", {})
    path = req.get("uri", "?")
    method = req.get("method", "?")
    # client_ip — реальный IP (за прокси); remote_ip — подключающийся (docker network)
    remote_ip = req.get("client_ip") or req.get("remote_ip", "?")
    host = req.get("host", "?")
    # headers.User-Agent — список строк по Caddy JSON schema
    headers = req.get("headers") or {}
    ua_list = headers.get("User-Agent", [])
    ua = ua_list[0] if ua_list else "?"
    ts = entry.get("ts")
    user_id = entry.get("user_id", "")
    attempted_username = _extract_attempted_username(entry)

    with sentry_sdk.new_scope() as scope:
        scope.set_tag("event_type", "basic_auth_failed")
        scope.set_tag("http_method", method)
        scope.set_tag("remote_ip", remote_ip)
        if attempted_username:
            scope.set_tag("attempted_username", attempted_username)
        scope.set_context(
            "caddy_log",
            {
                "uri": path,
                "method": method,
                "host": host,
                "remote_ip": remote_ip,
                "user_agent": ua,
                "caddy_ts": ts,
                "user_id": user_id or "(none)",
                "attempted_username": attempted_username or "(none)",
            },
        )
        message = f"basic_auth 401 — {method} {path} from {remote_ip}"
        if attempted_username:
            message += f" (tried: {attempted_username})"
        sentry_sdk.capture_message(message, level="warning")


def emit_digest(count: int, window_s: int, monitor_count: int) -> None:
    """Отправляет digest event при storm'е."""
    with sentry_sdk.new_scope() as scope:
        scope.set_tag("event_type", "basic_auth_storm")
        sentry_sdk.capture_message(
            f"basic_auth storm — {count} failed attempts in {window_s}s"
            + (f" (including {monitor_count} monitor probes)" if monitor_count else ""),
            level="error",
        )


def is_401_event(entry: dict) -> bool:  # type: ignore[type-arg]
    """Проверяет что entry — failed auth (401)."""
    return entry.get("status") == 401


def main() -> None:
    signal.signal(signal.SIGTERM, _signal_handler)
    signal.signal(signal.SIGINT, _signal_handler)

    sentry_sdk.init(
        dsn=DSN,
        environment=ENV,
        release=os.getenv("APP_RELEASE", "auth-forwarder-1"),
        traces_sample_rate=0.0,
        attach_stacktrace=False,
        send_default_pii=False,
        before_send=_drop_basic_auth_noise,
        # Отключаем интеграции которые не нужны тонкому sidecar
        default_integrations=False,
    )

    print(
        f"[forwarder] starting; log={LOG_FILE}; state={STATE_FILE}; env={ENV}",
        flush=True,
    )

    # Ожидаем появления лог-файла (Caddy пишет его при первом запросе)
    if not LOG_FILE.exists():
        print(
            f"[forwarder] log file {LOG_FILE} does not exist yet, waiting...",
            flush=True,
        )
        while not LOG_FILE.exists() and not _shutdown:
            time.sleep(2)
        if _shutdown:
            return

    offset = load_offset()
    file_size = LOG_FILE.stat().st_size
    # Если лог был ротирован или обрезан — сбрасываем offset
    if offset > file_size:
        print(
            f"[forwarder] offset={offset} > file_size={file_size}, log rotated, reset to 0",
            flush=True,
        )
        offset = 0
        save_offset(offset)

    # Sliding window для throttle
    recent_timestamps: deque[float] = deque()
    monitor_timestamps: deque[float] = deque()
    digest_sent = False

    print(f"[forwarder] starting tail from offset={offset}", flush=True)

    while not _shutdown:
        try:
            current_size = LOG_FILE.stat().st_size
            # Log rotation detection: файл стал меньше чем наш offset
            if current_size < offset:
                print("[forwarder] log rotation detected, resetting offset to 0", flush=True)
                offset = 0
                save_offset(offset)
                recent_timestamps.clear()
                monitor_timestamps.clear()
                digest_sent = False

            if current_size > offset:
                with LOG_FILE.open("rb") as f:
                    f.seek(offset)
                    for raw_line in f:
                        if _shutdown:
                            break
                        line = raw_line.decode("utf-8", errors="replace").strip()
                        if not line:
                            continue
                        try:
                            entry = json.loads(line)
                        except json.JSONDecodeError:
                            continue

                        if not is_401_event(entry):
                            continue

                        now = time.monotonic()

                        # Определяем User-Agent для классификации
                        req = entry.get("request", {})
                        headers = req.get("headers") or {}
                        ua_list = headers.get("User-Agent", [])
                        ua = ua_list[0] if ua_list else ""
                        is_monitor = is_monitor_ua(ua)

                        # Обновляем скользящее окно
                        recent_timestamps.append(now)
                        if is_monitor:
                            monitor_timestamps.append(now)

                        cutoff = now - THROTTLE_WINDOW_S
                        while recent_timestamps and recent_timestamps[0] < cutoff:
                            recent_timestamps.popleft()
                        while monitor_timestamps and monitor_timestamps[0] < cutoff:
                            monitor_timestamps.popleft()

                        window_count = len(recent_timestamps)

                        if window_count > THROTTLE_THRESHOLD:
                            # Storm mode: один digest, индивидуальные events пропускаем
                            if not digest_sent:
                                emit_digest(
                                    window_count,
                                    THROTTLE_WINDOW_S,
                                    len(monitor_timestamps),
                                )
                                digest_sent = True
                                print(
                                    f"[forwarder] storm digest sent ({window_count} events)",
                                    flush=True,
                                )
                        else:
                            digest_sent = False
                            # Мониторы (GlitchTip uptime, SentryUptimeBot) получают 401 постоянно —
                            # не шлём их как individual events, только учитываем в throttle
                            if not is_monitor:
                                emit_event(entry)
                                remote_ip = req.get("client_ip") or req.get("remote_ip", "?")
                                path = req.get("uri", "?")
                                print(
                                    f"[forwarder] 401 event sent: {remote_ip} -> {path}",
                                    flush=True,
                                )

                    offset = f.tell()

                save_offset(offset)

            time.sleep(POLL_INTERVAL_S)

        except FileNotFoundError:
            print("[forwarder] log file disappeared, waiting...", flush=True)
            time.sleep(5)
        except Exception as exc:
            print(f"[forwarder] unhandled error: {exc}", file=sys.stderr, flush=True)
            global _last_exc_sent
            now_wall = time.time()
            if now_wall - _last_exc_sent > _EXC_THROTTLE_S:
                try:
                    sentry_sdk.capture_exception(exc)
                except Exception:
                    pass
                _last_exc_sent = now_wall
            time.sleep(5)

    print("[forwarder] shutting down gracefully", flush=True)
    sentry_sdk.flush(timeout=5)


if __name__ == "__main__":
    main()