audit v0.1.0 [base]

"""audit.Module — журнал действий (audit-trail).

Append-only JSONL (data/audit/audit.jsonl), ротация до AUDIT_KEEP записей.
Пишут: хук RightsService (rights.*) + явные audit.record из choke points
(службы, модули, входы, секреты, привязки). Никогда не роняет вызывающего.
"""
import json
import logging
import os
import threading
import time

from core.base_module import BaseModule

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("audit", mod_dir)
        self.ctx = ctx
        try:
            self._keep = max(100, min(int(
                (self.config.get("AUDIT_KEEP", "") or "5000") or 5000),
                50000))
        except (TypeError, ValueError):
            self._keep = 5000
        self._dir = os.path.join(ctx.data_dir, "audit")
        self._file = os.path.join(self._dir, "audit.jsonl")
        self._lock = threading.Lock()
        api = ctx.api.register
        api("audit", "record", self.api_record)
        api("audit", "list", self.api_list)
        api("audit", "stats", self.api_stats)
        try:
            ctx.rights.audit_fn = self._rights_hook
        except Exception:  # noqa: BLE001
            pass

    def _rights_hook(self, action, key, detail=""):
        import re as _re
        m = _re.search(r"actor=(\S+)", detail or "")
        actor = m.group(1) if m else "?"
        self._write(actor, action, key, detail, True)

    # --- хранилище ---

    def _write(self, actor, action, target, detail, ok):
        entry = {"at": int(time.time()),
                 "actor": (actor or "?")[:120],
                 "action": (action or "?")[:80],
                 "target": (target or "")[:300],
                 "detail": (detail or "")[:500],
                 "ok": bool(ok)}
        try:
            with self._lock:
                os.makedirs(self._dir, mode=0o700, exist_ok=True)
                with open(self._file, "a", encoding="utf-8") as f:
                    f.write(json.dumps(entry, ensure_ascii=False) + "\n")
                self._rotate()
        except Exception as e:  # noqa: BLE001
            logger.warning("audit: запись не удалась: %s", e)
        return entry

    def _rotate(self):
        try:
            with open(self._file, "r", encoding="utf-8",
                      errors="replace") as f:
                lines = f.read().splitlines()
        except OSError:
            return
        if len(lines) <= self._keep:
            return
        tmp = self._file + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            f.write("\n".join(lines[-self._keep:]) + "\n")
        os.replace(tmp, self._file)

    def _read(self, limit=200):
        try:
            with open(self._file, "r", encoding="utf-8",
                      errors="replace") as f:
                lines = f.read().splitlines()
        except OSError:
            return []
        out = []
        for line in lines[-max(1, limit * 3):]:
            try:
                out.append(json.loads(line))
            except ValueError:
                continue
        return out

    # --- API ---

    def api_record(self, actor="", action="", target="", detail="",
                   ok=True):
        """Записать событие (право не нужно — gate на вызывающем)."""
        action = (action or "").strip() or "?"
        if len(action) > 80 or len(target or "") > 300:
            raise ValueError("слишком длинное событие")
        self._write(actor, action, target, detail, ok)
        return {"ok": True}

    def api_list(self, limit=200, action="", actor="", since=0):
        """Последние события с фильтрами (право audit_view — в ui_web)."""
        try:
            limit = max(1, min(int(limit or 200), 1000))
        except (TypeError, ValueError):
            limit = 200
        try:
            since = int(since or 0)
        except (TypeError, ValueError):
            since = 0
        out = []
        for e in reversed(self._read(limit)):
            if action and e.get("action") != action:
                continue
            if actor and e.get("actor") != actor:
                continue
            if since and (e.get("at") or 0) < since:
                continue
            out.append(e)
            if len(out) >= limit:
                break
        return out

    def api_stats(self):
        """Сводка (право audit_view — в ui_web)."""
        from collections import Counter as _Counter
        c = _Counter()
        total = 0
        for e in self._read(5000):
            total += 1
            c[e.get("action") or "?"] += 1
        return {"total": total, "by_action": dict(c.most_common(50))}

    def health(self) -> dict:
        try:
            st = self.api_stats()
            return {"ok": True, "module": self.name,
                    "events": st.get("total", 0)}
        except Exception as e:  # noqa: BLE001
            return {"ok": False, "module": self.name, "error": str(e)}

"""audit.ui_web — кабинет «Журнал действий»."""

ROUTES = [
    ("GET", "/api/audit/log", "log", {"right": "audit_view",
     "desc": "События (?limit, ?action, ?actor, ?since)"}),
]

SERVICE = [
    {"key": "audit_view", "icon": "📝", "title": "Журнал действий", "right": "audit_view",
     "phase": 1, "desc": "Кто/когда/что: права, службы, модули, входы"},
]

PANELS = []


def handle_api(ctx, config, method, req):
    from core.errors import UserError

    if method == "log":
        query = req.get("query") or {}
        try:
            limit = int(query.get("limit") or 200)
        except (TypeError, ValueError):
            limit = 200
        try:
            since = int(query.get("since") or 0)
        except (TypeError, ValueError):
            since = 0
        return {"ok": True, "events": ctx.api.call(
            "audit.list", limit, query.get("action") or "",
            query.get("actor") or "", since)}

    raise UserError("неизвестный метод: %s" % method)