#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from urllib.parse import quote from urllib.request import Request, urlopen def _env(name: str, default: str) -> str: value = os.environ.get(name) return value if value not in (None, "") else default AW_API_BASE = _env("AW_SERVER_URL", "http://127.0.0.1:5600/api/0").rstrip("/") OUTPUT_DIR = Path(_env("AW_DLP_COMPLIANCE_REPORT_DIR", "/opt/activitywatch/dlp-compliance/reports")) BASE_DIR = Path(__file__).resolve().parent PROFILE_TEMPLATE_MAP = { "152-fz": BASE_DIR / "templates" / "152-fz-report.html", "pci-dss": BASE_DIR / "templates" / "pci-dss-report.html", } @dataclass class ReportStats: total_incidents: int high: int medium: int low: int by_host: dict[str, int] channels: dict[str, int] def _http_json(url: str) -> object: req = Request(url, headers={"Accept": "application/json"}) with urlopen(req, timeout=30) as response: return json.loads(response.read().decode("utf-8")) def _parse_ts(value: str | None) -> datetime | None: if not value: return None text = value.replace("Z", "+00:00") try: return datetime.fromisoformat(text).astimezone(UTC) except ValueError: return None def _load_incidents(start: datetime, end: datetime) -> list[dict]: buckets = _http_json(f"{AW_API_BASE}/buckets") if not isinstance(buckets, dict): return [] bucket_ids = sorted([bid for bid in buckets.keys() if str(bid).startswith("aw-dlp-incidents_")]) incidents: list[dict] = [] for bucket_id in bucket_ids: encoded = quote(str(bucket_id), safe="") events = _http_json(f"{AW_API_BASE}/buckets/{encoded}/events?limit=2000") if not isinstance(events, list): continue for event in events: if not isinstance(event, dict): continue ts = _parse_ts(event.get("timestamp")) if ts is None or ts < start or ts > end: continue incidents.append(event) return incidents def _build_stats(incidents: list[dict]) -> ReportStats: by_host: dict[str, int] = {} channels: dict[str, int] = {} high = medium = low = 0 for event in incidents: data = event.get("data") or {} if not isinstance(data, dict): data = {} host = str(data.get("hostname") or "unknown") by_host[host] = by_host.get(host, 0) + 1 severity = str(data.get("severity") or "low").lower() if severity == "high": high += 1 elif severity == "medium": medium += 1 else: low += 1 channel = str(data.get("signalType") or data.get("source") or "unknown") channels[channel] = channels.get(channel, 0) + 1 return ReportStats( total_incidents=len(incidents), high=high, medium=medium, low=low, by_host=dict(sorted(by_host.items(), key=lambda item: item[1], reverse=True)), channels=dict(sorted(channels.items(), key=lambda item: item[1], reverse=True)), ) def _render_table(title: str, rows: list[tuple[str, int]]) -> str: if not rows: return f"

{title}

Нет данных

" body = "".join([f"{name}{count}" for name, count in rows]) return f"

{title}

{body}
ПараметрЗначение
" def _resolve_template_path(profile: str) -> Path: if profile == "152-fz": explicit = _env("AW_DLP_COMPLIANCE_TEMPLATE", str(PROFILE_TEMPLATE_MAP["152-fz"])) return Path(explicit) return PROFILE_TEMPLATE_MAP.get(profile, PROFILE_TEMPLATE_MAP["152-fz"]) def _render_html(profile: str, period_label: str, stats: ReportStats, generated_at: str) -> str: template = _resolve_template_path(profile).read_text(encoding="utf-8") return ( template.replace("{{PERIOD}}", period_label) .replace("{{PROFILE}}", profile) .replace("{{GENERATED_AT}}", generated_at) .replace("{{TOTAL}}", str(stats.total_incidents)) .replace("{{HIGH}}", str(stats.high)) .replace("{{MEDIUM}}", str(stats.medium)) .replace("{{LOW}}", str(stats.low)) .replace("{{HOST_TABLE}}", _render_table("Инциденты по хостам", list(stats.by_host.items()))) .replace("{{CHANNEL_TABLE}}", _render_table("Инциденты по каналам", list(stats.channels.items()))) ) def _period_bounds(month: str | None) -> tuple[datetime, datetime, str]: if month: start = datetime.fromisoformat(f"{month}-01T00:00:00+00:00").astimezone(UTC) else: now = datetime.now(UTC) start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) if start.month == 12: end = start.replace(year=start.year + 1, month=1) else: end = start.replace(month=start.month + 1) end = end.replace(second=0, microsecond=0) return start, end, start.strftime("%Y-%m") def generate_report(month: str | None = None, profile: str = "152-fz") -> dict[str, object]: start, end, period_label = _period_bounds(month) incidents = _load_incidents(start, end) stats = _build_stats(incidents) OUTPUT_DIR.mkdir(parents=True, exist_ok=True) generated_at = datetime.now(UTC).isoformat().replace("+00:00", "Z") html_out = OUTPUT_DIR / f"{profile}-{period_label}.html" html_out.write_text(_render_html(profile, period_label, stats, generated_at), encoding="utf-8") metadata = { "profile": profile, "period": period_label, "generated_at": generated_at, "aw_api_base": AW_API_BASE, "report_path": str(html_out), "stats": { "total_incidents": stats.total_incidents, "high": stats.high, "medium": stats.medium, "low": stats.low, }, } (OUTPUT_DIR / f"{profile}-{period_label}.json").write_text( json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8", ) return metadata def main() -> None: parser = argparse.ArgumentParser(description="Generate DLP compliance report from AW DLP incidents") parser.add_argument("--month", help="Month in YYYY-MM format (default: current month)") parser.add_argument("--profile", default="152-fz", help="Profile name: 152-fz or pci-dss") parser.add_argument("--stdout-json", action="store_true", help="Print report metadata as JSON") args = parser.parse_args() metadata = generate_report(month=args.month, profile=args.profile) if args.stdout_json: print(json.dumps(metadata, ensure_ascii=False)) if __name__ == "__main__": main()