diff --git a/.env.example b/.env.example index c214bc2..d8a3efb 100644 --- a/.env.example +++ b/.env.example @@ -48,6 +48,6 @@ ACME_HTTP_PORT=80 # BACKUP_PROGRESS=auto # BACKUP_PROGRESS_INTERVAL_SECONDS=10 # BACKUP_STATUS_FILE=/state/backup-status.json -# AUDIT_COMPRESS_AFTER_HOURS=24 +# STATE_DB_PATH=/state/shares.db # AUDIT_QUERY_MAX_DAYS=31 # AUDIT_SKIP_USER_SUFFIXES=_svc,_ServiceAcc diff --git a/Dockerfile b/Dockerfile index b6d194f..d98371d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -29,6 +29,8 @@ COPY --from=step-cli /usr/local/bin/step /usr/local/bin/step COPY app/reconcile_shares.py /app/reconcile_shares.py COPY app/backup_to_destination.py /app/backup_to_destination.py COPY app/audit_policy.py /app/audit_policy.py +COPY app/state_db.py /app/state_db.py +COPY app/audit_store.py /app/audit_store.py COPY app/audit_collector.py /app/audit_collector.py COPY app/web_ui.py /app/web_ui.py COPY app/web /app/web diff --git a/README.md b/README.md index 685f3a3..0fb5c88 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ This repository provides a production-oriented Samba file server container that - `\\server\FSLogix` -> `/data/fslogix` - FS_* groups are projected as folders inside the Data share (`/data/groups/data/`). - Data folder ACLs expand nested AD group membership recursively and detect group cycles. -- Group records are persisted in SQLite at `/state/shares.db`. +- Group records, normalized activity events, collector offsets, and web caches share one SQLite database at `/state/shares.db`. - Group folders are name-based while active and moved to archive on deactivation: - active: `/data/groups/data/` - inactive/deleted groups: `/data/groups/archive/` @@ -23,7 +23,7 @@ This repository provides a production-oriented Samba file server container that - `FSLOGIX_GROUP_SID` controls who can access the default FSLogix share (defaults to `DOMAIN_USERS_SID`). - Startup resolves those SIDs to NSS group names via winbind, then uses those resolved groups in Samba `valid users` rules. - Samba `full_audit` is restricted to successful and failed reads, writes, renames/moves, and deletions. -- A collector normalizes those four actions and persists them in daily NDJSON files under `/state/audit`; closed files are gzip-compressed and are never automatically deleted. +- A collector normalizes those four actions and persists them in indexed SQLite tables; activity is never automatically deleted. - A read-only HTTPS web console provides group membership trees, storage usage, searchable activity, live backup progress, and system health. - Web sign-in validates the submitted username/password with Kerberos, permits only users whose winbind group SID set contains `DOMAIN_ADMINS_SID`, and issues an expiring JWT in a Secure, HttpOnly, SameSite=Strict cookie. The browser does not use NTLM/SPNEGO or Kerberos negotiation. - HTTPS certificates are requested from a configured local Smallstep CA and renewed automatically. Pre-issued certificate files are also supported. @@ -44,23 +44,20 @@ The reconciliation script (`/app/reconcile_shares.py`) enforces these rules: 3. Group removed or no longer matching `FS_*` -> set `isActive=0` and move folder to `/data/groups/archive/...`. 4. Previously inactive group returns -> set `isActive=1`, move back into `/data/groups/data/...`. -## SQLite State Database +## Shared SQLite State Database -Database path: `/state/shares.db` +The default database path is `/state/shares.db`; `STATE_DB_PATH` can override it. `SHARE_DB_PATH` remains a backward-compatible fallback. -Table schema: +The database contains: -```sql -CREATE TABLE shares ( - objectGUID TEXT PRIMARY KEY, - samAccountName TEXT NOT NULL, - shareName TEXT NOT NULL, - path TEXT NOT NULL, - createdAt TIMESTAMP NOT NULL, - lastSeenAt TIMESTAMP NOT NULL, - isActive INTEGER NOT NULL -); -``` +- `shares`: AD group-to-folder lifecycle and ACL reconciliation state; +- `audit_events`: normalized read, write, move, and delete events; +- `audit_sources`: Samba log inode/offset checkpoints; +- `web_cache`: cached storage scan results. + +Activity lookup uses UTC epoch seconds, keyset pagination, and multi-column indexes for time, user, share, action, and result. WAL mode allows the collector, reconciler, scanner, and read-only web requests to operate concurrently. + +Standard SQLite does not provide transparent general-purpose compression, so this project deliberately does not depend on a non-core compression VFS. Structured columns avoid repeated JSON field names and make indexed queries much cheaper; the state volume still needs capacity for retained activity. ## AD Requirements @@ -100,6 +97,8 @@ Kerberos requires close time alignment. - `app/reconcile_shares.py` - `app/backup_to_destination.py` - `app/audit_collector.py` +- `app/audit_store.py` +- `app/state_db.py` - `app/web_ui.py` - `app/web/` - `etc/samba/smb.conf` @@ -123,7 +122,7 @@ The launcher builds the current application and starts an isolated, run-scoped n - the actual file-server image, joined to the dummy domain; - a continuous SMB client that exercises reads, writes, renames/moves, and deletions, including activity from an excluded dummy service account. -The preview starts with group, Private, FSLogix, and historical audit data. The client keeps current-day audit activity moving, while a real backup runs immediately and repeats in the background. Open the URL and use the credentials printed by the launcher. Defaults are: +The preview starts with group, Private, and FSLogix data. The client keeps current activity moving, while a real backup runs immediately and repeats in the background. Open the URL and use the credentials printed by the launcher. Defaults are: ```text URL: https://files.localhost:8443 @@ -175,7 +174,7 @@ The E2E suite verifies: - SMB allow/deny behavior and real file operations; - Data, Private, and FSLogix usage aggregation; - high-level `full_audit` ingestion for all four actions, service-account exclusion, filters, facets, and pagination; -- compression and querying of a closed daily audit log; +- shared SQLite schema, integrity, indexes, legacy-log removal, and ordered read deduplication; - real rsync transfer progress, completed backup status, log output, and remote snapshot marker; - overview and system-health aggregation. @@ -284,11 +283,11 @@ The console is intentionally operational and plain: - **Overview**: current capacity totals, active group count, recent activity, and backup state. - **File shares**: one selectable tree per active `FS_*` group/folder with recursively expanded user and nested-group membership, cycle markers, and a live filter. -- **Storage / Data groups**: cached recursive size of every top-level `/Data` group folder. -- **Storage / Users**: per-user `/Private + /FSLogix` totals with component sizes. -- **Activity logs**: dynamic date, user, share, action, operation, result, and path filters with pagination. +- **Data usage**: cached recursive size of every top-level `/Data` group folder. +- **User usage**: per-user `/Private + /FSLogix` totals with component sizes. +- **Activity**: dynamic date, user, share, action, result, and path filters with pagination. - **Backups**: read-only live progress, active transfer rows, snapshot name, and recent backup output. -- **System**: domain trust, Samba configuration, TLS certificate, scanner, and audit archive health. +- **System**: domain trust, Samba configuration, TLS certificate, scanner, and activity database health. The web API has no mutation endpoint other than session login/logout. Files, groups, ACLs, backup schedules, and retention cannot be changed from the console. @@ -372,24 +371,25 @@ Useful optional settings: | `WEB_USAGE_SCAN_INTERVAL_SECONDS` | `900` | Recursive storage scan interval | | `WEB_DIRECTORY_CACHE_SECONDS` | `300` | AD membership tree cache time | | `WEB_MAX_GROUP_NODES` | `10000` | Membership expansion safety limit | -| `AUDIT_COMPRESS_AFTER_HOURS` | `24` | Minimum idle age before compressing a closed day | +| `STATE_DB_PATH` | `/state/shares.db` | Shared SQLite database for shares, activity, collector offsets, and caches | | `AUDIT_QUERY_MAX_DAYS` | `31` | Largest activity query window | | `AUDIT_SKIP_USER_SUFFIXES` | `_svc,_ServiceAcc` | Case-insensitive account suffixes excluded from collection and queries; set empty to disable | If `WEB_ENABLED` is absent on an upgraded installation and no TLS settings/certificate exist, the web service stays disabled while Samba and audit collection continue. Set `WEB_ENABLED=true` after adding TLS configuration. -## Audit Archive +## Activity Database -Samba emits only the selected high-level `full_audit` operations, and the collector makes them durable: +Samba emits only the selected high-level `full_audit` operations, and the collector stores them durably: -- it tails every `/var/log/samba/log.*` source, remembers inode and byte offsets in `/state/audit/collector-state.json`, and follows Samba rotation without duplicating a rotated file; -- each event records timestamp, user, client address/name, share, result, path, and one of the actions `read`, `write`, `move`, or `delete`; +- it tails every `/var/log/samba/log.*` source and stores inode/byte offsets transactionally in `audit_sources`, so source progress and inserted events commit together; +- each event records a UTC timestamp, user, client address/name, share, result, path, and one of `read`, `write`, `move`, or `delete`; - directory listings, sessions, metadata access, file-open/create noise, and all other VFS operations are discarded; users ending in a configured `AUDIT_SKIP_USER_SUFFIXES` value are also discarded; -- records are appended to `/state/audit/YYYY-MM-DD.jsonl`; -- a closed, idle daily file becomes `.jsonl.gz`; -- no audit retention deletion is performed, so capacity planning for the `state_data` volume is the operator's responsibility. +- immediately consecutive identical reads within the same UTC second are collapsed into one event, including across collector polling cycles; a different event interrupts the sequence and preserves later reads; +- activity queries run directly against indexed SQLite columns and use a stable time/id cursor; +- no activity retention deletion is performed. + +Collection starts even when the web UI is disabled. On the first collector start after this migration, recognized legacy daily `.jsonl`/`.jsonl.gz` files and the old collector state file under `/state/audit` are deleted without import. Existing raw Samba log content is then processed using the current action, suffix, and deduplication policy. -Collection starts even when the web UI is disabled. Existing Samba log content is imported when the collector first starts, but the same operation and user-suffix policy is applied during import. Audit data that Samba rotated away before this version was deployed cannot be recovered. ## Backups @@ -400,7 +400,7 @@ Collection starts even when the web UI is disabled. Existing Samba log content i - `/data/private` -> `data/private` - `/data/groups` -> `data/groups` - `/data/fslogix` -> `data/fslogix` - - `/state` -> `state` + - `/state` -> `state` (the live WAL database is replaced by a consistent SQLite online snapshot) - `/var/lib/samba/private` -> `samba/private` - Retention policy env vars (defaults): - `BACKUP_RETENTION_YEARLY=1` @@ -408,7 +408,7 @@ Collection starts even when the web UI is disabled. Existing Samba log content i - `BACKUP_RETENTION_WEEKLY=2` - `BACKUP_RETENTION_DAILY=3` - The backup script writes directly to `BACKUP_LOG_FILE` (default: `/var/log/backup.log`). Cron does not redirect backup output into the logfile. -- Before uploading, the backup script measures all source files so it can report total upload progress. +- Before uploading, the backup script creates a temporary, integrity-checked SQLite online snapshot of the shared state database, then measures all source files so it can report total upload progress. - Upload progress is logged per file every `BACKUP_PROGRESS_INTERVAL_SECONDS` seconds and again when a file reaches 100%, including percentage and transferred/remaining bytes with auto-scaled units. - `BACKUP_PROGRESS=auto` shows an interactive multi-line progress view only for TTY/manual runs. Current file uploads are shown as separate rows, capped at 12 rows, with the total progress row at the bottom. Use `always` to force it or `never` to suppress the bar. File and total progress are still logged. - Every run atomically updates `BACKUP_STATUS_FILE` (default `/state/backup-status.json`) with its state, current source, active files, byte totals, percentage, snapshot, and final result for the live web view. @@ -446,11 +446,11 @@ Collection starts even when the web UI is disabled. Existing Samba log content i docker compose logs -f samba docker compose logs -f samba | grep -E '\\[web\\]|\\[audit\\]' docker compose exec samba python3 /app/reconcile_shares.py -docker compose exec samba sqlite3 /state/shares.db 'SELECT * FROM shares;' +docker compose exec samba sqlite3 /state/shares.db 'PRAGMA quick_check;' +docker compose exec samba sqlite3 /state/shares.db 'SELECT action, count(*) FROM audit_events GROUP BY action;' docker compose exec samba testparm -s docker compose exec samba sh -lc 'tail -n 200 /var/log/samba/log.*' docker compose exec samba sh -lc 'tail -n 200 /var/log/backup.log' -docker compose exec samba sh -lc 'ls -lh /state/audit' docker compose exec samba python3 -m json.tool /state/backup-status.json ``` diff --git a/app/audit_collector.py b/app/audit_collector.py index 1dd0e01..a04a555 100644 --- a/app/audit_collector.py +++ b/app/audit_collector.py @@ -1,30 +1,26 @@ #!/usr/bin/env python3 -"""Persist Samba full_audit records as immutable daily NDJSON archives.""" +"""Persist selected Samba full_audit records in indexed SQLite tables.""" import datetime as dt import glob -import gzip -import json import os import re import signal import sys import time -from typing import Dict, Iterable, Optional +from typing import Dict, Optional try: from .audit_policy import action_for, skip_user + from .audit_store import AuditStore, drop_legacy_audit_files except ImportError: from audit_policy import action_for, skip_user + from audit_store import AuditStore, drop_legacy_audit_files SAMBA_LOG_GLOB = os.getenv("AUDIT_SOURCE_GLOB", "/var/log/samba/log.*") -ARCHIVE_DIR = os.getenv("AUDIT_ARCHIVE_DIR", "/state/audit") -STATE_FILE = os.path.join(ARCHIVE_DIR, "collector-state.json") +LEGACY_AUDIT_DIR = os.getenv("AUDIT_ARCHIVE_DIR", "/state/audit") POLL_SECONDS = max(0.2, float(os.getenv("AUDIT_POLL_SECONDS", "1"))) -COMPRESS_AFTER_HOURS = max( - 1, int(os.getenv("AUDIT_COMPRESS_AFTER_HOURS", "24")) -) AUDIT_MARKER_RE = re.compile(r"smbd_audit:\s*(.*)$") AUDIT_PAYLOAD_RE = re.compile( r"^\s*\d{4}/\d{2}/\d{2}\s+\d{2}:\d{2}:\d{2}(?:\.\d+)?\|" @@ -43,26 +39,6 @@ def utc_now() -> dt.datetime: return dt.datetime.now(dt.timezone.utc) -def atomic_json(path: str, value: object) -> None: - temp_path = f"{path}.tmp" - with open(temp_path, "w", encoding="utf-8") as handle: - json.dump(value, handle, separators=(",", ":"), sort_keys=True) - handle.flush() - os.fsync(handle.fileno()) - os.replace(temp_path, path) - - -def load_state() -> Dict[str, Dict[str, object]]: - try: - with open(STATE_FILE, encoding="utf-8") as handle: - value = json.load(handle) - if isinstance(value, dict): - return value - except (OSError, ValueError): - pass - return {} - - def parse_samba_timestamp(raw_line: str, fallback: dt.datetime) -> str: match = SAMBA_LOG_TIME_RE.match(raw_line) if not match: @@ -108,29 +84,6 @@ def parse_audit_line(raw_line: str, source: str) -> Optional[Dict[str, object]]: } -def archive_events(events: Iterable[Dict[str, object]]) -> int: - handles: Dict[str, object] = {} - count = 0 - try: - for event in events: - day = str(event["ingestedAt"])[:10] - path = os.path.join(ARCHIVE_DIR, f"{day}.jsonl") - handle = handles.get(path) - if handle is None: - handle = open(path, "a", encoding="utf-8") - handles[path] = handle - handle.write(json.dumps(event, separators=(",", ":"), sort_keys=True)) - handle.write("\n") - count += 1 - for handle in handles.values(): - handle.flush() - os.fsync(handle.fileno()) - finally: - for handle in handles.values(): - handle.close() - return count - - def read_new_events(path: str, entry: Dict[str, object]): stat = os.stat(path) inode = int(stat.st_ino) @@ -157,60 +110,40 @@ def read_new_events(path: str, entry: Dict[str, object]): return events, {"inode": inode, "offset": new_offset} -def compress_old_archives() -> None: - cutoff = utc_now() - dt.timedelta(hours=COMPRESS_AFTER_HOURS) - today = utc_now().date().isoformat() - for path in glob.glob(os.path.join(ARCHIVE_DIR, "????-??-??.jsonl")): - day = os.path.basename(path)[:10] - if day == today: - continue - try: - modified = dt.datetime.fromtimestamp(os.path.getmtime(path), dt.timezone.utc) - if modified > cutoff: - continue - target = f"{path}.gz" - temp_target = f"{target}.tmp" - with open(path, "rb") as source, gzip.open(temp_target, "wb", compresslevel=6) as output: - while True: - chunk = source.read(1024 * 1024) - if not chunk: - break - output.write(chunk) - os.replace(temp_target, target) - os.remove(path) - log(f"Compressed {os.path.basename(path)}") - except OSError as exc: - log(f"Unable to compress {path}: {exc}") - - -def collect_once(state: Dict[str, Dict[str, object]]) -> int: - total = 0 +def collect_once(store: AuditStore) -> int: + state = store.source_entries() + all_events = [] + source_updates = {} seen = set() initial_by_inode = { int(entry.get("inode", -1)): entry for entry in state.values() - if isinstance(entry, dict) and int(entry.get("inode", -1)) >= 0 + if int(entry.get("inode", -1)) >= 0 } - for path in sorted(glob.glob(SAMBA_LOG_GLOB)): + candidates = [] + for path in glob.glob(SAMBA_LOG_GLOB): if not os.path.isfile(path): continue + try: + stat = os.stat(path) + except OSError as exc: + log(f"Unable to inspect {path}: {exc}") + continue + known_inode = int(stat.st_ino) in initial_by_inode + candidates.append((not known_inode, stat.st_mtime_ns, path, int(stat.st_ino))) + + for _is_new, _mtime_ns, path, current_inode in sorted(candidates): seen.add(path) try: path_entry = state.get(path, {}) - current_inode = os.stat(path).st_ino if int(path_entry.get("inode", -1)) != current_inode: path_entry = initial_by_inode.get(current_inode, {}) events, new_entry = read_new_events(path, path_entry) - if events: - total += archive_events(events) - state[path] = new_entry + all_events.extend(events) + source_updates[path] = new_entry except OSError as exc: log(f"Unable to read {path}: {exc}") - for stale_path in list(state): - if stale_path not in seen: - state.pop(stale_path, None) - atomic_json(STATE_FILE, state) - return total + return store.append_batch(all_events, source_updates, seen) def stop(_signum, _frame) -> None: @@ -219,23 +152,24 @@ def stop(_signum, _frame) -> None: def main() -> int: - os.makedirs(ARCHIVE_DIR, mode=0o750, exist_ok=True) signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGINT, stop) - state = load_state() - last_compress = 0.0 - log(f"Watching {SAMBA_LOG_GLOB}") - while not STOP: - try: - count = collect_once(state) - if count: - log(f"Archived {count} event(s)") - if time.monotonic() - last_compress >= 3600: - compress_old_archives() - last_compress = time.monotonic() - except Exception as exc: # pylint: disable=broad-except - log(f"Collector cycle failed: {exc}") - time.sleep(POLL_SECONDS) + store = AuditStore() + removed = drop_legacy_audit_files(LEGACY_AUDIT_DIR) + if removed: + log(f"Dropped {removed} legacy audit archive file(s)") + log(f"Watching {SAMBA_LOG_GLOB}; storing events in SQLite") + try: + while not STOP: + try: + count = collect_once(store) + if count: + log(f"Stored {count} event(s)") + except Exception as exc: # pylint: disable=broad-except + log(f"Collector cycle failed: {exc}") + time.sleep(POLL_SECONDS) + finally: + store.close() return 0 diff --git a/app/audit_policy.py b/app/audit_policy.py index 91eacd0..18c1d42 100644 --- a/app/audit_policy.py +++ b/app/audit_policy.py @@ -1,8 +1,9 @@ #!/usr/bin/env python3 """Shared policy for the small set of user-facing audit events.""" +import datetime as dt import os -from typing import Optional, Tuple +from typing import Mapping, Optional, Tuple AUDIT_ACTIONS = frozenset({"read", "write", "move", "delete"}) @@ -51,3 +52,35 @@ def account_name(user: str) -> str: def skip_user(user: str) -> bool: account = account_name(user).casefold() return any(account.endswith(suffix) for suffix in skipped_user_suffixes()) + + +ReadEventKey = Tuple[str, ...] + + +def read_deduplication_key(event: Mapping[str, object]) -> Optional[ReadEventKey]: + if str(event.get("action", "")).casefold() != "read": + return None + try: + timestamp = dt.datetime.fromisoformat( + str(event.get("timestamp", "")).replace("Z", "+00:00") + ) + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=dt.timezone.utc) + second = ( + timestamp.astimezone(dt.timezone.utc) + .replace(microsecond=0) + .isoformat() + ) + except ValueError: + second = str(event.get("timestamp", "")) + + success = bool(event.get("success", False)) + return ( + second, + str(event.get("user", "")), + str(event.get("clientIp", "")), + str(event.get("share", "")), + str(event.get("path", "")), + "success" if success else "failure", + "" if success else str(event.get("result", "")), + ) diff --git a/app/audit_store.py b/app/audit_store.py new file mode 100644 index 0000000..31d0cb7 --- /dev/null +++ b/app/audit_store.py @@ -0,0 +1,369 @@ +#!/usr/bin/env python3 +"""Indexed SQLite storage for normalized Samba audit events.""" + +import datetime as dt +import glob +import os +import sqlite3 +from typing import Dict, Iterable, List, Optional, Set, Tuple + +try: + from .audit_policy import ( + account_name, + read_deduplication_key, + skipped_user_suffixes, + ) + from .state_db import connect_state_db +except ImportError: + from audit_policy import account_name, read_deduplication_key, skipped_user_suffixes + from state_db import connect_state_db + + +AUDIT_SCHEMA = """ +CREATE TABLE IF NOT EXISTS audit_events ( + id INTEGER PRIMARY KEY, + occurred_at TEXT NOT NULL, + occurred_second INTEGER NOT NULL, + ingested_at TEXT NOT NULL, + user TEXT NOT NULL, + account TEXT NOT NULL, + client_ip TEXT NOT NULL, + client TEXT NOT NULL, + share TEXT NOT NULL, + action TEXT NOT NULL CHECK (action IN ('read', 'write', 'move', 'delete')), + result TEXT NOT NULL, + success INTEGER NOT NULL CHECK (success IN (0, 1)), + path TEXT NOT NULL, + source TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS audit_sources ( + path TEXT PRIMARY KEY, + inode INTEGER NOT NULL, + offset INTEGER NOT NULL +) WITHOUT ROWID; +CREATE INDEX IF NOT EXISTS audit_events_time + ON audit_events (occurred_second DESC, id DESC); +CREATE INDEX IF NOT EXISTS audit_events_action_time + ON audit_events (action, occurred_second DESC, id DESC); +CREATE INDEX IF NOT EXISTS audit_events_success_time + ON audit_events (success, occurred_second DESC, id DESC); +CREATE INDEX IF NOT EXISTS audit_events_user_time + ON audit_events (user COLLATE NOCASE, occurred_second DESC, id DESC); +CREATE INDEX IF NOT EXISTS audit_events_account_time + ON audit_events (account, occurred_second DESC, id DESC); +CREATE INDEX IF NOT EXISTS audit_events_share_time + ON audit_events (share COLLATE NOCASE, occurred_second DESC, id DESC); +CREATE INDEX IF NOT EXISTS audit_events_result_time + ON audit_events (result COLLATE NOCASE, occurred_second DESC, id DESC); +""" + + +def ensure_audit_schema(conn: sqlite3.Connection) -> None: + conn.executescript(AUDIT_SCHEMA) + conn.commit() + + +def parse_event_second(timestamp: object) -> int: + parsed = dt.datetime.fromisoformat(str(timestamp).replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=dt.timezone.utc) + return int(parsed.astimezone(dt.timezone.utc).timestamp()) + + +def row_to_event(row: sqlite3.Row) -> Dict[str, object]: + action = str(row["action"]) + return { + "id": int(row["id"]), + "timestamp": str(row["occurred_at"]), + "ingestedAt": str(row["ingested_at"]), + "user": str(row["user"]), + "clientIp": str(row["client_ip"]), + "client": str(row["client"]), + "share": str(row["share"]), + "operation": action, + "action": action, + "result": str(row["result"]), + "success": bool(row["success"]), + "path": str(row["path"]), + "source": str(row["source"]), + } + + +def last_event(conn: sqlite3.Connection) -> Optional[Dict[str, object]]: + row = conn.execute( + """ + SELECT id, occurred_at, ingested_at, user, client_ip, client, share, + action, result, success, path, source + FROM audit_events + ORDER BY id DESC + LIMIT 1 + """ + ).fetchone() + return row_to_event(row) if row is not None else None + + +class AuditStore: + def __init__(self, database_path: Optional[str] = None): + self.conn = connect_state_db(database_path) + ensure_audit_schema(self.conn) + + def close(self) -> None: + self.conn.close() + + def source_entries(self) -> Dict[str, Dict[str, object]]: + return { + str(row["path"]): { + "inode": int(row["inode"]), + "offset": int(row["offset"]), + } + for row in self.conn.execute( + "SELECT path, inode, offset FROM audit_sources" + ) + } + + def append_batch( + self, + events: Iterable[Dict[str, object]], + source_updates: Dict[str, Dict[str, object]], + seen_paths: Set[str], + ) -> int: + inserted = 0 + self.conn.execute("BEGIN IMMEDIATE") + try: + previous_key = read_deduplication_key(last_event(self.conn) or {}) + for event in events: + current_key = read_deduplication_key(event) + if current_key is not None and current_key == previous_key: + continue + self.conn.execute( + """ + INSERT INTO audit_events ( + occurred_at, occurred_second, ingested_at, user, account, + client_ip, client, share, action, result, success, path, + source + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + str(event["timestamp"]), + parse_event_second(event["timestamp"]), + str(event["ingestedAt"]), + str(event["user"]), + account_name(str(event["user"])).casefold(), + str(event["clientIp"]), + str(event["client"]), + str(event["share"]), + str(event["action"]), + str(event["result"]), + int(bool(event["success"])), + str(event["path"]), + str(event["source"]), + ), + ) + inserted += 1 + previous_key = current_key + + for path, entry in source_updates.items(): + self.conn.execute( + """ + INSERT INTO audit_sources (path, inode, offset) + VALUES (?, ?, ?) + ON CONFLICT(path) DO UPDATE SET + inode = excluded.inode, + offset = excluded.offset + """, + (path, int(entry["inode"]), int(entry["offset"])), + ) + for row in self.conn.execute("SELECT path FROM audit_sources").fetchall(): + path = str(row["path"]) + if path not in seen_paths: + self.conn.execute("DELETE FROM audit_sources WHERE path = ?", (path,)) + self.conn.commit() + except Exception: + self.conn.rollback() + raise + return inserted + + +def escape_like(value: str) -> str: + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def date_seconds(day: dt.date) -> int: + return int( + dt.datetime.combine(day, dt.time.min, tzinfo=dt.timezone.utc).timestamp() + ) + + +def activity_conditions( + start: dt.date, + end: dt.date, + params: Dict[str, List[str]], + *, + include_filters: bool, +) -> Tuple[List[str], List[object]]: + conditions = ["occurred_second >= ?", "occurred_second < ?"] + values: List[object] = [ + date_seconds(start), + date_seconds(end + dt.timedelta(days=1)), + ] + for suffix in skipped_user_suffixes(): + conditions.append("lower(account) NOT LIKE ? ESCAPE '\\'") + values.append(f"%{escape_like(suffix)}") + + if not include_filters: + return conditions, values + + filters = { + key: params.get(key, [""])[0].casefold().strip() + for key in ("user", "share", "operation", "action", "path", "result") + } + user = filters["user"] + if user: + if "\\" in user or "@" in user: + conditions.append("user = ? COLLATE NOCASE") + else: + conditions.append("account = ?") + values.append(user) + share = filters["share"] + if share: + conditions.append("share = ? COLLATE NOCASE") + values.append(share) + path = filters["path"] + if path: + conditions.append("lower(path) LIKE ? ESCAPE '\\'") + values.append(f"%{escape_like(path)}%") + action = filters["action"] or filters["operation"] + if action: + conditions.append("action = ?") + values.append(action) + result = filters["result"] + if result == "fail": + conditions.append("success = 0") + elif result: + conditions.append("result = ? COLLATE NOCASE") + values.append(result) + return conditions, values + + +def query_activity( + conn: sqlite3.Connection, + start: dt.date, + end: dt.date, + params: Dict[str, List[str]], +) -> Dict[str, object]: + limit = min(500, max(1, int(params.get("limit", ["100"])[0]))) + conditions, values = activity_conditions(start, end, params, include_filters=True) + where_sql = " AND ".join(conditions) + matched = int( + conn.execute( + f"SELECT count(*) FROM audit_events WHERE {where_sql}", + values, + ).fetchone()[0] + ) + + cursor = params.get("cursor", [""])[0].strip() + page_conditions = list(conditions) + page_values = list(values) + if cursor: + try: + cursor_second, cursor_id = (int(part) for part in cursor.split(":", 1)) + except (TypeError, ValueError) as exc: + raise ValueError("Ungültiger Seitenzeiger") from exc + page_conditions.append( + "(occurred_second < ? OR (occurred_second = ? AND id < ?))" + ) + page_values.extend((cursor_second, cursor_second, cursor_id)) + + rows = conn.execute( + f""" + SELECT id, occurred_at, occurred_second, ingested_at, user, client_ip, + client, share, action, result, success, path, source + FROM audit_events + WHERE {" AND ".join(page_conditions)} + ORDER BY occurred_second DESC, id DESC + LIMIT ? + """, + (*page_values, limit + 1), + ).fetchall() + has_more = len(rows) > limit + rows = rows[:limit] + events = [row_to_event(row) for row in rows] + next_cursor = None + if has_more and rows: + next_cursor = f"{int(rows[-1]['occurred_second'])}:{int(rows[-1]['id'])}" + + facet_conditions, facet_values = activity_conditions( + start, end, params, include_filters=False + ) + facet_where = " AND ".join(facet_conditions) + + def distinct(column: str) -> List[str]: + return [ + str(row[0]) + for row in conn.execute( + f""" + SELECT DISTINCT {column} + FROM audit_events + WHERE {facet_where} AND {column} <> '' + ORDER BY {column} COLLATE NOCASE + """, + facet_values, + ) + ] + + actions = distinct("action") + return { + "events": events, + "nextCursor": next_cursor, + "matched": matched, + "facets": { + "users": distinct("user"), + "shares": distinct("share"), + "operations": actions, + "actions": actions, + }, + } + + +def audit_summary(conn: sqlite3.Connection, database_path: str) -> Dict[str, object]: + row = conn.execute( + """ + SELECT count(DISTINCT substr(occurred_at, 1, 10)), + min(substr(occurred_at, 1, 10)), + max(substr(occurred_at, 1, 10)) + FROM audit_events + """ + ).fetchone() + database_bytes = 0 + for suffix in ("", "-wal"): + try: + database_bytes += os.path.getsize(f"{database_path}{suffix}") + except OSError: + pass + return { + "days": int(row[0] or 0), + "bytes": database_bytes, + "oldest": row[1], + "newest": row[2], + } + + +def drop_legacy_audit_files(directory: str) -> int: + removed = 0 + patterns = ( + "????-??-??.jsonl", + "????-??-??.jsonl.gz", + "collector-state.json", + "collector-state.json.tmp", + ) + for pattern in patterns: + for path in glob.glob(os.path.join(directory, pattern)): + if not os.path.isfile(path): + continue + os.remove(path) + removed += 1 + try: + os.rmdir(directory) + except OSError: + pass + return removed diff --git a/app/backup_to_destination.py b/app/backup_to_destination.py index ffb4331..0975562 100644 --- a/app/backup_to_destination.py +++ b/app/backup_to_destination.py @@ -5,6 +5,8 @@ import fcntl import json import os import re +import shutil +import sqlite3 import subprocess import sys import tempfile @@ -14,7 +16,12 @@ from typing import Callable, Dict, List, Optional, Set, TextIO, Tuple from urllib.parse import SplitResult, unquote, urlsplit -LOCK_PATH = "/state/backup.lock" +STATE_ROOT = os.getenv("STATE_ROOT", "/state") +STATE_DB_PATH = os.getenv( + "STATE_DB_PATH", + os.getenv("SHARE_DB_PATH", os.path.join(STATE_ROOT, "shares.db")), +) +LOCK_PATH = os.path.join(STATE_ROOT, "backup.lock") DEFAULT_BACKUP_LOG_FILE = "/var/log/backup.log" DEFAULT_BACKUP_STATUS_FILE = "/state/backup-status.json" DEFAULT_PROGRESS_MODE = "auto" @@ -43,7 +50,7 @@ BACKUP_SOURCES: List[Tuple[str, str]] = [ ("/data/private", "data/private"), ("/data/groups", "data/groups"), ("/data/fslogix", "data/fslogix"), - ("/state", "state"), + (STATE_ROOT, "state"), ("/var/lib/samba/private", "samba/private"), ] @@ -917,11 +924,75 @@ def redact_destination(raw_url: str) -> str: return f"{parts.scheme}://{netloc}{parts.path}" -def available_sources() -> List[Tuple[str, str]]: +def prepare_state_snapshot( + state_root: str = STATE_ROOT, + database_path: str = STATE_DB_PATH, +) -> Tuple[str, str]: + """Copy state and replace the live WAL database with an online snapshot.""" + state_root = os.path.abspath(state_root) + database_path = os.path.abspath(database_path) + database_relative = os.path.relpath(database_path, state_root) + if database_relative == ".." or database_relative.startswith(f"..{os.sep}"): + raise RuntimeError("STATE_DB_PATH must be located below STATE_ROOT") + + temporary_root = tempfile.mkdtemp(prefix="backup-state-") + staged_state = os.path.join(temporary_root, "state") + excluded = { + database_relative, + f"{database_relative}-wal", + f"{database_relative}-shm", + } + + def ignore_live_database(directory: str, names: List[str]) -> Set[str]: + relative_directory = os.path.relpath(directory, state_root) + return { + name + for name in names + if os.path.normpath(os.path.join(relative_directory, name)) in excluded + } + + try: + shutil.copytree( + state_root, + staged_state, + symlinks=True, + ignore=ignore_live_database, + ) + if os.path.isfile(database_path): + staged_database = os.path.join(staged_state, database_relative) + os.makedirs(os.path.dirname(staged_database), exist_ok=True) + source = sqlite3.connect( + f"file:{database_path}?mode=ro", + uri=True, + timeout=30, + ) + destination = sqlite3.connect(staged_database, timeout=30) + try: + source.execute("PRAGMA busy_timeout = 30000") + source.backup(destination) + if destination.execute("PRAGMA quick_check").fetchone()[0] != "ok": + raise RuntimeError("SQLite state snapshot failed its integrity check") + finally: + destination.close() + source.close() + return temporary_root, staged_state + except Exception: + shutil.rmtree(temporary_root, ignore_errors=True) + raise + + +def available_sources( + state_snapshot: Optional[str] = None, +) -> List[Tuple[str, str]]: sources: List[Tuple[str, str]] = [] for source_path, destination_path in BACKUP_SOURCES: - if os.path.isdir(source_path): - sources.append((source_path, destination_path)) + effective_source = ( + state_snapshot + if destination_path == "state" and state_snapshot is not None + else source_path + ) + if os.path.isdir(effective_source): + sources.append((effective_source, destination_path)) else: log(f"Skipping missing source: {source_path}") return sources @@ -1455,11 +1526,6 @@ def run_backup() -> int: return 0 policy = parse_retention_policy() - sources = available_sources() - if not sources: - log("No backup sources are available, skipping backup") - return 0 - progress_interval_seconds = parse_int_env( "BACKUP_PROGRESS_INTERVAL_SECONDS", DEFAULT_PROGRESS_INTERVAL_SECONDS, @@ -1474,8 +1540,19 @@ def run_backup() -> int: os.getenv("BACKUP_STATUS_FILE", DEFAULT_BACKUP_STATUS_FILE).strip() ) BACKUP_STATUS.begin(redact_destination(destination.raw_url)) - backend = build_backend(destination) + backend = None + state_snapshot_root = None try: + state_snapshot = None + if os.path.isdir(STATE_ROOT): + BACKUP_STATUS.write(message="Creating consistent state database snapshot") + state_snapshot_root, state_snapshot = prepare_state_snapshot() + sources = available_sources(state_snapshot) + if not sources: + log("No backup sources are available, skipping backup") + return 0 + + backend = build_backend(destination) log(f"Starting backup to {redact_destination(destination.raw_url)}") existing = set(backend.list_snapshots()) @@ -1494,7 +1571,7 @@ def run_backup() -> int: ) for source_path, destination_path in sources: - log(f"Syncing {source_path}") + log(f"Syncing {destination_path}") backend.sync_source( snapshot_name, source_path, @@ -1527,7 +1604,10 @@ def run_backup() -> int: ) return 0 finally: - backend.close() + if backend is not None: + backend.close() + if state_snapshot_root is not None: + shutil.rmtree(state_snapshot_root, ignore_errors=True) def with_lock() -> int: diff --git a/app/init.sh b/app/init.sh index 9d95fa8..c9354ed 100755 --- a/app/init.sh +++ b/app/init.sh @@ -584,7 +584,7 @@ if [[ -n "${JOIN_PASSWORD:-}" ]]; then export JOIN_PASSWORD fi -mkdir -p /data/private /data/fslogix /data/groups/data /data/groups/archive /state /state/audit /state/web /var/log/samba +mkdir -p /data/private /data/fslogix /data/groups/data /data/groups/archive /state /var/log/samba touch /var/log/reconcile.log /var/log/backup.log append_winbind_to_nss diff --git a/app/reconcile_shares.py b/app/reconcile_shares.py index 6bc433e..c981650 100755 --- a/app/reconcile_shares.py +++ b/app/reconcile_shares.py @@ -15,8 +15,13 @@ import uuid from dataclasses import dataclass, field from typing import Callable, Dict, Iterable, List, Optional, Set, Tuple +try: + from .state_db import STATE_DB_PATH, connect_state_db +except ImportError: + from state_db import STATE_DB_PATH, connect_state_db -DB_PATH = "/state/shares.db" + +DB_PATH = STATE_DB_PATH LOCK_PATH = "/state/reconcile.lock" GROUP_ROOT = "/data/groups/data" GROUP_ARCHIVE_ROOT = "/data/groups/archive" @@ -720,9 +725,7 @@ def ensure_share_db_schema(conn: sqlite3.Connection) -> None: def open_db() -> sqlite3.Connection: - os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) - conn = sqlite3.connect(DB_PATH) - conn.row_factory = sqlite3.Row + conn = connect_state_db(DB_PATH) conn.execute( """ CREATE TABLE IF NOT EXISTS shares ( diff --git a/app/state_db.py b/app/state_db.py new file mode 100644 index 0000000..22108af --- /dev/null +++ b/app/state_db.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +"""Connections to the shared persistent SQLite state database.""" + +import os +import sqlite3 +from typing import Optional + + +STATE_DB_PATH = os.getenv( + "STATE_DB_PATH", + os.getenv("SHARE_DB_PATH", "/state/shares.db"), +) + + +def connect_state_db( + path: Optional[str] = None, + *, + read_only: bool = False, +) -> sqlite3.Connection: + database_path = path or STATE_DB_PATH + if read_only: + conn = sqlite3.connect( + f"file:{database_path}?mode=ro", + uri=True, + timeout=30, + ) + else: + directory = os.path.dirname(database_path) + if directory: + os.makedirs(directory, exist_ok=True) + conn = sqlite3.connect(database_path, timeout=30) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA busy_timeout = 30000") + conn.execute("PRAGMA foreign_keys = ON") + if not read_only: + conn.execute("PRAGMA journal_mode = WAL") + conn.execute("PRAGMA synchronous = NORMAL") + return conn + + +def ensure_web_cache_schema(conn: sqlite3.Connection) -> None: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS web_cache ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """ + ) + conn.commit() + + +def drop_legacy_web_cache(directory: Optional[str] = None) -> int: + legacy_directory = directory or os.getenv("WEB_STATE_DIR", "/state/web") + removed = 0 + for name in ("usage.json", "usage.json.tmp"): + path = os.path.join(legacy_directory, name) + try: + os.remove(path) + removed += 1 + except FileNotFoundError: + pass + try: + os.rmdir(legacy_directory) + except OSError: + pass + return removed + diff --git a/app/web/app.js b/app/web/app.js index 422febf..69aa8ad 100644 --- a/app/web/app.js +++ b/app/web/app.js @@ -306,12 +306,12 @@ async function renderSystem() {

Dienstprüfungen

${esc(data.hostname)}
Domänenvertrauen${data.checks.domainTrust ? badge("In Ordnung") : badge("Fehlgeschlagen", "error")}
Samba-Konfiguration${data.checks.sambaConfig ? badge("Gültig") : badge("Fehlgeschlagen", "error")}
-
Aktivitätsarchiv${data.audit.days} Tage · ${bytes(data.audit.bytes)}
+
Aktivitätsdatenbank${data.audit.days} Tage · ${bytes(data.audit.bytes)}
Speicherprüfung${usage.scannedAt ? esc(utcTime(usage.scannedAt)) : "Ausstehend"}

TLS-Zertifikat

Allgemeiner Name
${esc(data.tls.subject?.commonName || "—")}
Aussteller
${esc(data.tls.issuer?.commonName || "—")}
Gültig bis
${esc(utcTime(data.tls.notAfter))}
Namen
${esc((data.tls.sans || []).map(value => value[1]).join(", ") || "—")}
-

Protokollaufbewahrung

Ältester UTC-Tag
${esc(data.audit.oldest || "—")}
Neuester UTC-Tag
${esc(data.audit.newest || "—")}
Archivgröße
${bytes(data.audit.bytes)}

Abgeschlossene Tagesdateien werden automatisch komprimiert und aufbewahrt, bis ein Administrator sie entfernt.

+

Protokollspeicherung

Ältester UTC-Tag
${esc(data.audit.oldest || "—")}
Neuester UTC-Tag
${esc(data.audit.newest || "—")}
Datenbankgröße
${bytes(data.audit.bytes)}

Ereignisse werden dauerhaft und indexiert in der gemeinsamen SQLite-Datenbank gespeichert.

`; } diff --git a/app/web_ui.py b/app/web_ui.py index 2e9ce0b..8e79e35 100644 --- a/app/web_ui.py +++ b/app/web_ui.py @@ -3,7 +3,6 @@ import base64 import datetime as dt -import gzip import hashlib import hmac import http.cookies @@ -23,7 +22,7 @@ import urllib.parse from collections import deque from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from typing import Dict, Iterable, List, Optional, Tuple +from typing import Dict, List, Optional, Tuple try: from app import reconcile_shares as directory @@ -31,20 +30,37 @@ except ImportError: # Container execution uses /app as the import root. import reconcile_shares as directory try: - from app.audit_policy import AUDIT_ACTIONS, action_for, skip_user + from app.audit_store import ( + audit_summary as sqlite_audit_summary, + ensure_audit_schema, + query_activity, + ) + from app.state_db import ( + STATE_DB_PATH, + connect_state_db, + drop_legacy_web_cache, + ensure_web_cache_schema, + ) except ImportError: # Container execution uses /app as the import root. - from audit_policy import AUDIT_ACTIONS, action_for, skip_user + from audit_store import ( + audit_summary as sqlite_audit_summary, + ensure_audit_schema, + query_activity, + ) + from state_db import ( + STATE_DB_PATH, + connect_state_db, + drop_legacy_web_cache, + ensure_web_cache_schema, + ) -STATE_ROOT = os.getenv("WEB_STATE_DIR", "/state/web") -AUDIT_ROOT = os.getenv("AUDIT_ARCHIVE_DIR", "/state/audit") STATIC_ROOT = os.getenv("WEB_STATIC_DIR", "/app/web") -SHARE_DB = os.getenv("SHARE_DB_PATH", "/state/shares.db") +STATE_DB = STATE_DB_PATH BACKUP_STATUS_FILE = os.getenv("BACKUP_STATUS_FILE", "/state/backup-status.json") BACKUP_LOG_FILE = os.getenv("BACKUP_LOG_FILE", "/var/log/backup.log") TLS_CERT_FILE = os.getenv("WEB_TLS_CERT_FILE", "/state/tls/web.crt") TLS_KEY_FILE = os.getenv("WEB_TLS_KEY_FILE", "/state/tls/web.key") -USAGE_CACHE_FILE = os.path.join(STATE_ROOT, "usage.json") JWT_COOKIE = "adfs_session" JWT_ISSUER = "ad-file-server-web" JWT_AUDIENCE = "domain-admins" @@ -69,16 +85,6 @@ def env_int(name: str, default: int, minimum: int, maximum: int) -> int: return default -def atomic_json(path: str, value: object) -> None: - os.makedirs(os.path.dirname(path), exist_ok=True) - temp = f"{path}.tmp" - with open(temp, "w", encoding="utf-8") as handle: - json.dump(value, handle, separators=(",", ":"), sort_keys=True) - handle.flush() - os.fsync(handle.fileno()) - os.replace(temp, path) - - def read_json(path: str, default): try: with open(path, encoding="utf-8") as handle: @@ -316,11 +322,49 @@ def scan_children(root: str) -> List[Dict[str, object]]: return rows +def load_usage_cache() -> Dict[str, object]: + try: + conn = connect_state_db(STATE_DB) + try: + ensure_web_cache_schema(conn) + row = conn.execute( + "SELECT value FROM web_cache WHERE key = 'usage'" + ).fetchone() + value = json.loads(str(row[0])) if row is not None else {} + return value if isinstance(value, dict) else {} + finally: + conn.close() + except (sqlite3.Error, ValueError): + return {} + + +def store_usage_cache(value: Dict[str, object]) -> None: + conn = connect_state_db(STATE_DB) + try: + ensure_web_cache_schema(conn) + with conn: + conn.execute( + """ + INSERT INTO web_cache (key, value, updated_at) + VALUES ('usage', ?, ?) + ON CONFLICT(key) DO UPDATE SET + value = excluded.value, + updated_at = excluded.updated_at + """, + ( + json.dumps(value, separators=(",", ":"), sort_keys=True), + now_utc().isoformat(timespec="seconds"), + ), + ) + finally: + conn.close() + + class UsageScanner: def __init__(self): self.interval = env_int("WEB_USAGE_SCAN_INTERVAL_SECONDS", 900, 60, 86400) self.lock = threading.Lock() - self.data = read_json(USAGE_CACHE_FILE, {}) + self.data = load_usage_cache() self.stop = threading.Event() def snapshot(self) -> Dict[str, object]: @@ -368,7 +412,7 @@ class UsageScanner: "fslogixBytes": sum(int(row["fslogixBytes"]) for row in user_rows), }, } - atomic_json(USAGE_CACHE_FILE, value) + store_usage_cache(value) with self.lock: self.data = value return value @@ -445,7 +489,7 @@ class DirectoryCache: folder_map = {} try: - conn = sqlite3.connect(f"file:{SHARE_DB}?mode=ro", uri=True) + conn = connect_state_db(STATE_DB, read_only=True) try: folder_map = { row[0]: {"folder": os.path.basename(row[1]), "active": bool(row[2])} @@ -515,60 +559,6 @@ class DirectoryCache: } -def iter_audit_file(path: str) -> Iterable[Dict[str, object]]: - opener = gzip.open if path.endswith(".gz") else open - try: - with opener(path, "rt", encoding="utf-8", errors="replace") as handle: - for line in handle: - try: - value = json.loads(line) - if isinstance(value, dict): - yield value - except json.JSONDecodeError: - continue - except OSError: - return - - -def iter_audit_file_reverse(path: str) -> Iterable[Dict[str, object]]: - if path.endswith(".gz"): - yield from reversed(list(iter_audit_file(path))) - return - try: - with open(path, "rb") as handle: - position = handle.seek(0, os.SEEK_END) - remainder = b"" - while position > 0: - size = min(1024 * 1024, position) - position -= size - handle.seek(position) - parts = (handle.read(size) + remainder).split(b"\n") - remainder = parts[0] - for line in reversed(parts[1:]): - if not line: - continue - try: - value = json.loads(line.decode("utf-8", errors="replace")) - if isinstance(value, dict): - yield value - except (json.JSONDecodeError, UnicodeDecodeError): - continue - if remainder: - try: - value = json.loads(remainder.decode("utf-8", errors="replace")) - if isinstance(value, dict): - yield value - except (json.JSONDecodeError, UnicodeDecodeError): - pass - except OSError: - return - - -def date_range(start: dt.date, end: dt.date): - day = start - while day <= end: - yield day.isoformat() - day += dt.timedelta(days=1) def query_audit(params: Dict[str, List[str]]) -> Dict[str, object]: @@ -582,62 +572,11 @@ def query_audit(params: Dict[str, List[str]]) -> Dict[str, object]: max_days = env_int("AUDIT_QUERY_MAX_DAYS", 31, 1, 366) if end < start or (end - start).days >= max_days: raise ValueError(f"Der Datumsbereich darf höchstens {max_days} Tage umfassen") - limit = min(500, max(1, int(params.get("limit", ["100"])[0]))) - offset = max(0, int(params.get("cursor", ["0"])[0])) - filters = { - key: params.get(key, [""])[0].casefold().strip() - for key in ("user", "share", "operation", "action", "path", "result") - } - page = [] - matched = 0 - facets = {"users": set(), "shares": set(), "operations": set(), "actions": set()} - for day in reversed(list(date_range(start, end))): - candidates = [os.path.join(AUDIT_ROOT, f"{day}.jsonl"), os.path.join(AUDIT_ROOT, f"{day}.jsonl.gz")] - for path in candidates: - if not os.path.isfile(path): - continue - for event in iter_audit_file_reverse(path): - user = str(event.get("user", "")) - operation = str(event.get("operation", "")) - action = action_for(operation) - if action is None and not operation: - stored_action = str(event.get("action", "")).casefold() - action = stored_action if stored_action in AUDIT_ACTIONS else None - if action is None or skip_user(user): - continue - if event.get("action") != action: - event = {**event, "action": action} - facets["users"].add(user) - facets["shares"].add(str(event.get("share", ""))) - facets["operations"].add(operation) - facets["actions"].add(action) - failed_filter = filters["result"] == "fail" - if failed_filter and bool(event.get("success", False)): - continue - if ( - filters["result"] - and not failed_filter - and filters["result"] - not in str(event.get("result", "")).casefold() - ): - continue - if any( - value and value not in str(event.get(key, "")).casefold() - for key, value in filters.items() - if key != "result" - ): - continue - if matched >= offset and len(page) < limit: - page.append(event) - matched += 1 - page.sort(key=lambda event: str(event.get("timestamp", "")), reverse=True) - next_cursor = offset + limit if offset + limit < matched else None - return { - "events": page, - "nextCursor": next_cursor, - "matched": matched, - "facets": {key: sorted(value, key=str.casefold) for key, value in facets.items()}, - } + conn = connect_state_db(STATE_DB, read_only=True) + try: + return query_activity(conn, start, end, params) + finally: + conn.close() def tail_lines(path: str, count: int) -> List[str]: @@ -666,7 +605,7 @@ def backup_payload() -> Dict[str, object]: def share_count() -> int: try: - conn = sqlite3.connect(f"file:{SHARE_DB}?mode=ro", uri=True) + conn = connect_state_db(STATE_DB, read_only=True) try: return int(conn.execute("SELECT count(*) FROM shares WHERE isActive = 1").fetchone()[0]) finally: @@ -676,21 +615,14 @@ def share_count() -> int: def audit_archive_summary() -> Dict[str, object]: - files = [] - total_bytes = 0 try: - names = os.listdir(AUDIT_ROOT) - except OSError: - names = [] - for name in names: - if re.match(r"^\d{4}-\d{2}-\d{2}\.jsonl(?:\.gz)?$", name): - path = os.path.join(AUDIT_ROOT, name) - try: - total_bytes += os.path.getsize(path) - files.append(name) - except OSError: - continue - return {"days": len(files), "bytes": total_bytes, "oldest": min(files)[:10] if files else None, "newest": max(files)[:10] if files else None} + conn = connect_state_db(STATE_DB, read_only=True) + try: + return sqlite_audit_summary(conn, STATE_DB) + finally: + conn.close() + except sqlite3.Error: + return {"days": 0, "bytes": 0, "oldest": None, "newest": None} def tls_summary() -> Dict[str, object]: @@ -711,7 +643,16 @@ class App: def __init__(self): secret = os.environ.get("WEB_JWT_SECRET", "") self.tokens = TokenManager(secret, env_int("WEB_JWT_TTL_SECONDS", 28800, 300, 604800)) + removed = drop_legacy_web_cache() + if removed: + log(f"Dropped {removed} legacy web cache file(s)") self.usage = UsageScanner() + conn = connect_state_db(STATE_DB) + try: + ensure_audit_schema(conn) + ensure_web_cache_schema(conn) + finally: + conn.close() self.directory = DirectoryCache() threading.Thread(target=self.usage.run, name="usage-scanner", daemon=True).start() @@ -916,7 +857,6 @@ def serve_https() -> None: def main() -> int: global APP - os.makedirs(STATE_ROOT, mode=0o750, exist_ok=True) if not os.path.isfile(TLS_CERT_FILE) or not os.path.isfile(TLS_KEY_FILE): raise RuntimeError("TLS certificate or key is missing") APP = App() diff --git a/dev/e2e.py b/dev/e2e.py index 33171db..c4d0500 100755 --- a/dev/e2e.py +++ b/dev/e2e.py @@ -2,7 +2,6 @@ """End-to-end checks for the disposable preview domain and file server.""" import base64 -import datetime as dt import json import os import socket @@ -302,25 +301,64 @@ def main() -> int: check(len(one_event.get("events", [])) == 1, "activity limit was ignored") check(one_event.get("nextCursor") is not None, "activity pagination cursor missing") - announce("closed daily log compression and querying gzip history") - old_day = (dt.datetime.now(dt.timezone.utc).date() - dt.timedelta(days=2)).isoformat() - eventually( - "historical audit gzip", - lambda: engine_run( - "exec", FILES_CONTAINER, "test", "-f", f"/state/audit/{old_day}.jsonl.gz", - check_result=False, - ).returncode, - lambda returncode: returncode == 0, - timeout=30, + announce("shared SQLite state, indexes, integrity, and ordered read deduplication") + integrity = engine_run( + "exec", FILES_CONTAINER, "sqlite3", "/state/shares.db", "PRAGMA quick_check;" ) - archived = http( - query_path( - "/api/activity", - {"from": old_day, "to": old_day, "user": "archived-user", "limit": "10"}, - ), - token=token, + check(integrity.stdout.strip() == "ok", "shared SQLite database failed quick_check") + tables = set( + engine_run( + "exec", + FILES_CONTAINER, + "sqlite3", + "/state/shares.db", + "SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name;", + ).stdout.splitlines() ) - check(archived.status == 200 and archived.json().get("matched") == 1, "compressed audit history is not searchable") + check( + {"shares", "audit_events", "audit_sources", "web_cache"}.issubset(tables), + f"shared SQLite tables are incomplete: {sorted(tables)}", + ) + indexes = set( + engine_run( + "exec", + FILES_CONTAINER, + "sqlite3", + "/state/shares.db", + "SELECT name FROM sqlite_schema WHERE type='index' AND name LIKE 'audit_events_%';", + ).stdout.splitlines() + ) + check( + {"audit_events_time", "audit_events_user_time", "audit_events_action_time"}.issubset(indexes), + f"audit indexes are incomplete: {sorted(indexes)}", + ) + duplicate_reads = engine_run( + "exec", + FILES_CONTAINER, + "sqlite3", + "/state/shares.db", + """SELECT count(*) FROM ( + SELECT action, occurred_second, user, client_ip, share, path, success, result, + lag(action) OVER (ORDER BY id) AS previous_action, + lag(occurred_second) OVER (ORDER BY id) AS previous_second, + lag(user) OVER (ORDER BY id) AS previous_user, + lag(client_ip) OVER (ORDER BY id) AS previous_client_ip, + lag(share) OVER (ORDER BY id) AS previous_share, + lag(path) OVER (ORDER BY id) AS previous_path, + lag(success) OVER (ORDER BY id) AS previous_success, + lag(result) OVER (ORDER BY id) AS previous_result + FROM audit_events + ) WHERE action='read' AND previous_action='read' + AND occurred_second=previous_second AND user=previous_user + AND client_ip=previous_client_ip AND share=previous_share + AND path=previous_path AND success=previous_success + AND (success=1 OR result=previous_result);""", + ) + check(duplicate_reads.stdout.strip() == "0", "uninterrupted duplicate reads remain") + legacy_archive = engine_run( + "exec", FILES_CONTAINER, "test", "!", "-e", "/state/audit", check_result=False + ) + check(legacy_archive.returncode == 0, "legacy JSONL audit archive still exists") announce("real rsync backup, status API, log tail, and remote completion marker") backup = eventually( @@ -350,7 +388,7 @@ def main() -> int: check(system_payload.get("checks", {}).get("domainTrust") is True, "system reports broken AD trust") check(system_payload.get("checks", {}).get("sambaConfig") is True, "system reports invalid Samba config") check(system_payload.get("tls", {}).get("sans"), "system TLS summary is empty") - check(system_payload.get("audit", {}).get("days", 0) >= 2, "system audit archive summary is incomplete") + check(system_payload.get("audit", {}).get("days", 0) >= 1, "system activity database summary is incomplete") logout = http("/api/logout", method="POST", value={}, token=token) check(logout.status == 200 and "Max-Age=0" in logout.headers.get("Set-Cookie", ""), "logout did not clear session cookie") diff --git a/dev/preview-client.sh b/dev/preview-client.sh index 3f44fa5..5055cc1 100755 --- a/dev/preview-client.sh +++ b/dev/preview-client.sh @@ -35,7 +35,7 @@ fi printf 'Generated by the preview SMB client.\n' > /tmp/live-note.txt smb alice Data 'cd Finance; mkdir Reports; cd Reports; put /tmp/live-note.txt live-note.txt; ls' >/dev/null -smb alice Data 'cd Finance; cd Reports; put /tmp/live-note.txt audit-source.txt; get audit-source.txt /tmp/audit-readback.txt; rename audit-source.txt audit-moved.txt; del audit-moved.txt' >/dev/null +smb alice Data 'cd Finance; cd Reports; put /tmp/live-note.txt audit-source.txt; get audit-source.txt /tmp/audit-readback-1.txt; get audit-source.txt /tmp/audit-readback-2.txt; get audit-source.txt /tmp/audit-readback-3.txt; rename audit-source.txt audit-moved.txt; del audit-moved.txt' >/dev/null smb carol Data 'cd Engineering; mkdir Designs; cd Designs; put /tmp/live-note.txt architecture.txt; ls' >/dev/null smb alice Private 'cd alice; put /tmp/live-note.txt notes.txt; ls' >/dev/null smb report_svc Data 'cd Finance; cd Reports; put /tmp/live-note.txt ignored-service-event.txt; get ignored-service-event.txt /tmp/ignored-service-readback.txt; del ignored-service-event.txt' >/dev/null diff --git a/dev/seed-files.sh b/dev/seed-files.sh index 2cf67a2..a3c4e37 100755 --- a/dev/seed-files.sh +++ b/dev/seed-files.sh @@ -13,8 +13,7 @@ mkdir -p \ /data/private/dave \ /data/private/eve \ /data/fslogix/alice_S-1-5-21-111-222-333-1101 \ - /data/fslogix/carol_S-1-5-21-111-222-333-1103 \ - /state/audit + /data/fslogix/carol_S-1-5-21-111-222-333-1103 printf 'Quarter,Forecast\nQ1,120000\nQ2,135000\n' > /data/groups/data/Finance/Reports/forecast.csv printf '# Architecture\n\nPreview design notes.\n' > /data/groups/data/Engineering/Designs/architecture.md @@ -29,9 +28,5 @@ dd if=/dev/zero of=/data/groups/data/Finance/Reports/history.bin bs=1M count="$s dd if=/dev/zero of=/data/groups/data/Engineering/Designs/models.bin bs=1M count="$seed_mb" status=none dd if=/dev/zero of=/data/fslogix/alice_S-1-5-21-111-222-333-1101/profile-data.bin bs=1M count="$seed_mb" status=none -old_day=$(date -u -d '2 days ago' +%F) -printf '%s\n' "{\"action\":\"read\",\"client\":\"archived-client\",\"clientIp\":\"192.0.2.50\",\"ingestedAt\":\"${old_day}T12:00:00+00:00\",\"operation\":\"read\",\"path\":\"Finance/Reports/archive.csv\",\"result\":\"OK\",\"share\":\"Data\",\"source\":\"log.archived-client\",\"success\":true,\"timestamp\":\"${old_day}T12:00:00+00:00\",\"user\":\"archived-user\"}" \ - > "/state/audit/${old_day}.jsonl" -touch -d '2 days ago' "/state/audit/${old_day}.jsonl" -printf '[preview-seed] Seeded group, private, FSLogix, and historical audit data.\n' +printf '[preview-seed] Seeded group, private, and FSLogix data.\n' diff --git a/scripts/dev b/scripts/dev index 072ed0a..7fb16cd 100755 --- a/scripts/dev +++ b/scripts/dev @@ -279,7 +279,7 @@ if ! wait_for_exec "$dc_container" 150 test -f /run/preview-ready; then fi domain_sid=$("$engine" exec "$dc_container" cat /run/domain-sid) -printf 'seeding preview files and an old audit archive\n' +printf 'seeding preview files\n' "$engine" run --rm \ --network "$network_name" \ -e "DEV_SEED_MB=${DEV_SEED_MB:-8}" \ @@ -325,7 +325,6 @@ printf 'starting actual file server and HTTPS web UI\n' -e "WEB_USAGE_SCAN_INTERVAL_SECONDS=60" \ -e "WEB_DIRECTORY_CACHE_SECONDS=30" \ -e "AUDIT_POLL_SECONDS=0.25" \ - -e "AUDIT_COMPRESS_AFTER_HOURS=1" \ -e "BACKUP_DESTINATION=rsync://${dev_backup_user}:${dev_backup_password}@backup.${dev_dns_domain}/backups/fileserver" \ -e "BACKUP_PROGRESS=never" \ -e "BACKUP_PROGRESS_INTERVAL_SECONDS=1" \ diff --git a/setup b/setup index 5061b3d..cbaf5b2 100755 --- a/setup +++ b/setup @@ -411,7 +411,7 @@ ACME_HTTP_PORT=${acme_http_port} # WEB_JWT_TTL_SECONDS=28800 # WEB_USAGE_SCAN_INTERVAL_SECONDS=900 # WEB_DIRECTORY_CACHE_SECONDS=300 -# AUDIT_COMPRESS_AFTER_HOURS=24 +# STATE_DB_PATH=/state/shares.db # AUDIT_QUERY_MAX_DAYS=31 # AUDIT_SKIP_USER_SUFFIXES=_svc,_ServiceAcc # ACME_RENEW_CHECK_SECONDS=900 diff --git a/tests/test_backup_to_destination.py b/tests/test_backup_to_destination.py index 93babad..72a44c2 100644 --- a/tests/test_backup_to_destination.py +++ b/tests/test_backup_to_destination.py @@ -1,5 +1,7 @@ import io import os +import shutil +import sqlite3 import tempfile import unittest from unittest import mock @@ -186,6 +188,56 @@ class ProgressParsingTests(unittest.TestCase): self.assertEqual(total, 8) +class StateSnapshotTests(unittest.TestCase): + def test_online_snapshot_includes_wal_commits_and_other_state(self): + with tempfile.TemporaryDirectory() as tmpdir: + state_root = os.path.join(tmpdir, "state") + os.mkdir(state_root) + database = os.path.join(state_root, "shares.db") + with open(os.path.join(state_root, "backup-status.json"), "w", encoding="utf-8") as handle: + handle.write("status") + + writer = sqlite3.connect(database) + writer.execute("PRAGMA journal_mode = WAL") + writer.execute("CREATE TABLE events (id INTEGER PRIMARY KEY, value TEXT)") + writer.execute("INSERT INTO events (value) VALUES ('committed-in-wal')") + writer.commit() + snapshot_root = None + try: + snapshot_root, staged_state = backup.prepare_state_snapshot( + state_root, + database, + ) + staged_database = os.path.join(staged_state, "shares.db") + reader = sqlite3.connect(staged_database) + try: + self.assertEqual( + reader.execute("SELECT value FROM events").fetchone()[0], + "committed-in-wal", + ) + self.assertEqual(reader.execute("PRAGMA quick_check").fetchone()[0], "ok") + finally: + reader.close() + with open(os.path.join(staged_state, "backup-status.json"), encoding="utf-8") as handle: + self.assertEqual(handle.read(), "status") + self.assertFalse(os.path.exists(f"{staged_database}-wal")) + self.assertFalse(os.path.exists(f"{staged_database}-shm")) + finally: + writer.close() + if snapshot_root is not None: + shutil.rmtree(snapshot_root, ignore_errors=True) + + def test_state_database_must_be_inside_state_root(self): + with tempfile.TemporaryDirectory() as tmpdir: + state_root = os.path.join(tmpdir, "state") + os.mkdir(state_root) + database = os.path.join(tmpdir, "outside.db") + sqlite3.connect(database).close() + + with self.assertRaisesRegex(RuntimeError, "below STATE_ROOT"): + backup.prepare_state_snapshot(state_root, database) + + class BackendProgressCommandTests(unittest.TestCase): def test_rclone_sync_uses_progress_flags(self): destination = backup.parse_destination("sftp://user@example.com/backups") diff --git a/tests/test_web_ui.py b/tests/test_web_ui.py index 5b1ec1c..4190677 100644 --- a/tests/test_web_ui.py +++ b/tests/test_web_ui.py @@ -1,13 +1,14 @@ import datetime as dt -import gzip -import json import os +import sqlite3 import tempfile import unittest from unittest import mock from app import audit_collector +from app import audit_store from app import web_ui +from app import state_db class TokenManagerTests(unittest.TestCase): @@ -132,9 +133,8 @@ class AuditParsingTests(unittest.TestCase): def test_tracks_rotated_file_by_inode_without_reingesting_it(self): with tempfile.TemporaryDirectory() as tmpdir: - archive = os.path.join(tmpdir, "audit") - os.mkdir(archive) active = os.path.join(tmpdir, "log.pc01") + database = os.path.join(tmpdir, "state.db") line = ( "[2026/07/31 12:34:56.000000, 1] smbd_audit: " "x|alice|192.0.2.5|PC01|Data|pread|OK|a.txt\n" @@ -142,61 +142,252 @@ class AuditParsingTests(unittest.TestCase): with open(active, "w", encoding="utf-8") as handle: handle.write(line) - with mock.patch.object(audit_collector, "SAMBA_LOG_GLOB", os.path.join(tmpdir, "log.*")), mock.patch.object(audit_collector, "ARCHIVE_DIR", archive), mock.patch.object(audit_collector, "STATE_FILE", os.path.join(archive, "state.json")): - state = {} - self.assertEqual(audit_collector.collect_once(state), 1) - rotated = f"{active}.old" - os.rename(active, rotated) - with open(active, "w", encoding="utf-8") as handle: - handle.write(line.replace("a.txt", "b.txt")) - self.assertEqual(audit_collector.collect_once(state), 1) + store = audit_store.AuditStore(database) + try: + with mock.patch.object( + audit_collector, + "SAMBA_LOG_GLOB", + os.path.join(tmpdir, "log.*"), + ): + self.assertEqual(audit_collector.collect_once(store), 1) + rotated = f"{active}.old" + os.rename(active, rotated) + with open(active, "w", encoding="utf-8") as handle: + handle.write(line.replace("a.txt", "b.txt")) + self.assertEqual(audit_collector.collect_once(store), 1) + paths = [ + row[0] + for row in store.conn.execute( + "SELECT path FROM audit_events ORDER BY id" + ) + ] + self.assertEqual(paths, ["a.txt", "b.txt"]) + finally: + store.close() + + def test_processes_known_rotated_inode_before_new_active_file(self): + with tempfile.TemporaryDirectory() as tmpdir: + active = os.path.join(tmpdir, "log.pc01") + database = os.path.join(tmpdir, "state.db") + read = ( + "[2026/07/31 12:34:56.000000, 1] smbd_audit: " + "x|alice|192.0.2.5|PC01|Data|pread|OK|a.txt\n" + ) + write = read.replace("pread|OK|a.txt", "pwrite|OK|changed.txt") + with open(active, "w", encoding="utf-8") as handle: + handle.write(read) + + store = audit_store.AuditStore(database) + try: + with mock.patch.object( + audit_collector, + "SAMBA_LOG_GLOB", + os.path.join(tmpdir, "log.*"), + ): + self.assertEqual(audit_collector.collect_once(store), 1) + with open(active, "a", encoding="utf-8") as handle: + handle.write(read) + os.rename(active, f"{active}.old") + with open(active, "w", encoding="utf-8") as handle: + handle.write(write) + self.assertEqual(audit_collector.collect_once(store), 1) + + rows = store.conn.execute( + "SELECT action, path FROM audit_events ORDER BY id" + ).fetchall() + self.assertEqual( + [(row["action"], row["path"]) for row in rows], + [("read", "a.txt"), ("write", "changed.txt")], + ) + finally: + store.close() + + def test_deduplicates_only_uninterrupted_identical_reads_in_one_second(self): + with tempfile.TemporaryDirectory() as tmpdir: + active = os.path.join(tmpdir, "log.pc01") + database = os.path.join(tmpdir, "state.db") + + def line(operation, path, second="12:34:56"): + return ( + f"[2026/07/31 {second}.000000, 1] smbd_audit: " + f"x|alice|192.0.2.5|PC01|Data|{operation}|OK|{path}\n" + ) + + with open(active, "w", encoding="utf-8") as handle: + handle.writelines( + [ + line("pread", "a.txt"), + line("pread", "a.txt"), + line("pread", "b.txt"), + line("pread", "a.txt"), + line("pread", "a.txt"), + ] + ) + + store = audit_store.AuditStore(database) + try: + with mock.patch.object( + audit_collector, + "SAMBA_LOG_GLOB", + os.path.join(tmpdir, "log.*"), + ): + self.assertEqual(audit_collector.collect_once(store), 3) + with open(active, "a", encoding="utf-8") as handle: + handle.write(line("pread", "a.txt")) + self.assertEqual(audit_collector.collect_once(store), 0) + with open(active, "a", encoding="utf-8") as handle: + handle.write(line("pwrite", "changed.txt")) + handle.write(line("pread", "a.txt")) + self.assertEqual(audit_collector.collect_once(store), 2) + with open(active, "a", encoding="utf-8") as handle: + handle.write(line("pread", "a.txt", "12:34:57")) + self.assertEqual(audit_collector.collect_once(store), 1) + + rows = store.conn.execute( + "SELECT action, path, occurred_at FROM audit_events ORDER BY id" + ).fetchall() + self.assertEqual( + [(row["action"], row["path"]) for row in rows], + [ + ("read", "a.txt"), + ("read", "b.txt"), + ("read", "a.txt"), + ("write", "changed.txt"), + ("read", "a.txt"), + ("read", "a.txt"), + ], + ) + self.assertTrue(rows[-1]["occurred_at"].endswith("12:34:57+00:00")) + finally: + store.close() class AuditQueryTests(unittest.TestCase): def make_event(self, timestamp, user, success=True): return { "timestamp": timestamp, + "ingestedAt": timestamp, "user": user, "clientIp": "192.0.2.5", + "client": "PC01", "share": "Data", "operation": "pread" if success else "unlinkat", "action": "read" if success else "delete", "path": "folder/file.txt", "result": "OK" if success else "NT_STATUS_ACCESS_DENIED", "success": success, + "source": "log.pc01", } - def test_queries_plain_and_compressed_days_with_filters_and_cursor(self): + def test_queries_indexed_events_with_filters_and_keyset_cursor(self): with tempfile.TemporaryDirectory() as tmpdir: today = dt.datetime.now(dt.timezone.utc).date() yesterday = today - dt.timedelta(days=1) - current = os.path.join(tmpdir, f"{today.isoformat()}.jsonl") - old = os.path.join(tmpdir, f"{yesterday.isoformat()}.jsonl.gz") - with open(current, "w", encoding="utf-8") as handle: - for index in range(3): - handle.write(json.dumps(self.make_event(f"{today}T12:00:0{index}+00:00", "alice")) + "\n") - handle.write(json.dumps(self.make_event(f"{today}T12:00:04+00:00", "bob", False)) + "\n") - ignored = self.make_event(f"{today}T12:00:05+00:00", "metadata-user") - ignored.update({"operation": "create_file", "action": "write"}) - handle.write(json.dumps(ignored) + "\n") - handle.write(json.dumps(self.make_event(f"{today}T12:00:06+00:00", "robot_svc")) + "\n") - with gzip.open(old, "wt", encoding="utf-8") as handle: - handle.write(json.dumps(self.make_event(f"{yesterday}T12:00:00+00:00", "alice")) + "\n") + database = os.path.join(tmpdir, "state.db") + store = audit_store.AuditStore(database) + events = [ + self.make_event(f"{today}T12:00:0{index}+00:00", "alice") + for index in range(3) + ] + events.extend( + [ + self.make_event(f"{today}T12:00:04+00:00", "bob", False), + self.make_event(f"{today}T12:00:06+00:00", "robot_svc"), + self.make_event(f"{yesterday}T12:00:00+00:00", "alice"), + ] + ) + store.append_batch(events, {}, set()) + store.close() - with mock.patch.object(web_ui, "AUDIT_ROOT", tmpdir): - first = web_ui.query_audit({"from": [yesterday.isoformat()], "to": [today.isoformat()], "user": ["alice"], "limit": ["2"]}) - failed = web_ui.query_audit({"from": [today.isoformat()], "to": [today.isoformat()], "result": ["fail"]}) - second = web_ui.query_audit({"from": [yesterday.isoformat()], "to": [today.isoformat()], "user": ["alice"], "limit": ["2"], "cursor": [str(first["nextCursor"])]}) - visible = web_ui.query_audit({"from": [today.isoformat()], "to": [today.isoformat()], "limit": ["100"]}) + with mock.patch.object(web_ui, "STATE_DB", database): + first = web_ui.query_audit( + { + "from": [yesterday.isoformat()], + "to": [today.isoformat()], + "user": ["alice"], + "limit": ["2"], + } + ) + failed = web_ui.query_audit( + { + "from": [today.isoformat()], + "to": [today.isoformat()], + "result": ["fail"], + } + ) + second = web_ui.query_audit( + { + "from": [yesterday.isoformat()], + "to": [today.isoformat()], + "user": ["alice"], + "limit": ["2"], + "cursor": [str(first["nextCursor"])], + } + ) + visible = web_ui.query_audit( + { + "from": [today.isoformat()], + "to": [today.isoformat()], + "limit": ["100"], + } + ) self.assertEqual(first["matched"], 4) self.assertEqual(len(first["events"]), 2) self.assertEqual(len(second["events"]), 2) self.assertEqual(failed["events"][0]["user"], "bob") - self.assertNotIn("metadata-user", visible["facets"]["users"]) self.assertNotIn("robot_svc", visible["facets"]["users"]) self.assertEqual(set(visible["facets"]["actions"]), {"read", "delete"}) + def test_schema_has_filter_and_time_indexes(self): + with tempfile.TemporaryDirectory() as tmpdir: + store = audit_store.AuditStore(os.path.join(tmpdir, "state.db")) + try: + indexes = { + row[0] + for row in store.conn.execute( + "SELECT name FROM sqlite_schema WHERE type = 'index'" + ) + } + finally: + store.close() + self.assertTrue( + { + "audit_events_time", + "audit_events_action_time", + "audit_events_success_time", + "audit_events_user_time", + "audit_events_account_time", + "audit_events_share_time", + "audit_events_result_time", + }.issubset(indexes) + ) + + def test_drops_only_known_legacy_audit_files(self): + with tempfile.TemporaryDirectory() as tmpdir: + for name in ( + "2026-07-30.jsonl", + "2026-07-29.jsonl.gz", + "collector-state.json", + "keep.txt", + ): + with open(os.path.join(tmpdir, name), "w", encoding="utf-8"): + pass + + self.assertEqual(audit_store.drop_legacy_audit_files(tmpdir), 3) + self.assertEqual(os.listdir(tmpdir), ["keep.txt"]) + + +class StateDatabaseMigrationTests(unittest.TestCase): + def test_drops_only_known_recomputable_web_cache_files(self): + with tempfile.TemporaryDirectory() as tmpdir: + for name in ("usage.json", "usage.json.tmp", "keep.txt"): + with open(os.path.join(tmpdir, name), "w", encoding="utf-8"): + pass + + self.assertEqual(state_db.drop_legacy_web_cache(tmpdir), 2) + self.assertEqual(os.listdir(tmpdir), ["keep.txt"]) + class WebPresentationTests(unittest.TestCase): def asset(self, name): @@ -283,16 +474,26 @@ class UsageScannerTests(unittest.TestCase): handle.write(b"b" * 3) with open(os.path.join(fslogix_root, "alice_S-1-5-21-1-2-3-1001", "c"), "wb") as handle: handle.write(b"c" * 5) - cache = os.path.join(tmpdir, "usage.json") + database = os.path.join(tmpdir, "state.db") env = {"GROUP_ROOT": group_root, "PRIVATE_ROOT": private_root, "FSLOGIX_ROOT": fslogix_root} - with mock.patch.dict(os.environ, env), mock.patch.object(web_ui, "USAGE_CACHE_FILE", cache), mock.patch.object(web_ui, "fslogix_username", return_value="alice"): + with mock.patch.dict(os.environ, env), mock.patch.object(web_ui, "STATE_DB", database), mock.patch.object(web_ui, "fslogix_username", return_value="alice"): value = web_ui.UsageScanner().scan() - self.assertEqual(value["totals"]["dataBytes"], 7) - self.assertEqual(value["users"][0]["privateBytes"], 3) - self.assertEqual(value["users"][0]["fslogixBytes"], 5) - self.assertEqual(value["users"][0]["totalBytes"], 8) + self.assertEqual(value["totals"]["dataBytes"], 7) + self.assertEqual(value["users"][0]["privateBytes"], 3) + self.assertEqual(value["users"][0]["fslogixBytes"], 5) + self.assertEqual(value["users"][0]["totalBytes"], 8) + conn = sqlite3.connect(database) + try: + self.assertEqual( + conn.execute( + "SELECT count(*) FROM web_cache WHERE key = 'usage'" + ).fetchone()[0], + 1, + ) + finally: + conn.close() if __name__ == "__main__":