104 lines
2.6 KiB
Markdown
104 lines
2.6 KiB
Markdown
# 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 '<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:
|
|
|
|
```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;
|
|
```
|