AWatch-rus/docs/dlp-aggregator.md

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:

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:

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:

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:

create database aw_dlp;
create user aw_dlp_ingest with password '<strong generated 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:

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:

* * * * * 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:

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:

select
  date_trunc('hour', event_ts) as time,
  severity,
  count(*) as incidents
from dlp_incidents
group by 1, 2
order by 1 desc;