362 lines
14 KiB
Python
362 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""R001 — one-time backfill of legacy runtime data into the SQLite store.
|
|
|
|
Imports existing JSON/CSV/git-history data so the DB becomes the authoritative
|
|
source without losing anything already accumulated on a deployed instance:
|
|
|
|
- cumulative blocklist from the git-history reconstruction (old source of
|
|
truth) -> ``blocklist_entries`` (grow-only ledger)
|
|
- firewall snapshots from ``data/mikrotik/*.json`` -> ``firewall_snapshots``
|
|
- per-source requests from ``reports/<source>/data/requests.csv`` (best effort)
|
|
|
|
Idempotent: skips rows already present. Safe to run repeatedly.
|
|
|
|
Usage:
|
|
python -m app.db.backfill [--dry-run]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
from ..utils.constants import (
|
|
APP_REPORT_DIR,
|
|
APP_REPORT_DATA_DIR,
|
|
DATA_DIR,
|
|
MIKROTIK_DATA_DIR,
|
|
MIKROTIK_REPORT_DIR,
|
|
NPM_REPORT_DIR,
|
|
COPYPARTY_REPORT_DIR,
|
|
)
|
|
|
|
MAX_REPORT_ARTIFACT_BYTES = 10 * 1024 * 1024
|
|
CANONICAL_REPORT_EXPORTS = {
|
|
"requests.csv", "ips.json", "ips_ipv4.csv", "ips_ipv6.csv",
|
|
"blocked_ipv4.txt", "blocked_ipv6.txt", "address_list_ipv4.json",
|
|
"address_list_ipv6.json", "firewall_ipv4.json", "firewall_ipv6.json",
|
|
}
|
|
|
|
|
|
def utcnow() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def _read_cidr_lines(path: str) -> set[str]:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return {line.strip() for line in f if line.strip()}
|
|
except OSError:
|
|
return set()
|
|
|
|
|
|
def backfill_blocklist(conn, dry_run: bool = False) -> dict:
|
|
"""Import the legacy cumulative blocklist into the ledger."""
|
|
from . import repositories as repo
|
|
from ..utils.blocklist_history import (
|
|
committed_lines, current_file_lines,
|
|
)
|
|
|
|
v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")
|
|
v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt")
|
|
# Legacy source of truth: union of all git revisions + working file +
|
|
# (there is no earlier scan target to add here).
|
|
complete_v4 = committed_lines(v4_path) | current_file_lines(v4_path)
|
|
complete_v6 = committed_lines(v6_path) | current_file_lines(v6_path)
|
|
cidrs = sorted(complete_v4 | complete_v6)
|
|
if not cidrs:
|
|
return {"blocklist": {"cidrs": 0, "added": 0}}
|
|
if dry_run:
|
|
return {"blocklist": {"cidrs": len(cidrs), "added": None, "dry_run": True}}
|
|
result = repo.record_blocklist(conn, cidrs, added_by="backfill")
|
|
return {"blocklist": result}
|
|
|
|
|
|
def backfill_firewall(conn, dry_run: bool = False) -> dict:
|
|
"""Import the latest firewall snapshot from ``data/mikrotik/*.json``."""
|
|
from . import repositories as repo
|
|
|
|
def _load(name: str) -> list:
|
|
try:
|
|
with open(os.path.join(MIKROTIK_DATA_DIR, name), "r",
|
|
encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return data if isinstance(data, list) else []
|
|
except (OSError, ValueError):
|
|
return []
|
|
|
|
addr_v4 = _load("address_list_ipv4.json")
|
|
addr_v6 = _load("address_list_ipv6.json")
|
|
fw_v4 = _load("firewall_ipv4.json")
|
|
fw_v6 = _load("firewall_ipv6.json")
|
|
if not (addr_v4 or addr_v6 or fw_v4 or fw_v6):
|
|
return {"firewall": {"dry_run": dry_run, "snapshot": None}}
|
|
# Idempotent: if a firewall snapshot already exists (previous backfill),
|
|
# don't create another copy of the same latest state.
|
|
n = conn.execute(
|
|
"SELECT COUNT(*) FROM firewall_snapshots").fetchone()[0]
|
|
if n:
|
|
return {"firewall": {"dry_run": dry_run, "snapshot": "exists",
|
|
"skipped": True}}
|
|
if dry_run:
|
|
return {"firewall": {"dry_run": True, "snapshot": None}}
|
|
sid = repo.save_firewall_snapshot(
|
|
conn, address_lists_v4=addr_v4, address_lists_v6=addr_v6,
|
|
rules_v4=fw_v4, rules_v6=fw_v6)
|
|
return {"firewall": {"snapshot": sid}}
|
|
|
|
|
|
def backfill_requests(conn, dry_run: bool = False) -> dict:
|
|
"""Best-effort import of per-source ``requests.csv`` files.
|
|
|
|
Only ingests rows that don't already exist (keyed on source+log_file+ts+ip).
|
|
Legacy CSV columns vary by source; we keep a conservative mapping and never
|
|
fail the whole backfill over a single bad file.
|
|
"""
|
|
from . import repositories as repo
|
|
|
|
seen = set()
|
|
for r in conn.execute(
|
|
"SELECT source, log_file, ts, ip FROM requests"):
|
|
seen.add((r["source"], r["log_file"], r["ts"], r["ip"]))
|
|
|
|
candidates = [
|
|
(os.path.join(COPYPARTY_REPORT_DIR, "data", "requests.csv"), "copyparty"),
|
|
(os.path.join(NPM_REPORT_DIR, "data", "requests.csv"), "npm"),
|
|
]
|
|
total = 0
|
|
for path, source in candidates:
|
|
if not os.path.exists(path):
|
|
continue
|
|
try:
|
|
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
|
reader = csv.DictReader(f)
|
|
rows = []
|
|
for row in reader:
|
|
ip = (row.get("ip") or "").strip()
|
|
ts = row.get("ts") or row.get("timestamp") or ""
|
|
log_file = row.get("log_file") or row.get("file") or ""
|
|
if not ip:
|
|
continue
|
|
key = (source, log_file, ts, ip)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
rows.append({
|
|
"source": source, "scan_id": None, "log_file": log_file,
|
|
"ts": ts, "ip": ip,
|
|
"version": 4 if ":" not in ip else 6,
|
|
"method": row.get("method"), "path": row.get("path"),
|
|
"status": _int_or_none(row.get("status")),
|
|
"upstream_status": _int_or_none(
|
|
row.get("upstream_status")),
|
|
"host": row.get("host"), "user": row.get("user"),
|
|
"ua": row.get("ua") or row.get("user_agent"),
|
|
"referer": row.get("referer"),
|
|
"denied": _bool_str(row.get("denied")),
|
|
"banned": _bool_str(row.get("banned")),
|
|
"is_error": _bool_str(row.get("is_error")),
|
|
"extra": None,
|
|
})
|
|
total += 1
|
|
if not dry_run and total:
|
|
repo.insert_requests(conn, rows)
|
|
except Exception as exc:
|
|
print(f" WARN: requests.csv backfill {source}: {exc}")
|
|
return {"requests": {"imported": total, "dry_run": dry_run}}
|
|
|
|
|
|
def backfill_report_artifacts(conn, dry_run: bool = False,
|
|
roots: dict[str, str] | None = None) -> dict:
|
|
"""Import supported report files as versioned, hash-addressed artifacts.
|
|
|
|
Canonical event exports and current firewall/blocklist data are skipped;
|
|
those belong in their normalized tables. Original files are never changed
|
|
or removed. Oversized or unreadable files are reported for manual review.
|
|
"""
|
|
from . import repositories as repo
|
|
|
|
roots = roots or {
|
|
"copyparty": COPYPARTY_REPORT_DIR,
|
|
"npm": NPM_REPORT_DIR,
|
|
"app": APP_REPORT_DIR,
|
|
"mikrotik": MIKROTIK_REPORT_DIR,
|
|
}
|
|
result = {
|
|
"discovered": 0, "imported": 0, "would_import": 0,
|
|
"already_imported": 0, "skipped_canonical": [], "unmapped": [],
|
|
"by_source": {}, "by_format": {}, "dry_run": dry_run,
|
|
}
|
|
|
|
def count(bucket: str, key: str, field: str) -> None:
|
|
stats = result[bucket].setdefault(key, {})
|
|
stats[field] = stats.get(field, 0) + 1
|
|
|
|
for source, root in roots.items():
|
|
if not os.path.isdir(root):
|
|
continue
|
|
for directory, _, filenames in os.walk(root):
|
|
for filename in sorted(filenames):
|
|
path = os.path.join(directory, filename)
|
|
suffix = os.path.splitext(filename)[1].lower()
|
|
if suffix not in (".txt", ".json", ".csv", ".rsc"):
|
|
continue
|
|
relative = os.path.relpath(path, root).replace(os.sep, "/")
|
|
result["discovered"] += 1
|
|
report_format = suffix.lstrip(".")
|
|
count("by_source", source, "discovered")
|
|
count("by_format", report_format, "discovered")
|
|
if filename.lower() in CANONICAL_REPORT_EXPORTS:
|
|
result["skipped_canonical"].append(
|
|
{"source": source, "path": relative})
|
|
count("by_source", source, "skipped_canonical")
|
|
count("by_format", report_format, "skipped_canonical")
|
|
continue
|
|
|
|
try:
|
|
size = os.path.getsize(path)
|
|
if size > MAX_REPORT_ARTIFACT_BYTES:
|
|
raise ValueError(
|
|
f"file exceeds {MAX_REPORT_ARTIFACT_BYTES} bytes")
|
|
with open(path, "rb") as file:
|
|
raw = file.read()
|
|
text = raw.decode("utf-8", errors="replace")
|
|
payload = _report_payload(suffix, text)
|
|
digest = hashlib.sha256(raw).hexdigest()
|
|
modified = datetime.fromtimestamp(
|
|
os.path.getmtime(path), timezone.utc)
|
|
generated_at = _report_generated_at(filename, modified)
|
|
except (OSError, ValueError) as exc:
|
|
result["unmapped"].append(
|
|
{"source": source, "path": relative,
|
|
"reason": str(exc)})
|
|
count("by_source", source, "unmapped")
|
|
count("by_format", report_format, "unmapped")
|
|
continue
|
|
|
|
original_path = f"{source}/{relative}"
|
|
exists = conn.execute(
|
|
"SELECT 1 FROM report_artifacts "
|
|
"WHERE original_path=? AND content_sha256=?",
|
|
(original_path, digest),
|
|
).fetchone()
|
|
if exists:
|
|
result["already_imported"] += 1
|
|
count("by_source", source, "already_imported")
|
|
count("by_format", report_format, "already_imported")
|
|
continue
|
|
if dry_run:
|
|
result["would_import"] += 1
|
|
count("by_source", source, "would_import")
|
|
count("by_format", report_format, "would_import")
|
|
continue
|
|
|
|
try:
|
|
repo.insert_report_artifact(
|
|
conn, source=source,
|
|
report_type=_report_type(suffix),
|
|
generated_at=generated_at, payload=payload,
|
|
original_path=original_path,
|
|
content_sha256=digest,
|
|
)
|
|
result["imported"] += 1
|
|
count("by_source", source, "imported")
|
|
count("by_format", report_format, "imported")
|
|
except (OSError, ValueError, TypeError) as exc:
|
|
result["unmapped"].append(
|
|
{"source": source, "path": relative,
|
|
"reason": str(exc)})
|
|
count("by_source", source, "unmapped")
|
|
count("by_format", report_format, "unmapped")
|
|
return {"reports": result}
|
|
|
|
|
|
def _report_payload(suffix: str, text: str) -> dict:
|
|
if suffix == ".json":
|
|
try:
|
|
return {"format": "json", "data": json.loads(text)}
|
|
except ValueError as exc:
|
|
return {"format": "json", "content": text,
|
|
"parse_error": str(exc)}
|
|
if suffix == ".csv":
|
|
try:
|
|
reader = csv.DictReader(text.splitlines())
|
|
return {"format": "csv", "columns": reader.fieldnames or [],
|
|
"rows": list(reader)}
|
|
except csv.Error as exc:
|
|
return {"format": "csv", "content": text,
|
|
"parse_error": str(exc)}
|
|
return {"format": "text", "content": text}
|
|
|
|
|
|
def _report_type(suffix: str) -> str:
|
|
return {".json": "json_report", ".csv": "csv_report",
|
|
".rsc": "routeros_script"}.get(suffix, "text_report")
|
|
|
|
|
|
def _report_generated_at(filename: str, modified: datetime) -> str:
|
|
match = re.search(
|
|
r"(?P<date>\d{4}-\d{2}-\d{2})(?:[-_T](?P<hour>\d{2})[-_:]"
|
|
r"(?P<minute>\d{2})[-_:](?P<second>\d{2}))?", filename)
|
|
if not match:
|
|
return modified.isoformat(timespec="seconds")
|
|
values = match.groupdict()
|
|
hour = int(values["hour"] or 0)
|
|
minute = int(values["minute"] or 0)
|
|
second = int(values["second"] or 0)
|
|
try:
|
|
parsed = datetime.strptime(values["date"], "%Y-%m-%d").replace(
|
|
hour=hour, minute=minute, second=second, tzinfo=timezone.utc)
|
|
except ValueError:
|
|
return modified.isoformat(timespec="seconds")
|
|
return parsed.isoformat(timespec="seconds")
|
|
|
|
|
|
def _int_or_none(v):
|
|
try:
|
|
if v in (None, "", "-"):
|
|
return None
|
|
return int(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _bool_str(v):
|
|
if isinstance(v, str):
|
|
return int(v.strip().lower() in ("1", "true", "yes", "t"))
|
|
return int(bool(v))
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> None:
|
|
p = argparse.ArgumentParser(prog="python -m app.db.backfill",
|
|
description="Backfill legacy runtime data into SQLite.")
|
|
p.add_argument("--dry-run", action="store_true",
|
|
help="report what would be imported without writing")
|
|
p.add_argument("--reports-only", action="store_true",
|
|
help="import legacy report artifacts only")
|
|
opts = p.parse_args(argv)
|
|
|
|
from . import get_connection
|
|
conn = get_connection()
|
|
print(f"== backfill (dry_run={opts.dry_run}) ==")
|
|
if opts.reports_only:
|
|
print("reports :", backfill_report_artifacts(
|
|
conn, dry_run=opts.dry_run))
|
|
print("== done ==")
|
|
return
|
|
print("blocklist:", backfill_blocklist(conn, dry_run=opts.dry_run))
|
|
print("firewall :", backfill_firewall(conn, dry_run=opts.dry_run))
|
|
print("requests :", backfill_requests(conn, dry_run=opts.dry_run))
|
|
print("reports :", backfill_report_artifacts(conn, dry_run=opts.dry_run))
|
|
print("== done ==")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|