90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
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"))
|