worked poc

app/maintenance.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import os
+import shutil
+from pathlib import Path
+
+
+def storage_info(data_path: str, log_path: str) -> dict:
+    data_dir = _existing_parent(data_path)
+    total, used, free = shutil.disk_usage(data_dir)
+    log_dir = os.path.dirname(log_path) or "/var/log/suricata"
+    return {
+        "path": data_dir,
+        "total_bytes": int(total),
+        "used_bytes": int(used),
+        "free_bytes": int(free),
+        "used_percent": round((used / total) * 100.0, 2) if total else 0.0,
+        "suricata_log_bytes": directory_size(log_dir, limit_files=500),
+        "containerized": _detect_container(),
+        "hostname": os.uname().nodename,
+    }
+
+
+def clear_suricata_logs(eve_path: str) -> dict:
+    log_dir = os.path.realpath(os.path.dirname(eve_path) or "/var/log/suricata")
+    allowed_names = {
+        os.path.basename(eve_path),
+        "fast.log",
+        "stats.log",
+        "suricata.log",
+    }
+    cleared: list[dict] = []
+    for name in sorted(allowed_names):
+        path = os.path.realpath(os.path.join(log_dir, name))
+        if os.path.dirname(path) != log_dir:
+            continue
+        try:
+            stat = os.stat(path)
+        except FileNotFoundError:
+            continue
+        if not os.path.isfile(path):
+            continue
+        size = int(stat.st_size)
+        with open(path, "w", encoding="utf-8"):
+            pass
+        cleared.append({"name": name, "bytes": size})
+    return {
+        "files": cleared,
+        "bytes_freed": sum(item["bytes"] for item in cleared),
+    }
+
+
+def directory_size(path: str, limit_files: int = 500) -> int:
+    total = 0
+    count = 0
+    try:
+        entries = Path(path).iterdir()
+    except OSError:
+        return 0
+    for item in entries:
+        if count >= limit_files:
+            break
+        count += 1
+        try:
+            if item.is_file():
+                total += int(item.stat().st_size)
+        except OSError:
+            continue
+    return total
+
+
+def _existing_parent(path: str) -> str:
+    candidate = os.path.abspath(os.path.dirname(path) or ".")
+    while not os.path.exists(candidate):
+        parent = os.path.dirname(candidate)
+        if parent == candidate:
+            return "/"
+        candidate = parent
+    return candidate
+
+
+def _detect_container() -> bool:
+    if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"):
+        return True
+    try:
+        text = Path("/proc/1/cgroup").read_text(encoding="utf-8", errors="replace").lower()
+    except OSError:
+        return False
+    return any(token in text for token in ("docker", "containerd", "kubepods", "libpod", "lxc"))