250 lines
8.4 KiB
Python
250 lines
8.4 KiB
Python
"""
|
|
audit.py — Centralized audit log for NetAdmin
|
|
|
|
All tables live in the shared MariaDB database (cfg.database.name).
|
|
"""
|
|
|
|
import datetime
|
|
import json
|
|
import logging
|
|
import uuid
|
|
|
|
from db import get_db as _conn
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _init_tables():
|
|
"""Create audit and sync_history tables if they don't exist."""
|
|
try:
|
|
with _conn() as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS audit_log (
|
|
id VARCHAR(36) PRIMARY KEY,
|
|
ts DATETIME NOT NULL DEFAULT NOW(),
|
|
user_id VARCHAR(36),
|
|
username VARCHAR(64),
|
|
ip VARCHAR(45),
|
|
service VARCHAR(16) NOT NULL,
|
|
action VARCHAR(64) NOT NULL,
|
|
target VARCHAR(255),
|
|
detail TEXT,
|
|
success TINYINT NOT NULL DEFAULT 1,
|
|
error TEXT
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
|
""")
|
|
for idx_name, idx_col in [
|
|
("idx_audit_ts", "ts"),
|
|
("idx_audit_user", "user_id"),
|
|
("idx_audit_service", "service"),
|
|
]:
|
|
try:
|
|
conn.execute(f"CREATE INDEX {idx_name} ON audit_log({idx_col})")
|
|
except Exception:
|
|
pass # index already exists
|
|
except Exception as e:
|
|
log.warning(f"[audit] init_tables: {e}")
|
|
|
|
|
|
_init_tables()
|
|
|
|
|
|
def record(
|
|
service: str,
|
|
action: str,
|
|
target: str | None = None,
|
|
detail: dict | None = None,
|
|
success: bool = True,
|
|
error: str | None = None,
|
|
user: dict | None = None, # FastAPI user dict from get_current_user
|
|
ip: str | None = None,
|
|
) -> str:
|
|
"""Write one audit entry. Returns the entry id."""
|
|
entry_id = str(uuid.uuid4())
|
|
user_id = user.get("id") if user else None
|
|
username = user.get("username") if user else "system"
|
|
try:
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO audit_log
|
|
(id, user_id, username, ip, service, action, target, detail, success, error)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
|
|
(
|
|
entry_id,
|
|
user_id,
|
|
username,
|
|
ip,
|
|
service,
|
|
action,
|
|
target,
|
|
json.dumps(detail) if detail else None,
|
|
int(success),
|
|
error,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
# Audit must never break the main flow
|
|
log.warning(f"[audit] Failed to write log entry: {e}")
|
|
return entry_id
|
|
|
|
|
|
def query(
|
|
service: str | None = None,
|
|
username: str | None = None,
|
|
action: str | None = None,
|
|
limit: int = 200,
|
|
offset: int = 0,
|
|
) -> list[dict]:
|
|
"""Query audit log entries, newest first."""
|
|
conditions = []
|
|
params = []
|
|
if service:
|
|
conditions.append("service = %s"); params.append(service)
|
|
if username:
|
|
conditions.append("username LIKE %s"); params.append(f"%{username}%")
|
|
if action:
|
|
conditions.append("action = %s"); params.append(action)
|
|
|
|
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
|
params += [limit, offset]
|
|
|
|
try:
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT * FROM audit_log {where} ORDER BY ts DESC LIMIT %s OFFSET %s",
|
|
params
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
except Exception as e:
|
|
log.warning(f"[audit] Query failed: {e}")
|
|
return []
|
|
|
|
|
|
def count(
|
|
service: str | None = None,
|
|
username: str | None = None,
|
|
action: str | None = None,
|
|
) -> int:
|
|
conditions = []
|
|
params = []
|
|
if service:
|
|
conditions.append("service = %s"); params.append(service)
|
|
if username:
|
|
conditions.append("username LIKE %s"); params.append(f"%{username}%")
|
|
if action:
|
|
conditions.append("action = %s"); params.append(action)
|
|
|
|
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
|
try:
|
|
with _conn() as conn:
|
|
return conn.execute(
|
|
f"SELECT COUNT(*) AS n FROM audit_log {where}", params
|
|
).fetchone()["n"]
|
|
except Exception as e:
|
|
log.warning(f"[audit] Count failed: {e}")
|
|
return 0
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
# Sync history
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
|
|
def _ensure_sync_history():
|
|
"""Create sync_history table if not exists."""
|
|
try:
|
|
with _conn() as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS sync_history (
|
|
id VARCHAR(36) PRIMARY KEY,
|
|
ts DATETIME NOT NULL DEFAULT NOW(),
|
|
service VARCHAR(16) NOT NULL,
|
|
`trigger` VARCHAR(32) NOT NULL,
|
|
duration_ms INT,
|
|
zones INT DEFAULT 0,
|
|
records INT DEFAULT 0,
|
|
success TINYINT NOT NULL DEFAULT 1,
|
|
error TEXT,
|
|
detail TEXT
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
|
""")
|
|
try:
|
|
conn.execute("CREATE INDEX idx_synchist_ts ON sync_history(ts)")
|
|
except Exception:
|
|
pass # index already exists
|
|
except Exception as e:
|
|
log.warning(f"[audit] sync_history init failed: {e}")
|
|
|
|
|
|
_ensure_sync_history()
|
|
|
|
|
|
def record_sync(
|
|
service: str,
|
|
trigger: str,
|
|
duration_ms: int | None = None,
|
|
zones: int = 0,
|
|
records: int = 0,
|
|
success: bool = True,
|
|
error: str | None = None,
|
|
detail: dict | None = None,
|
|
) -> str:
|
|
"""Record one sync operation in sync_history."""
|
|
entry_id = str(uuid.uuid4())
|
|
try:
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO sync_history
|
|
(id, service, `trigger`, duration_ms, zones, records, success, error, detail)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
|
|
(
|
|
entry_id, service, trigger,
|
|
duration_ms, zones, records,
|
|
int(success), error,
|
|
json.dumps(detail) if detail else None,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
log.warning(f"[audit] Failed to write sync_history: {e}")
|
|
return entry_id
|
|
|
|
|
|
def query_sync_history(
|
|
service: str | None = None,
|
|
trigger: str | None = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
) -> list[dict]:
|
|
"""Return sync history entries, newest first."""
|
|
conditions, params = [], []
|
|
if service:
|
|
conditions.append("service=%s"); params.append(service)
|
|
if trigger:
|
|
conditions.append("`trigger`=%s"); params.append(trigger)
|
|
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
|
params += [limit, offset]
|
|
try:
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT * FROM sync_history {where} ORDER BY ts DESC LIMIT %s OFFSET %s",
|
|
params
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
except Exception as e:
|
|
log.warning(f"[audit] sync_history query failed: {e}")
|
|
return []
|
|
|
|
|
|
def count_sync_history(service: str | None = None) -> int:
|
|
conditions, params = [], []
|
|
if service:
|
|
conditions.append("service=%s"); params.append(service)
|
|
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
|
try:
|
|
with _conn() as conn:
|
|
return conn.execute(
|
|
f"SELECT COUNT(*) AS n FROM sync_history {where}", params
|
|
).fetchone()["n"]
|
|
except Exception as e:
|
|
log.warning(f"[audit] sync_history count failed: {e}")
|
|
return 0
|