604 lines
23 KiB
Python
604 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import sqlite3
|
|
import tempfile
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from ..core.config import settings
|
|
from ..core.version import APP_VERSION
|
|
from ..db.models import (
|
|
AppContainerMapping,
|
|
AppSetting,
|
|
AuditEvent,
|
|
DiscoveredContainer,
|
|
NotificationChannel,
|
|
NotificationRule,
|
|
Policy,
|
|
)
|
|
from .audit_logger import log_event
|
|
from .secret_store import secret_storage_status
|
|
from .setup_service import get_setup_status, list_connector_settings
|
|
|
|
_BACKUP_LOCK = threading.Lock()
|
|
_OPERATOR_AUTH_SETTING_KEY = "security.operator_auth"
|
|
|
|
|
|
def apply_stored_operator_auth_setting(session: Session) -> None:
|
|
stored = session.get(AppSetting, _OPERATOR_AUTH_SETTING_KEY)
|
|
if stored and isinstance(stored.value_json, dict) and isinstance(stored.value_json.get("required"), bool):
|
|
settings.operator_auth_required = stored.value_json["required"]
|
|
|
|
|
|
def operator_auth_policy() -> dict[str, bool]:
|
|
return {"required": settings.operator_auth_required}
|
|
|
|
|
|
def update_operator_auth_policy(session: Session, required: bool) -> dict[str, bool]:
|
|
stored = session.get(AppSetting, _OPERATOR_AUTH_SETTING_KEY)
|
|
if stored is None:
|
|
stored = AppSetting(key=_OPERATOR_AUTH_SETTING_KEY, is_secret=False)
|
|
stored.value_json = {"required": required}
|
|
stored.updated_at = datetime.now(timezone.utc)
|
|
session.add(stored)
|
|
settings.operator_auth_required = required
|
|
log_event(
|
|
session,
|
|
source="manual",
|
|
identity="operator",
|
|
action_id="security.operator_auth.update",
|
|
target="operator_session",
|
|
target_type="security_setting",
|
|
risk="medium",
|
|
policy_decision="allowed",
|
|
reason="Operator session requirement changed from Settings.",
|
|
result="success",
|
|
redacted_metadata={"required": required},
|
|
)
|
|
session.commit()
|
|
return operator_auth_policy()
|
|
|
|
|
|
def _data_dir() -> Path:
|
|
return Path(settings.data_dir)
|
|
|
|
|
|
def _static_dir() -> Path:
|
|
return Path(settings.static_dir)
|
|
|
|
|
|
def _backup_dir() -> Path:
|
|
return _data_dir() / "backups"
|
|
|
|
|
|
def is_data_dir_writable() -> bool:
|
|
try:
|
|
_data_dir().mkdir(parents=True, exist_ok=True)
|
|
probe = _data_dir() / ".appops-write-test"
|
|
probe.write_text("ok", encoding="utf-8")
|
|
probe.unlink(missing_ok=True)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def db_status(session: Session) -> str:
|
|
try:
|
|
session.exec(select(Policy).limit(1)).all()
|
|
return "ok"
|
|
except Exception:
|
|
return "error"
|
|
|
|
|
|
def frontend_assets_available() -> bool:
|
|
return (_static_dir() / "index.html").exists()
|
|
|
|
|
|
def production_warnings() -> list[str]:
|
|
warnings: list[str] = []
|
|
if settings.environment == "production":
|
|
proxy_operator_ready = bool(
|
|
settings.operator_proxy_auth_enabled
|
|
and settings.operator_proxy_trusted_networks
|
|
and (settings.operator_proxy_users or settings.operator_proxy_groups)
|
|
)
|
|
if not settings.operator_auth_required:
|
|
warnings.append("APPOPS_REQUIRE_OPERATOR_AUTH is disabled; private LAN operator routes are unlocked.")
|
|
elif not proxy_operator_ready and (not settings.operator_token or settings.operator_token in {"change-me", "changeme", "dev", "test"}):
|
|
warnings.append("APPOPS_OPERATOR_TOKEN is missing or unsafe; protected operator routes fail closed.")
|
|
if settings.operator_proxy_auth_enabled and not proxy_operator_ready:
|
|
warnings.append("Account-bound operator auth is enabled but trusted proxy networks and an allowed user or group are not fully configured.")
|
|
if not settings.voice_token or settings.voice_token in {"change-me", "changeme", "dev", "test"}:
|
|
warnings.append("APP_OPS_VOICE_TOKEN is missing or unsafe for production.")
|
|
warnings.extend(secret_storage_status()["warnings"])
|
|
return warnings
|
|
|
|
|
|
def connector_config_summary(session: Session) -> dict[str, Any]:
|
|
return {
|
|
item["connectorId"]: {
|
|
"configured": item["configured"],
|
|
"hasSecret": item["hasSecret"],
|
|
"missing": item["missing"],
|
|
}
|
|
for item in list_connector_settings(session)
|
|
}
|
|
|
|
|
|
def runtime_info(session: Session) -> dict[str, Any]:
|
|
setup_status = get_setup_status(session)
|
|
storage_status = secret_storage_status()
|
|
notification_rules = session.exec(select(NotificationRule)).all()
|
|
notification_channels = session.exec(select(NotificationChannel)).all()
|
|
source_metadata = Path("/etc/appops/source-metadata")
|
|
source_lines = source_metadata.read_text(encoding="utf-8").splitlines() if source_metadata.is_file() else []
|
|
source_repository, source_revision, source_tree = source_lines if len(source_lines) == 3 else ("unknown", "unknown", "unknown")
|
|
return {
|
|
"version": APP_VERSION,
|
|
"sourceRepository": source_repository,
|
|
"sourceRevision": source_revision,
|
|
"sourceTree": source_tree,
|
|
"environment": settings.environment,
|
|
"dataDirConfigured": bool(settings.data_dir),
|
|
"dataDir": str(_data_dir()),
|
|
"dbPath": settings.db_path,
|
|
"dbStatus": db_status(session),
|
|
"frontendMode": "static" if frontend_assets_available() else "api_only",
|
|
"frontendAssetsAvailable": frontend_assets_available(),
|
|
"setupComplete": setup_status["setupComplete"],
|
|
"liveReady": setup_status["liveReady"],
|
|
"firstRunCompleted": setup_status["firstRunCompleted"],
|
|
"coverageComplete": setup_status["coverageComplete"],
|
|
"missingOptionalSources": setup_status["missingOptionalSources"],
|
|
"canonicalPort": 1216,
|
|
"canonicalInstance": settings.port == 1216,
|
|
"secretStorageStatus": storage_status,
|
|
"notificationRulesStatus": {
|
|
"rules": len(notification_rules),
|
|
"channels": len(notification_channels),
|
|
"homeAssistantConfigured": any(channel.id == "home_assistant" and channel.enabled for channel in notification_channels),
|
|
},
|
|
"setupStatus": setup_status,
|
|
"connectorConfigSummary": connector_config_summary(session),
|
|
"secretStorage": storage_status,
|
|
"productionWarnings": production_warnings(),
|
|
}
|
|
|
|
|
|
def deep_health(session: Session) -> dict[str, Any]:
|
|
checks = {
|
|
"dbReachable": db_status(session) == "ok",
|
|
"dataDirWritable": is_data_dir_writable(),
|
|
"frontendAssetsAvailable": frontend_assets_available(),
|
|
"secretStorageReady": secret_storage_status()["status"] in {"ready"},
|
|
}
|
|
return {
|
|
"status": "ok" if all(checks.values()) or (checks["dbReachable"] and checks["dataDirWritable"]) else "degraded",
|
|
"checks": checks,
|
|
"setup": get_setup_status(session),
|
|
"connectors": connector_config_summary(session),
|
|
"secretStorage": secret_storage_status(),
|
|
"warnings": production_warnings(),
|
|
}
|
|
|
|
|
|
def _backup_metadata(path: Path) -> dict[str, Any]:
|
|
stat = path.stat()
|
|
created = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat().replace("+00:00", "Z")
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
manifest_path = path.with_suffix('.manifest.json')
|
|
return {
|
|
"name": path.name,
|
|
"path": str(path),
|
|
"sizeBytes": stat.st_size,
|
|
"createdAt": created,
|
|
"sha256": digest.hexdigest(),
|
|
"manifestAvailable": manifest_path.is_file(),
|
|
}
|
|
|
|
|
|
def _backup_manifest_path(path: Path) -> Path:
|
|
return path.with_suffix('.manifest.json')
|
|
|
|
|
|
def list_sqlite_backups(limit: int = 20) -> dict[str, Any]:
|
|
backup_dir = _backup_dir()
|
|
backups = []
|
|
if backup_dir.exists():
|
|
backups = sorted(
|
|
(item for item in backup_dir.glob("appops-db-*.db") if item.is_file()),
|
|
key=lambda item: item.stat().st_mtime,
|
|
reverse=True,
|
|
)[:limit]
|
|
return {
|
|
"dbPath": settings.db_path,
|
|
"backupDir": str(backup_dir),
|
|
"backups": [_backup_metadata(item) for item in backups],
|
|
"restoreMode": "manual_stop_container",
|
|
}
|
|
|
|
|
|
def sqlite_maintenance_status() -> dict[str, Any]:
|
|
db_path = Path(settings.db_path)
|
|
wal_path = Path(f"{settings.db_path}-wal")
|
|
result: dict[str, Any] = {
|
|
"status": "missing" if not db_path.is_file() else "ready",
|
|
"databaseBytes": db_path.stat().st_size if db_path.is_file() else 0,
|
|
"walBytes": wal_path.stat().st_size if wal_path.is_file() else 0,
|
|
"journalMode": "unknown",
|
|
"pageCount": 0,
|
|
"freePages": 0,
|
|
"reclaimableBytes": 0,
|
|
"recommendation": "none",
|
|
}
|
|
if not db_path.is_file():
|
|
return result
|
|
try:
|
|
connection = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5)
|
|
try:
|
|
page_size = int(connection.execute("PRAGMA page_size").fetchone()[0])
|
|
result["pageCount"] = int(connection.execute("PRAGMA page_count").fetchone()[0])
|
|
result["freePages"] = int(connection.execute("PRAGMA freelist_count").fetchone()[0])
|
|
result["journalMode"] = str(connection.execute("PRAGMA journal_mode").fetchone()[0])
|
|
result["reclaimableBytes"] = result["freePages"] * page_size
|
|
finally:
|
|
connection.close()
|
|
except (OSError, sqlite3.Error, TypeError, ValueError):
|
|
result["status"] = "error"
|
|
return result
|
|
|
|
if result["walBytes"] >= 16 * 1024 * 1024:
|
|
result["recommendation"] = "checkpoint"
|
|
elif result["reclaimableBytes"] >= max(32 * 1024 * 1024, result["databaseBytes"] // 4):
|
|
result["recommendation"] = "offline_vacuum_after_backup"
|
|
return result
|
|
|
|
|
|
def checkpoint_sqlite(session: Session) -> dict[str, Any]:
|
|
before = sqlite_maintenance_status()
|
|
backup = create_sqlite_backup(session)
|
|
if not backup.get("success"):
|
|
return {"success": False, "status": "backup_failed", "before": before, "backup": backup}
|
|
try:
|
|
connection = sqlite3.connect(settings.db_path, timeout=30)
|
|
try:
|
|
checkpoint = connection.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone()
|
|
finally:
|
|
connection.close()
|
|
after = sqlite_maintenance_status()
|
|
success = bool(checkpoint) and int(checkpoint[0]) == 0
|
|
except (OSError, sqlite3.Error, TypeError, ValueError):
|
|
checkpoint = None
|
|
after = sqlite_maintenance_status()
|
|
success = False
|
|
audit = log_event(
|
|
session,
|
|
source="manual",
|
|
identity="operator",
|
|
action_id="runtime.sqlite.checkpoint",
|
|
target="appops.db",
|
|
target_type="runtime_database",
|
|
risk="low",
|
|
policy_decision="allowed",
|
|
reason="Operator requested a backup-first SQLite WAL checkpoint.",
|
|
result="success" if success else "failed",
|
|
redacted_metadata={"beforeWalBytes": before["walBytes"], "afterWalBytes": after["walBytes"]},
|
|
)
|
|
return {
|
|
"success": success,
|
|
"status": "checkpointed" if success else "failed",
|
|
"before": before,
|
|
"after": after,
|
|
"backup": backup.get("backup"),
|
|
"auditId": audit.id,
|
|
}
|
|
|
|
|
|
def _sqlite_backup_path(name: str) -> Path | None:
|
|
normalized_name = Path(str(name)).name
|
|
if normalized_name != name or not normalized_name.startswith('appops-db-') or not normalized_name.endswith('.db'):
|
|
return None
|
|
backup_dir = _backup_dir().resolve()
|
|
candidate = (backup_dir / normalized_name).resolve()
|
|
if candidate.parent != backup_dir or not candidate.is_file():
|
|
return None
|
|
return candidate
|
|
|
|
|
|
def get_sqlite_backup_path(name: str) -> Path | None:
|
|
return _sqlite_backup_path(name)
|
|
|
|
|
|
def verify_sqlite_backup(
|
|
session: Session,
|
|
name: str,
|
|
*,
|
|
source: str = 'manual',
|
|
identity: str = 'operator',
|
|
reason: str = 'Existing SQLite backup integrity verification requested from Settings.',
|
|
) -> dict[str, Any]:
|
|
path = _sqlite_backup_path(name)
|
|
if not path:
|
|
return {
|
|
'success': False,
|
|
'status': 'not_found',
|
|
'message': 'SQLite backup not found.',
|
|
'restoreMode': 'manual_stop_container',
|
|
}
|
|
integrity_check = 'unknown'
|
|
restore_check = 'unknown'
|
|
manifest_check = 'legacy_missing'
|
|
try:
|
|
metadata = _backup_metadata(path)
|
|
manifest_path = _backup_manifest_path(path)
|
|
if manifest_path.is_file():
|
|
manifest = json.loads(manifest_path.read_text(encoding='utf-8'))
|
|
manifest_check = 'ok' if (
|
|
manifest.get('databaseFile') == path.name
|
|
and manifest.get('sha256') == metadata['sha256']
|
|
and manifest.get('sizeBytes') == metadata['sizeBytes']
|
|
) else 'mismatch'
|
|
connection = sqlite3.connect(f'file:{path}?mode=ro', uri=True)
|
|
try:
|
|
result = connection.execute('PRAGMA integrity_check').fetchone()
|
|
finally:
|
|
connection.close()
|
|
integrity_check = str(result[0]) if result else 'unknown'
|
|
with tempfile.TemporaryDirectory(prefix="appops-restore-check-") as temp_dir:
|
|
restored = Path(temp_dir) / "restored.db"
|
|
shutil.copy2(path, restored)
|
|
restored_connection = sqlite3.connect(restored)
|
|
try:
|
|
restored_integrity = restored_connection.execute("PRAGMA integrity_check").fetchone()
|
|
restored_connection.execute("SELECT name FROM sqlite_master LIMIT 1").fetchone()
|
|
finally:
|
|
restored_connection.close()
|
|
restore_check = str(restored_integrity[0]) if restored_integrity else 'unknown'
|
|
success = integrity_check == 'ok' and restore_check == 'ok' and manifest_check != 'mismatch'
|
|
except (OSError, ValueError, json.JSONDecodeError, sqlite3.Error):
|
|
success = False
|
|
|
|
audit = log_event(
|
|
session,
|
|
source=source,
|
|
identity=identity,
|
|
action_id='runtime.sqlite_backup.verify',
|
|
target=path.name,
|
|
target_type='runtime_backup',
|
|
risk='low',
|
|
policy_decision='allowed',
|
|
reason=reason,
|
|
result='success' if success else 'failed',
|
|
execution_plan={'steps': ['open backup read-only', 'PRAGMA integrity_check', 'write audit']},
|
|
post_checks={'integrityCheck': integrity_check, 'restoreCheck': restore_check, 'manifestCheck': manifest_check, 'backupUnchanged': True},
|
|
redacted_metadata={'backupName': path.name, 'sizeBytes': path.stat().st_size},
|
|
)
|
|
return {
|
|
'success': success,
|
|
'status': 'verified' if success else 'failed',
|
|
'message': 'SQLite backup integrity is valid.' if success else 'SQLite backup integrity check failed.',
|
|
'backup': _backup_metadata(path),
|
|
'integrityCheck': integrity_check,
|
|
'restoreCheck': restore_check,
|
|
'manifestCheck': manifest_check,
|
|
'auditId': audit.id,
|
|
'restoreMode': 'manual_stop_container',
|
|
}
|
|
|
|
|
|
def verify_latest_backup_drill(session: Session) -> dict[str, Any]:
|
|
backups = list_sqlite_backups(limit=1).get('backups', [])
|
|
if not backups:
|
|
return {'success': False, 'status': 'no_backup', 'restoreMode': 'manual_stop_container'}
|
|
return verify_sqlite_backup(
|
|
session,
|
|
str(backups[0]['name']),
|
|
source='system',
|
|
identity='restore-drill',
|
|
reason='Scheduled read-only backup integrity and disposable restore-copy drill.',
|
|
)
|
|
|
|
|
|
def create_sqlite_backup(session: Session) -> dict[str, Any]:
|
|
source = Path(settings.db_path)
|
|
destination_dir = _backup_dir()
|
|
timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
|
|
destination = destination_dir / f"appops-db-{timestamp}.db"
|
|
|
|
if not _BACKUP_LOCK.acquire(blocking=False):
|
|
return {"success": False, "status": "busy", "message": "A SQLite backup is already running.", "restoreMode": "manual_stop_container"}
|
|
try:
|
|
if not source.exists():
|
|
raise FileNotFoundError("configured SQLite database does not exist")
|
|
destination_dir.mkdir(parents=True, exist_ok=True)
|
|
free_bytes = shutil.disk_usage(destination_dir).free
|
|
required_bytes = max(source.stat().st_size * 2, 16 * 1024 * 1024)
|
|
if free_bytes < required_bytes:
|
|
raise OSError("insufficient_space_for_backup")
|
|
with sqlite3.connect(source) as source_conn, sqlite3.connect(destination) as destination_conn:
|
|
source_conn.backup(destination_conn)
|
|
integrity = destination_conn.execute("PRAGMA integrity_check").fetchone()
|
|
integrity_check = str(integrity[0]) if integrity else "unknown"
|
|
success = integrity_check == "ok"
|
|
metadata = _backup_metadata(destination)
|
|
manifest = {
|
|
"schemaVersion": 1,
|
|
"appVersion": APP_VERSION,
|
|
"databaseFile": destination.name,
|
|
"createdAt": metadata["createdAt"],
|
|
"sizeBytes": metadata["sizeBytes"],
|
|
"sha256": metadata["sha256"],
|
|
"integrityCheck": integrity_check,
|
|
"restoreMode": "manual_stop_container",
|
|
"secretKeySource": settings.secret_key_source,
|
|
}
|
|
_backup_manifest_path(destination).write_text(
|
|
json.dumps(manifest, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
metadata["manifestAvailable"] = True
|
|
retained = sorted(
|
|
(item for item in destination_dir.glob("appops-db-*.db") if item.is_file()),
|
|
key=lambda item: item.stat().st_mtime,
|
|
reverse=True,
|
|
)
|
|
for stale in retained[settings.backup_retention:]:
|
|
stale.unlink()
|
|
_backup_manifest_path(stale).unlink(missing_ok=True)
|
|
audit = log_event(
|
|
session,
|
|
source="manual",
|
|
identity="operator",
|
|
action_id="runtime.sqlite_backup",
|
|
target="appops-db",
|
|
target_type="runtime",
|
|
risk="low",
|
|
policy_decision="allowed",
|
|
reason="SQLite online backup requested from Settings.",
|
|
result="success" if success else "failed",
|
|
execution_plan={"steps": ["sqlite backup api", "integrity_check", "write audit"]},
|
|
post_checks={"integrityCheck": integrity_check, "backupCreated": destination.exists()},
|
|
redacted_metadata={"backupPath": str(destination), "sizeBytes": metadata["sizeBytes"]},
|
|
)
|
|
return {
|
|
"success": success,
|
|
"status": "created" if success else "failed",
|
|
"message": "SQLite backup created." if success else "SQLite backup integrity check failed.",
|
|
"backup": metadata,
|
|
"integrityCheck": integrity_check,
|
|
"auditId": audit.id,
|
|
"restoreMode": "manual_stop_container",
|
|
}
|
|
except (OSError, sqlite3.Error) as exc:
|
|
audit = log_event(
|
|
session,
|
|
source="manual",
|
|
identity="operator",
|
|
action_id="runtime.sqlite_backup",
|
|
target="appops-db",
|
|
target_type="runtime",
|
|
risk="low",
|
|
policy_decision="allowed",
|
|
reason="SQLite online backup requested from Settings.",
|
|
result="failed",
|
|
execution_plan={"steps": ["sqlite backup api", "integrity_check", "write audit"]},
|
|
post_checks={"backupCreated": False},
|
|
redacted_metadata={"errorType": type(exc).__name__},
|
|
)
|
|
return {
|
|
"success": False,
|
|
"status": "failed",
|
|
"message": "SQLite backup failed.",
|
|
"reason": type(exc).__name__,
|
|
"auditId": audit.id,
|
|
"restoreMode": "manual_stop_container",
|
|
}
|
|
finally:
|
|
_BACKUP_LOCK.release()
|
|
|
|
|
|
def export_config(session: Session) -> dict[str, Any]:
|
|
return {
|
|
"runtime": {
|
|
"environment": settings.environment,
|
|
"dataDir": str(_data_dir()),
|
|
"dbPath": settings.db_path,
|
|
"frontendOrigin": settings.frontend_origin,
|
|
"secretStorage": secret_storage_status(),
|
|
},
|
|
"connectors": list_connector_settings(session),
|
|
"setup": get_setup_status(session),
|
|
"note": "Secret values are intentionally excluded.",
|
|
}
|
|
|
|
|
|
def export_policies(session: Session) -> dict[str, Any]:
|
|
policies = session.exec(select(Policy).order_by(Policy.id)).all()
|
|
return {
|
|
"policies": [
|
|
{
|
|
"id": policy.id,
|
|
"action": policy.action,
|
|
"target": policy.target,
|
|
"voice": policy.voice,
|
|
"manual": policy.manual,
|
|
"risk": policy.risk,
|
|
"reason": policy.reason,
|
|
}
|
|
for policy in policies
|
|
]
|
|
}
|
|
|
|
|
|
def export_inventory(session: Session) -> dict[str, Any]:
|
|
discovered = session.exec(select(DiscoveredContainer).order_by(DiscoveredContainer.name)).all()
|
|
mappings = session.exec(select(AppContainerMapping).order_by(AppContainerMapping.container_name)).all()
|
|
return {
|
|
"discovered": [
|
|
{
|
|
"id": item.id,
|
|
"containerId": item.container_id,
|
|
"name": item.name,
|
|
"image": item.image,
|
|
"state": item.state,
|
|
"status": item.status,
|
|
"suggestedAppType": item.suggested_app_type,
|
|
"suggestedCategory": item.suggested_category,
|
|
"suggestedCriticality": item.suggested_criticality,
|
|
"suggestedConnector": item.suggested_connector,
|
|
"importStatus": item.import_status,
|
|
"matchedAppId": item.matched_app_id,
|
|
}
|
|
for item in discovered
|
|
],
|
|
"mappings": [
|
|
{
|
|
"id": mapping.id,
|
|
"appId": mapping.app_id,
|
|
"containerId": mapping.container_id,
|
|
"containerName": mapping.container_name,
|
|
"role": mapping.role,
|
|
"source": mapping.source,
|
|
"isPrimary": mapping.is_primary,
|
|
}
|
|
for mapping in mappings
|
|
],
|
|
}
|
|
|
|
|
|
def export_audit_summary(session: Session) -> dict[str, Any]:
|
|
events = session.exec(select(AuditEvent).order_by(AuditEvent.timestamp.desc())).all()
|
|
by_result: dict[str, int] = {}
|
|
by_action: dict[str, int] = {}
|
|
for event in events:
|
|
by_result[event.result] = by_result.get(event.result, 0) + 1
|
|
by_action[event.action_id] = by_action.get(event.action_id, 0) + 1
|
|
return {
|
|
"total": len(events),
|
|
"byResult": by_result,
|
|
"byAction": by_action,
|
|
"recent": [
|
|
{
|
|
"id": event.id,
|
|
"timestamp": event.timestamp.isoformat(),
|
|
"source": event.source,
|
|
"action": event.action_id,
|
|
"target": event.target,
|
|
"risk": event.risk,
|
|
"policyDecision": event.policy_decision,
|
|
"result": event.result,
|
|
}
|
|
for event in events[:25]
|
|
],
|
|
}
|