Adolfo Reyna
452 lines
18 KiB
Python
452 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""Create Pi-hole daily reports and retain a short Deco identity history.
|
|
|
|
Pi-hole is queried remotely over the user's existing SSH access. Deco snapshots
|
|
and completed reports are stored locally in a separate SQLite database. No
|
|
Pi-hole credentials or router session material is read or written here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shlex
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import time as time_module
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
from zoneinfo import ZoneInfo
|
|
|
|
TZ = ZoneInfo("America/New_York")
|
|
DEFAULT_SERVER = "adolforeyna@192.168.68.110"
|
|
DEFAULT_PIHOLE_DB = "/home/adolforeyna/pihole/etc-pihole/pihole-FTL.db"
|
|
DEFAULT_STORAGE_DB = str(Path.home() / "brain/projects/network_usage/network_usage.sqlite3")
|
|
BLOCKED_STATUSES = (1, 4, 5, 6, 7, 8, 9, 10, 11, 18)
|
|
STATUS_NAMES = {
|
|
1: "gravity",
|
|
4: "regex",
|
|
5: "denylist",
|
|
6: "external-blocked-ip",
|
|
7: "external-blocked-null",
|
|
8: "external-blocked-nxra",
|
|
9: "gravity-cname",
|
|
10: "regex-cname",
|
|
11: "denylist-cname",
|
|
18: "external-blocked-ede15",
|
|
}
|
|
|
|
|
|
def parse_day(value: str, today: date) -> date:
|
|
if value == "today":
|
|
return today
|
|
if value == "yesterday":
|
|
return today - timedelta(days=1)
|
|
return date.fromisoformat(value)
|
|
|
|
|
|
def run_json(command: List[str], timeout: int) -> dict:
|
|
proc = subprocess.run(command, capture_output=True, text=True, timeout=timeout)
|
|
if proc.returncode:
|
|
detail = proc.stderr.strip() or proc.stdout.strip() or "exit %s" % proc.returncode
|
|
raise RuntimeError("command failed: %s" % detail[:500])
|
|
try:
|
|
return json.loads(proc.stdout)
|
|
except json.JSONDecodeError as exc:
|
|
raise RuntimeError("command returned invalid JSON: %s" % exc) from exc
|
|
|
|
|
|
def get_deco_clients(cli: str) -> List[dict]:
|
|
payload = run_json([cli, "deco", "clients", "--json"], timeout=120)
|
|
if not payload.get("ok"):
|
|
raise RuntimeError("Deco client lookup failed: %s" % payload)
|
|
return payload.get("result", {}).get("clients", [])
|
|
|
|
|
|
def connect_storage(path: str) -> sqlite3.Connection:
|
|
storage = Path(path).expanduser()
|
|
storage.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(storage), timeout=10)
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
conn.execute("PRAGMA busy_timeout=10000")
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS deco_snapshots (
|
|
snapshot_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
captured_at TEXT NOT NULL,
|
|
captured_epoch INTEGER NOT NULL,
|
|
client_count INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_deco_snapshots_epoch
|
|
ON deco_snapshots(captured_epoch);
|
|
CREATE TABLE IF NOT EXISTS deco_clients (
|
|
snapshot_id INTEGER NOT NULL REFERENCES deco_snapshots(snapshot_id) ON DELETE CASCADE,
|
|
ip TEXT NOT NULL,
|
|
mac TEXT,
|
|
hostname TEXT,
|
|
connection TEXT,
|
|
interface TEXT,
|
|
active INTEGER,
|
|
linked_deco_mac TEXT,
|
|
linked_deco_name TEXT,
|
|
linked_deco_role TEXT,
|
|
raw_json TEXT NOT NULL,
|
|
PRIMARY KEY(snapshot_id, ip, mac)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_deco_clients_ip
|
|
ON deco_clients(ip);
|
|
CREATE TABLE IF NOT EXISTS daily_reports (
|
|
report_day TEXT PRIMARY KEY,
|
|
generated_at TEXT NOT NULL,
|
|
report_json TEXT NOT NULL,
|
|
report_markdown TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
conn.commit()
|
|
return conn
|
|
|
|
|
|
def save_deco_snapshot(storage_path: str, clients: List[dict], captured: Optional[datetime] = None) -> dict:
|
|
captured = captured or datetime.now(timezone.utc)
|
|
captured_epoch = int(captured.timestamp())
|
|
captured_text = captured.isoformat()
|
|
conn = connect_storage(storage_path)
|
|
try:
|
|
with conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO deco_snapshots(captured_at,captured_epoch,client_count) VALUES(?,?,?)",
|
|
(captured_text, captured_epoch, len(clients)),
|
|
)
|
|
snapshot_id = cur.lastrowid
|
|
rows = []
|
|
for client in clients:
|
|
ip = str(client.get("ip") or "").strip()
|
|
if not ip:
|
|
continue
|
|
rows.append(
|
|
(
|
|
snapshot_id,
|
|
ip,
|
|
client.get("mac"),
|
|
client.get("hostname"),
|
|
client.get("connection"),
|
|
client.get("interface"),
|
|
1 if client.get("active") else 0,
|
|
client.get("linkedDecoMac"),
|
|
client.get("linkedDecoName"),
|
|
client.get("linkedDecoRole"),
|
|
json.dumps(client, separators=(",", ":"), sort_keys=True),
|
|
)
|
|
)
|
|
conn.executemany(
|
|
"""INSERT OR REPLACE INTO deco_clients
|
|
(snapshot_id,ip,mac,hostname,connection,interface,active,
|
|
linked_deco_mac,linked_deco_name,linked_deco_role,raw_json)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?)""",
|
|
rows,
|
|
)
|
|
return {"snapshot_id": snapshot_id, "captured_at": captured_text, "clients": len(rows)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def load_snapshot_map(storage_path: str, start_epoch: int, end_epoch: int) -> dict:
|
|
"""Return latest identity per IP observed during the report day.
|
|
|
|
If an IP was associated with multiple MACs during the day, the latest
|
|
record is selected but the mapping is explicitly marked ambiguous.
|
|
"""
|
|
conn = connect_storage(storage_path)
|
|
try:
|
|
rows = conn.execute(
|
|
"""SELECT s.captured_at,c.ip,c.mac,c.hostname,c.connection,c.interface,
|
|
c.active,c.linked_deco_mac,c.linked_deco_name,c.linked_deco_role
|
|
FROM deco_clients c JOIN deco_snapshots s ON s.snapshot_id=c.snapshot_id
|
|
WHERE s.captured_epoch>=? AND s.captured_epoch<?
|
|
ORDER BY s.captured_epoch DESC""",
|
|
(start_epoch, end_epoch),
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
by_ip = {}
|
|
identities = defaultdict(set)
|
|
for row in rows:
|
|
captured_at, ip, mac, hostname, connection, interface, active, linked_mac, linked_name, linked_role = row
|
|
identities[ip].add(mac or "(unknown-mac)")
|
|
if ip in by_ip:
|
|
continue
|
|
by_ip[ip] = {
|
|
"ip": ip,
|
|
"mac": mac,
|
|
"hostname": hostname,
|
|
"connection": connection,
|
|
"interface": interface,
|
|
"active": bool(active),
|
|
"linkedDecoMac": linked_mac,
|
|
"linkedDecoName": linked_name,
|
|
"linkedDecoRole": linked_role,
|
|
"mapping_status": "historical_snapshot",
|
|
"snapshot_captured_at": captured_at,
|
|
}
|
|
for ip, deco in by_ip.items():
|
|
if len(identities[ip]) > 1:
|
|
deco["mapping_status"] = "multiple_identities_seen"
|
|
deco["mapping_identity_count"] = len(identities[ip])
|
|
return by_ip
|
|
|
|
|
|
def prune_old_snapshots(storage_path: str, before_epoch: int) -> int:
|
|
conn = connect_storage(storage_path)
|
|
try:
|
|
with conn:
|
|
cur = conn.execute("DELETE FROM deco_snapshots WHERE captured_epoch<?", (before_epoch,))
|
|
return cur.rowcount
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def save_daily_report(storage_path: str, report: dict, markdown_text: str) -> None:
|
|
conn = connect_storage(storage_path)
|
|
try:
|
|
with conn:
|
|
conn.execute(
|
|
"""INSERT INTO daily_reports(report_day,generated_at,report_json,report_markdown)
|
|
VALUES(?,?,?,?)
|
|
ON CONFLICT(report_day) DO UPDATE SET
|
|
generated_at=excluded.generated_at,
|
|
report_json=excluded.report_json,
|
|
report_markdown=excluded.report_markdown""",
|
|
(
|
|
report["day"],
|
|
report["generated_at"],
|
|
json.dumps(report, indent=2),
|
|
markdown_text,
|
|
),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_pihole_data(server: str, db_path: str, start_epoch: int, end_epoch: int) -> dict:
|
|
blocked = ",".join(str(x) for x in BLOCKED_STATUSES)
|
|
remote = f'''import sqlite3,json
|
|
p={db_path!r}; start={start_epoch}; end={end_epoch}; blocked=({blocked},)
|
|
c=sqlite3.connect("file:"+p+"?mode=ro",uri=True)
|
|
c.row_factory=sqlite3.Row
|
|
base="""from query_storage q
|
|
join domain_by_id d on d.id=q.domain
|
|
join client_by_id cb on cb.id=q.client
|
|
where q.timestamp>=? and q.timestamp<?"""
|
|
|
|
def rows(sql, params=(start,end)):
|
|
return [dict(r) for r in c.execute(sql, params)]
|
|
|
|
client_sql="""select cb.ip as ip, coalesce(nullif(cb.name,''),'') as pihole_name,
|
|
count(*) as queries, sum(case when q.status in ({blocked}) then 1 else 0 end) as blocked,
|
|
count(distinct q.domain) as unique_domains, max(q.timestamp) as last_query
|
|
{{base}} group by cb.id order by queries desc""".format(blocked=','.join(str(x) for x in blocked),base=base)
|
|
domain_sql="""select cb.ip as ip, d.domain as domain, count(*) as queries,
|
|
sum(case when q.status in ({blocked}) then 1 else 0 end) as blocked,
|
|
sum(case when q.status not in ({blocked}) then 1 else 0 end) as allowed
|
|
{{base}} group by cb.ip,d.domain order by cb.ip,queries desc""".format(blocked=','.join(str(x) for x in blocked),base=base)
|
|
status_sql="""select q.status as status, count(*) as queries
|
|
from query_storage q where q.timestamp>=? and q.timestamp<? group by q.status order by q.status"""
|
|
print(json.dumps({{"clients":rows(client_sql),"domains":rows(domain_sql),"statuses":rows(status_sql),"window":{{"start":start,"end":end}}}},separators=(",",":")))'''
|
|
remote_cmd = "python3 -c " + shlex.quote(remote)
|
|
proc = subprocess.run(
|
|
["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", server, remote_cmd],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=180,
|
|
)
|
|
if proc.returncode:
|
|
detail = proc.stderr.strip() or proc.stdout.strip() or "exit %s" % proc.returncode
|
|
raise RuntimeError("Pi-hole query failed: %s" % detail[:500])
|
|
try:
|
|
return json.loads(proc.stdout)
|
|
except json.JSONDecodeError as exc:
|
|
raise RuntimeError("Pi-hole query returned invalid JSON: %s" % exc) from exc
|
|
|
|
|
|
def enrich(pihole: dict, deco_clients: List[dict], historical_map: dict) -> List[dict]:
|
|
by_ip = dict(historical_map)
|
|
for client in deco_clients:
|
|
ip = str(client.get("ip") or "")
|
|
if ip and ip not in by_ip:
|
|
copy = dict(client)
|
|
copy["mapping_status"] = "live_deco_lookup"
|
|
by_ip[ip] = copy
|
|
domains_by_ip = defaultdict(list)
|
|
for row in pihole["domains"]:
|
|
domains_by_ip[row["ip"]].append(row)
|
|
result = []
|
|
for row in pihole["clients"]:
|
|
ip = row["ip"]
|
|
deco = by_ip.get(ip)
|
|
visited = sorted(
|
|
(x for x in domains_by_ip[ip] if not x["blocked"]),
|
|
key=lambda x: (x["allowed"], x["queries"], x["domain"]),
|
|
reverse=True,
|
|
)
|
|
blocked = sorted(
|
|
(x for x in domains_by_ip[ip] if x["blocked"]),
|
|
key=lambda x: (x["blocked"], x["queries"], x["domain"]),
|
|
reverse=True,
|
|
)
|
|
result.append({
|
|
"ip": ip,
|
|
"pihole_name": row["pihole_name"],
|
|
"queries": row["queries"],
|
|
"blocked_queries": row["blocked"],
|
|
"block_rate_percent": round(100 * row["blocked"] / row["queries"], 2) if row["queries"] else 0,
|
|
"unique_domains": row["unique_domains"],
|
|
"last_query_epoch": row["last_query"],
|
|
"deco": deco or {
|
|
"hostname": None,
|
|
"mac": None,
|
|
"linkedDecoName": None,
|
|
"active": False,
|
|
"mapping_status": "not_in_snapshot_or_live_deco_list",
|
|
},
|
|
"top_visited": visited[:10],
|
|
"top_blocked": blocked[:10],
|
|
})
|
|
return result
|
|
|
|
|
|
def markdown(report: dict) -> str:
|
|
lines = [
|
|
f"# Pi-hole DNS usage — {report['day']}",
|
|
"",
|
|
f"Window: {report['window']['start_local']} to {report['window']['end_local']} ({report['timezone']})",
|
|
f"Generated: {report['generated_at']}",
|
|
f"Pi-hole: {report['pihole']['server']} (FTL database, read-only)",
|
|
f"Deco mapping source: {report['deco_mapping_source']}",
|
|
"",
|
|
"Blocked status codes counted: " + ", ".join(f"{k}={v}" for k, v in STATUS_NAMES.items()),
|
|
"",
|
|
"## Device summary",
|
|
"",
|
|
"| Device | IP | Deco/location | Queries | Blocked | Block rate | Top visited | Top blocked |",
|
|
"|---|---|---|---:|---:|---:|---|---|",
|
|
]
|
|
for d in report["devices"]:
|
|
deco = d["deco"]
|
|
label = deco.get("hostname") or d["pihole_name"] or "Unknown device"
|
|
location = deco.get("linkedDecoName") or "not mapped"
|
|
visited = d["top_visited"][0]["domain"] if d["top_visited"] else "—"
|
|
blocked = d["top_blocked"][0]["domain"] if d["top_blocked"] else "—"
|
|
lines.append(f"| {label} | {d['ip']} | {location} | {d['queries']:,} | {d['blocked_queries']:,} | {d['block_rate_percent']:.2f}% | `{visited}` | `{blocked}` |")
|
|
lines += ["", "## Per-device detail", ""]
|
|
for d in report["devices"]:
|
|
deco = d["deco"]
|
|
label = deco.get("hostname") or d["pihole_name"] or "Unknown device"
|
|
location = deco.get("linkedDecoName") or "not mapped"
|
|
lines += [
|
|
f"### {label} ({d['ip']})",
|
|
f"Location: {location}; MAC: {deco.get('mac') or 'unknown'}; mapping: {deco.get('mapping_status', 'unknown')}.",
|
|
f"Queries: {d['queries']:,}; blocked: {d['blocked_queries']:,}; block rate: {d['block_rate_percent']:.2f}%; unique domains: {d['unique_domains']:,}.",
|
|
"",
|
|
"Top visited (not blocked):",
|
|
]
|
|
lines += [f"- {x['domain']} — {x['queries']:,}" for x in d["top_visited"]] or ["- none"]
|
|
lines.append("Top blocked:")
|
|
lines += [f"- {x['domain']} — {x['blocked']:,} blocked of {x['queries']:,}" for x in d["top_blocked"]] or ["- none"]
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--date", default="today", help="today, yesterday, or YYYY-MM-DD")
|
|
parser.add_argument("--output-dir", default=str(Path.home() / "brain/projects/network_usage/reports"))
|
|
parser.add_argument("--storage-db", default=os.environ.get("NETWORK_USAGE_DB", DEFAULT_STORAGE_DB))
|
|
parser.add_argument("--server", default=os.environ.get("PIHOLE_SSH", DEFAULT_SERVER))
|
|
parser.add_argument("--db-path", default=os.environ.get("PIHOLE_DB_PATH", DEFAULT_PIHOLE_DB))
|
|
parser.add_argument("--deco-cli", default=os.environ.get("REYNA_CLI", "reyna-cli"))
|
|
parser.add_argument("--snapshot-only", action="store_true")
|
|
parser.add_argument("--prune", action="store_true", help="after saving the report, remove snapshots before today")
|
|
args = parser.parse_args()
|
|
|
|
if args.snapshot_only:
|
|
clients = get_deco_clients(args.deco_cli)
|
|
result = save_deco_snapshot(args.storage_db, clients)
|
|
print(json.dumps({"ok": True, "job": "deco_snapshot", "storage_db": args.storage_db, **result}, separators=(",", ":")))
|
|
return 0
|
|
|
|
today = datetime.now(TZ).date()
|
|
day = parse_day(args.date, today)
|
|
start_dt = datetime.combine(day, time.min, tzinfo=TZ)
|
|
end_dt = start_dt + timedelta(days=1)
|
|
now = datetime.now(TZ)
|
|
start_epoch = int(start_dt.timestamp())
|
|
end_epoch = int(end_dt.timestamp())
|
|
data = get_pihole_data(args.server, args.db_path, start_epoch, end_epoch)
|
|
|
|
historical_map = load_snapshot_map(args.storage_db, start_epoch, end_epoch)
|
|
live_clients = []
|
|
if day == today or not historical_map:
|
|
live_clients = get_deco_clients(args.deco_cli)
|
|
devices = enrich(data, live_clients, historical_map)
|
|
if historical_map:
|
|
mapping_source = "historical Deco snapshots plus live fallback"
|
|
else:
|
|
mapping_source = "live Deco lookup"
|
|
report = {
|
|
"schema_version": 2,
|
|
"day": day.isoformat(),
|
|
"timezone": str(TZ),
|
|
"generated_at": now.isoformat(),
|
|
"deco_mapping_source": mapping_source,
|
|
"window": {
|
|
"start_local": start_dt.isoformat(),
|
|
"end_local": end_dt.isoformat(),
|
|
"start_epoch": start_epoch,
|
|
"end_epoch": end_epoch,
|
|
},
|
|
"pihole": {
|
|
"server": args.server,
|
|
"database": args.db_path,
|
|
"blocked_statuses": list(BLOCKED_STATUSES),
|
|
"status_names": STATUS_NAMES,
|
|
"status_counts": data["statuses"],
|
|
},
|
|
"storage": {
|
|
"database": args.storage_db,
|
|
"historical_snapshot_client_ips": len(historical_map),
|
|
"live_deco_client_count": len(live_clients),
|
|
},
|
|
"deco_live_client_count": len(live_clients),
|
|
"devices": devices,
|
|
}
|
|
report_text = markdown(report)
|
|
save_daily_report(args.storage_db, report, report_text)
|
|
|
|
output_dir = Path(args.output_dir).expanduser()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
json_path = output_dir / f"{day.isoformat()}.json"
|
|
md_path = output_dir / f"{day.isoformat()}.md"
|
|
json_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
|
|
md_path.write_text(report_text + "\n", encoding="utf-8")
|
|
|
|
pruned = 0
|
|
if args.prune:
|
|
current_start = int(datetime.combine(today, time.min, tzinfo=TZ).timestamp())
|
|
pruned = prune_old_snapshots(args.storage_db, current_start)
|
|
print(json.dumps({"ok": True, "day": report["day"], "devices": len(devices), "historical_snapshot_ips": len(historical_map), "live_deco_clients": len(live_clients), "pruned_snapshots": pruned, "storage_db": args.storage_db, "json": str(json_path), "markdown": str(md_path)}, separators=(",", ":")))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except Exception as exc:
|
|
print(json.dumps({"ok": False, "error": str(exc)}))
|
|
raise SystemExit(1)
|