From 5a4064dc0fe0e25acad447ce9ea4795444fa0924 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 2 May 2026 20:55:44 +0000 Subject: [PATCH] Add DLP incident aggregation prototype --- README.md | 2 + docs/dlp-aggregator.md | 103 +++++++ docs/dlp-gap-analysis.md | 4 +- scripts/aggregate_dlp_events.py | 517 ++++++++++++++++++++++++++++++++ 4 files changed, 624 insertions(+), 2 deletions(-) create mode 100644 docs/dlp-aggregator.md create mode 100755 scripts/aggregate_dlp_events.py diff --git a/README.md b/README.md index f9f17d8..231b8ca 100755 --- a/README.md +++ b/README.md @@ -13,12 +13,14 @@ - `docs/linux-remote-worker.md` — полный Linux remote-worker stack: GUI, SSH/console и browser admin UI вроде Proxmox `:8006`. - `docs/console-ssh-logger.md` — логирование только консольных команд и SSH-сессий в AW. - `docs/dlp-gap-analysis.md` — разрыв до enterprise DLP и roadmap. +- `docs/dlp-aggregator.md` — прототип централизованной агрегации DLP/file-operation событий. - `proxmox/` — шаблонные скрипты подготовки и наполнения CT на стороне Proxmox. - `aw-server/` — установочные скрипты, env-шаблон, systemd unit и RU patch для Web UI. - `ansible/` — Ansible-ensemble для автоматизированного сервера (Debian/CT). - `pfsense/` — внешний poller для pfSense API и systemd unit под Debian/Ubuntu utility VM. - `windows/` — PowerShell toolkit: single-user, domain-users, ensemble orchestration, hardening/recovery, validation, Windows/RDP DLP telemetry (`aw-dlp-incidents_*`, `aw-dlp-endpoint-signals_*`) и session-level presence для удалённых Windows/RDP пользователей (`aw-worktime-sessions_*`). - `scripts/quality-gate.sh` — локальный preflight-пайплайн проверок. +- `scripts/aggregate_dlp_events.py` — сбор `aw-file-operations_*` и `aw-dlp-incidents_*` в SQLite/PostgreSQL. - `scripts/install_aw_linux_client.sh` — установка Linux bundle + autostart для remote AW server. - `scripts/install_aw_console_ssh_logger.sh` — user-space установка console/ssh logger. - `scripts/install_aw_linux_web_category_logger.sh` — user-space классификация browser admin UI по title/class. diff --git a/docs/dlp-aggregator.md b/docs/dlp-aggregator.md new file mode 100644 index 0000000..c54edbb --- /dev/null +++ b/docs/dlp-aggregator.md @@ -0,0 +1,103 @@ +# Central DLP aggregator prototype + +`scripts/aggregate_dlp_events.py` collects Phase 2 DLP telemetry from ActivityWatch buckets and stores normalized rows in one database for Grafana/SIEM-style reporting. + +## Streams + +The prototype reads: + +- `aw-file-operations_*` (`aw.file.operation`) — file create/delete/rename telemetry, including `archiveHint`. +- `aw-dlp-incidents_*` (`aw.dlp.incident`) — browser/endpoint DLP incidents and screenshot metadata when available. + +## SQLite smoke test + +SQLite is the default so the collector can be tested without deploying PostgreSQL: + +```bash +python3 scripts/aggregate_dlp_events.py \ + --aw-url http://10.10.10.13:5600/api/0 \ + --sqlite-path data/dlp-events.sqlite3 \ + --lookback-hours 24 +``` + +Useful checks: + +```bash +sqlite3 data/dlp-events.sqlite3 \ + "select stream_type, hostname, count(*) from dlp_events group by 1,2 order by 3 desc;" + +sqlite3 data/dlp-events.sqlite3 \ + "select event_ts, hostname, username, file_path from dlp_file_operations where archive_hint = 1 order by event_ts desc limit 20;" +``` + +## PostgreSQL mode + +For centralized reporting, pass a DSN through an environment variable instead of committing secrets: + +```bash +export DLP_AGGREGATOR_POSTGRES_DSN='postgresql://aw_dlp:${PASSWORD}@postgres.internal:5432/aw_dlp' +python3 -m pip install 'psycopg[binary]' +python3 scripts/aggregate_dlp_events.py \ + --aw-url http://10.10.10.13:5600/api/0 +``` + +Minimum database bootstrap: + +```sql +create database aw_dlp; +create user aw_dlp_ingest with password ''; +grant connect on database aw_dlp to aw_dlp_ingest; +grant usage, create on schema public to aw_dlp_ingest; +``` + +The script creates: + +- table `dlp_events` +- view `dlp_file_operations` +- view `dlp_incidents` + +## Incremental state + +By default, the aggregator stores the last successful end timestamp in: + +```text +data/dlp-aggregator-state.json +``` + +Future runs resume from that timestamp with a small overlap window to avoid missing late events. Duplicate inserts are ignored by `(bucket_id, event_id)`. + +## Scheduling example + +Cron every minute: + +```cron +* * * * * cd /opt/AWatch-rus && /usr/bin/python3 scripts/aggregate_dlp_events.py --aw-url http://10.10.10.13:5600/api/0 >> /var/log/aw-dlp-aggregator.log 2>&1 +``` + +## Example Grafana queries + +Archive creation by user: + +```sql +select + date_trunc('minute', event_ts) as time, + hostname, + username, + count(*) as archives +from dlp_file_operations +where archive_hint = true +group by 1, 2, 3 +order by 1 desc; +``` + +DLP incidents by severity: + +```sql +select + date_trunc('hour', event_ts) as time, + severity, + count(*) as incidents +from dlp_incidents +group by 1, 2 +order by 1 desc; +``` diff --git a/docs/dlp-gap-analysis.md b/docs/dlp-gap-analysis.md index cd0525e..cec20fb 100644 --- a/docs/dlp-gap-analysis.md +++ b/docs/dlp-gap-analysis.md @@ -29,8 +29,8 @@ - USB/print/clipboard collectors (endpoint signals) — внедрено. - Incident pipeline расширен на endpoint события — внедрено. -- File-operation telemetry (create/copy/archive/upload hints) — в backlog. -- Central incident aggregation/export — в backlog. +- File-operation telemetry (create/delete/rename/archive hints) — прототип внедрён (`windows/file-operations-collector.ps1`). +- Central incident aggregation/export — прототип внедрён (`scripts/aggregate_dlp_events.py`, `docs/dlp-aggregator.md`). ### Phase 3 diff --git a/scripts/aggregate_dlp_events.py b/scripts/aggregate_dlp_events.py new file mode 100755 index 0000000..7be7760 --- /dev/null +++ b/scripts/aggregate_dlp_events.py @@ -0,0 +1,517 @@ +#!/usr/bin/env python3 +import argparse +import json +import os +import sqlite3 +import sys +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Protocol, TypeAlias + + +JsonScalar: TypeAlias = str | int | float | bool | None +JsonValue: TypeAlias = JsonScalar | list["JsonValue"] | dict[str, "JsonValue"] + + +DEFAULT_BUCKET_PREFIXES = ("aw-file-operations_", "aw-dlp-incidents_") +DEFAULT_SQLITE_PATH = "data/dlp-events.sqlite3" +EVENT_COLUMNS = ( + "bucket_id", + "event_id", + "stream_type", + "hostname", + "username", + "event_ts", + "duration", + "operation", + "file_path", + "old_file_path", + "extension", + "archive_hint", + "rule_id", + "action", + "severity", + "signal_type", + "message", + "source", + "screenshot_path", + "raw_json", + "ingested_at", +) + + +@dataclass(frozen=True) +class Bucket: + id: str + type: str + client: str + hostname: str + + +@dataclass(frozen=True) +class AwEvent: + bucket_id: str + hostname: str + stream_type: str + event_id: str + timestamp: str + duration: float + data: dict[str, JsonValue] + + +class PsycopgConnection(Protocol): + def cursor(self): + ... + + def commit(self) -> None: + ... + + +def utc_now() -> datetime: + return datetime.now(tz=UTC) + + +def parse_timestamp(value: str) -> datetime: + normalized = value.replace("Z", "+00:00") + parsed = datetime.fromisoformat(normalized) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + + +def format_aw_timestamp(value: datetime) -> str: + return value.astimezone(UTC).isoformat().replace("+00:00", "Z") + + +def load_state(path: Path) -> dict[str, str]: + if not path.exists(): + return {} + return json.loads(path.read_text(encoding="utf-8")) + + +def save_state(path: Path, state: dict[str, str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(state, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def normalize_base_url(base_url: str) -> str: + return base_url.rstrip("/") + + +def aw_get_json(base_url: str, path: str, timeout: int) -> JsonValue: + url = normalize_base_url(base_url) + path + request = urllib.request.Request(url, headers={"Accept": "application/json"}) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + + +def list_buckets(base_url: str, timeout: int) -> list[Bucket]: + payload = aw_get_json(base_url, "/buckets", timeout) + if not isinstance(payload, dict): + raise ValueError("ActivityWatch /buckets response must be a JSON object") + buckets: list[Bucket] = [] + for bucket_id, bucket_data in payload.items(): + if not isinstance(bucket_data, dict): + continue + buckets.append( + Bucket( + id=str(bucket_id), + type=str(bucket_data.get("type", "")), + client=str(bucket_data.get("client", "")), + hostname=str(bucket_data.get("hostname", "")), + ) + ) + return buckets + + +def bucket_stream_type(bucket: Bucket) -> str | None: + if bucket.id.startswith("aw-file-operations_") or bucket.type == "aw.file.operation": + return "file_operation" + if bucket.id.startswith("aw-dlp-incidents_") or bucket.type == "aw.dlp.incident": + return "dlp_incident" + return None + + +def select_buckets(buckets: Iterable[Bucket], prefixes: tuple[str, ...]) -> list[tuple[Bucket, str]]: + selected: list[tuple[Bucket, str]] = [] + for bucket in buckets: + stream_type = bucket_stream_type(bucket) + if stream_type and any(bucket.id.startswith(prefix) for prefix in prefixes): + selected.append((bucket, stream_type)) + return selected + + +def build_events_path(bucket_id: str, start: datetime, end: datetime, limit: int) -> str: + query = urllib.parse.urlencode( + { + "start": format_aw_timestamp(start), + "end": format_aw_timestamp(end), + "limit": str(limit), + } + ) + return f"/buckets/{urllib.parse.quote(bucket_id, safe='')}/events?{query}" + + +def event_key(bucket_id: str, timestamp: str, duration: float, data: dict[str, JsonValue]) -> str: + payload = json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return f"{bucket_id}|{timestamp}|{duration}|{payload}" + + +def fetch_bucket_events( + base_url: str, + bucket: Bucket, + stream_type: str, + start: datetime, + end: datetime, + limit: int, + timeout: int, +) -> list[AwEvent]: + payload = aw_get_json(base_url, build_events_path(bucket.id, start, end, limit), timeout) + if not isinstance(payload, list): + raise ValueError(f"ActivityWatch events response for {bucket.id} must be a JSON array") + events: list[AwEvent] = [] + for item in payload: + if not isinstance(item, dict): + continue + timestamp = str(item["timestamp"]) + duration = float(item.get("duration", 0) or 0) + data = item.get("data") or {} + if not isinstance(data, dict): + data = {"raw": data} + item_id = str(item.get("id") or event_key(bucket.id, timestamp, duration, data)) + events.append( + AwEvent( + bucket_id=bucket.id, + hostname=bucket.hostname or str(data.get("hostname") or ""), + stream_type=stream_type, + event_id=item_id, + timestamp=timestamp, + duration=duration, + data=data, + ) + ) + return events + + +def connect_sqlite(path: Path) -> sqlite3.Connection: + path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(str(path)) + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA synchronous=NORMAL") + connection.execute("PRAGMA foreign_keys=ON") + return connection + + +def ensure_schema(connection: sqlite3.Connection) -> None: + connection.executescript( + """ + create table if not exists dlp_events ( + id integer primary key autoincrement, + bucket_id text not null, + event_id text not null, + stream_type text not null, + hostname text not null, + username text, + event_ts text not null, + duration real not null default 0, + operation text, + file_path text, + old_file_path text, + extension text, + archive_hint integer not null default 0, + rule_id text, + action text, + severity text, + signal_type text, + message text, + source text, + screenshot_path text, + raw_json text not null, + ingested_at text not null, + unique (bucket_id, event_id) + ); + create index if not exists idx_dlp_events_event_ts on dlp_events(event_ts); + create index if not exists idx_dlp_events_host_ts on dlp_events(hostname, event_ts); + create index if not exists idx_dlp_events_stream_ts on dlp_events(stream_type, event_ts); + create index if not exists idx_dlp_events_archive on dlp_events(archive_hint, event_ts); + create index if not exists idx_dlp_events_rule on dlp_events(rule_id, event_ts); + + create view if not exists dlp_file_operations as + select * + from dlp_events + where stream_type = 'file_operation'; + + create view if not exists dlp_incidents as + select * + from dlp_events + where stream_type = 'dlp_incident'; + """ + ) + connection.commit() + + +def ensure_postgres_schema(connection: PsycopgConnection) -> None: + with connection.cursor() as cursor: + cursor.execute( + """ + create table if not exists dlp_events ( + id bigserial primary key, + bucket_id text not null, + event_id text not null, + stream_type text not null, + hostname text not null, + username text, + event_ts timestamptz not null, + duration double precision not null default 0, + operation text, + file_path text, + old_file_path text, + extension text, + archive_hint boolean not null default false, + rule_id text, + action text, + severity text, + signal_type text, + message text, + source text, + screenshot_path text, + raw_json jsonb not null, + ingested_at timestamptz not null, + unique (bucket_id, event_id) + ); + create index if not exists idx_dlp_events_event_ts on dlp_events(event_ts); + create index if not exists idx_dlp_events_host_ts on dlp_events(hostname, event_ts); + create index if not exists idx_dlp_events_stream_ts on dlp_events(stream_type, event_ts); + create index if not exists idx_dlp_events_archive on dlp_events(archive_hint, event_ts); + create index if not exists idx_dlp_events_rule on dlp_events(rule_id, event_ts); + + create or replace view dlp_file_operations as + select * + from dlp_events + where stream_type = 'file_operation'; + + create or replace view dlp_incidents as + select * + from dlp_events + where stream_type = 'dlp_incident'; + """ + ) + connection.commit() + + +def first_string(data: dict[str, JsonValue], keys: tuple[str, ...]) -> str | None: + for key in keys: + value = data.get(key) + if value is not None and str(value) != "": + return str(value) + return None + + +def bool_as_int(value: JsonValue) -> int: + if isinstance(value, bool): + return int(value) + if isinstance(value, str): + return int(value.lower() in {"1", "true", "yes", "y"}) + return int(bool(value)) + + +def event_row(event: AwEvent, ingested_at: str) -> tuple[JsonValue, ...]: + data = event.data + event_id = event.event_id or event_key(event.bucket_id, event.timestamp, event.duration, data) + return ( + event.bucket_id, + event_id, + event.stream_type, + event.hostname, + first_string(data, ("username", "user")), + event.timestamp, + event.duration, + first_string(data, ("operation",)), + first_string(data, ("path", "filePath")), + first_string(data, ("oldPath", "oldFilePath")), + first_string(data, ("extension",)), + bool_as_int(data.get("archiveHint")), + first_string(data, ("ruleId", "rule")), + first_string(data, ("action",)), + first_string(data, ("severity",)), + first_string(data, ("signalType",)), + first_string(data, ("message",)), + first_string(data, ("source",)), + first_string(data, ("screenshotPath", "capturePath", "artifactPath")), + json.dumps(data, ensure_ascii=False, sort_keys=True), + ingested_at, + ) + + +def insert_events(connection: sqlite3.Connection, events: Iterable[AwEvent]) -> int: + inserted = 0 + now = format_aw_timestamp(utc_now()) + for event in events: + cursor = connection.execute( + """ + insert or ignore into dlp_events ( + bucket_id, + event_id, + stream_type, + hostname, + username, + event_ts, + duration, + operation, + file_path, + old_file_path, + extension, + archive_hint, + rule_id, + action, + severity, + signal_type, + message, + source, + screenshot_path, + raw_json, + ingested_at + ) + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + event_row(event, now), + ) + inserted += int(cursor.rowcount > 0) + connection.commit() + return inserted + + +def insert_postgres_events(dsn: str, events: Iterable[AwEvent]) -> int: + try: + import psycopg + except ImportError as exc: + raise SystemExit("PostgreSQL mode requires psycopg: python3 -m pip install 'psycopg[binary]'") from exc + + inserted = 0 + now = format_aw_timestamp(utc_now()) + columns = ", ".join(EVENT_COLUMNS) + placeholders = ", ".join(["%s"] * len(EVENT_COLUMNS)) + sql = f""" + insert into dlp_events ({columns}) + values ({placeholders}) + on conflict (bucket_id, event_id) do nothing + """ + with psycopg.connect(dsn) as connection: + ensure_postgres_schema(connection) + with connection.cursor() as cursor: + for event in events: + row = list(event_row(event, now)) + row[EVENT_COLUMNS.index("archive_hint")] = bool(row[EVENT_COLUMNS.index("archive_hint")]) + cursor.execute(sql, row) + inserted += int(cursor.rowcount > 0) + connection.commit() + return inserted + + +def get_start_time(args: argparse.Namespace, state: dict[str, str]) -> datetime: + if args.since: + return parse_timestamp(args.since) + if state.get("last_end"): + return parse_timestamp(state["last_end"]) - timedelta(seconds=args.overlap_seconds) + return utc_now() - timedelta(hours=args.lookback_hours) + + +def parse_prefixes(value: str) -> tuple[str, ...]: + prefixes = tuple(item.strip() for item in value.split(",") if item.strip()) + if not prefixes: + raise argparse.ArgumentTypeError("at least one bucket prefix is required") + return prefixes + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Aggregate AWatch-rus DLP buckets into a local warehouse database.") + parser.add_argument("--aw-url", default=os.environ.get("AW_URL", "http://127.0.0.1:5600/api/0")) + parser.add_argument("--postgres-dsn", default=os.environ.get("DLP_AGGREGATOR_POSTGRES_DSN")) + parser.add_argument("--sqlite-path", default=os.environ.get("DLP_AGGREGATOR_SQLITE_PATH", DEFAULT_SQLITE_PATH)) + parser.add_argument("--state-path", default=os.environ.get("DLP_AGGREGATOR_STATE_PATH", "data/dlp-aggregator-state.json")) + parser.add_argument("--bucket-prefixes", type=parse_prefixes, default=DEFAULT_BUCKET_PREFIXES) + parser.add_argument("--since", help="UTC ISO timestamp. Overrides saved state, for example 2026-05-02T00:00:00Z.") + parser.add_argument("--lookback-hours", type=int, default=24) + parser.add_argument("--overlap-seconds", type=int, default=60) + parser.add_argument("--limit", type=int, default=10000) + parser.add_argument("--timeout", type=int, default=15) + parser.add_argument("--dry-run", action="store_true") + return parser + + +def main() -> int: + args = build_parser().parse_args() + state_path = Path(args.state_path) + state = load_state(state_path) + start = get_start_time(args, state) + end = utc_now() + + buckets = select_buckets(list_buckets(args.aw_url, args.timeout), args.bucket_prefixes) + all_events: list[AwEvent] = [] + for bucket, stream_type in buckets: + all_events.extend(fetch_bucket_events(args.aw_url, bucket, stream_type, start, end, args.limit, args.timeout)) + + if args.dry_run: + print( + json.dumps( + { + "aw_url": args.aw_url, + "start": format_aw_timestamp(start), + "end": format_aw_timestamp(end), + "selected_buckets": [bucket.id for bucket, _stream_type in buckets], + "fetched_events": len(all_events), + }, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + if args.postgres_dsn: + target = "postgres" + target_path = args.postgres_dsn.split("@")[-1] + inserted = insert_postgres_events(args.postgres_dsn, all_events) + else: + target = "sqlite" + sqlite_path = Path(args.sqlite_path) + target_path = str(sqlite_path) + connection = connect_sqlite(sqlite_path) + try: + ensure_schema(connection) + inserted = insert_events(connection, all_events) + finally: + connection.close() + + state["last_end"] = format_aw_timestamp(end) + save_state(state_path, state) + print( + json.dumps( + { + "aw_url": args.aw_url, + "target": target, + "target_path": target_path, + "state_path": str(state_path), + "start": format_aw_timestamp(start), + "end": format_aw_timestamp(end), + "selected_buckets": len(buckets), + "fetched_events": len(all_events), + "inserted_events": inserted, + }, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except urllib.error.URLError as exc: + print(f"ActivityWatch API request failed: {exc}", file=sys.stderr) + raise SystemExit(2)