squid-manager/log_rotate.py

"""log_rotate.py - Squid log rotation manager.

Provides:
- get_log_file_info(path) -> dict with size/mtime/age/writability
- generate_logrotate_config(log_paths, ...) -> str (full /etc/logrotate.d/squid content)
- install_logrotate(config_content) -> (ok, msg)
- uninstall_logrotate() -> (ok, msg)
- trigger_squid_rotate(binary='squid') -> (ok, msg)
- is_logrotate_available() -> bool

All functions use only Python stdlib (os, shutil, subprocess, datetime, time).
"""
from __future__ import annotations

import os
import shutil
import subprocess
import time
from datetime import datetime, timezone

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

LOGROTATE_DIR = "/etc/logrotate.d"
LOGROTATE_PATH = os.path.join(LOGROTATE_DIR, "squid")
LOGROTATE_BIN_CANDIDATES = ("/usr/sbin/logrotate", "/usr/bin/logrotate", "logrotate")

# Header that is recognised by `logrotate -d` as managed-by-squid-manager.
MANAGED_HEADER = "# Managed by Squid Web Manager - do not edit by hand."


# ---------------------------------------------------------------------------
# File info
# ---------------------------------------------------------------------------

def _human_size(num_bytes: int) -> str:
    """Format a byte count as a human-readable string (KB/MB/GB)."""
    try:
        n = float(num_bytes)
    except (TypeError, ValueError):
        return "0 B"
    if n < 0:
        return "0 B"
    units = ("B", "KB", "MB", "GB", "TB", "PB")
    idx = 0
    while n >= 1024.0 and idx < len(units) - 1:
        n /= 1024.0
        idx += 1
    if idx == 0:
        return f"{int(n)} {units[idx]}"
    return f"{n:.2f} {units[idx]}"


def get_log_file_info(path: str) -> dict:
    """Return metadata for a single log file.

    Keys: exists, size_bytes, size_human, mtime, mtime_human,
          age_days, is_writable, path, error.

    Never raises - missing files / permission errors return a stub.
    """
    info = {
        "path": path,
        "exists": False,
        "size_bytes": 0,
        "size_human": "0 B",
        "mtime": None,
        "mtime_human": "",
        "age_days": None,
        "is_writable": False,
        "error": "",
    }
    if not path:
        info["error"] = "empty path"
        return info
    try:
        st = os.stat(path)
    except FileNotFoundError:
        info["error"] = "not found"
        return info
    except OSError as e:
        info["error"] = f"{type(e).__name__}: {e}"
        return info
    info["exists"] = True
    info["size_bytes"] = int(st.st_size)
    info["size_human"] = _human_size(st.st_size)
    info["mtime"] = st.st_mtime
    info["mtime_human"] = datetime.fromtimestamp(
        st.st_mtime, tz=timezone.utc
    ).strftime("%Y-%m-%d %H:%M:%S UTC")
    try:
        age_sec = max(0.0, time.time() - st.st_mtime)
        info["age_days"] = round(age_sec / 86400.0, 2)
    except Exception:
        info["age_days"] = None
    info["is_writable"] = os.access(path, os.W_OK)
    return info


# ---------------------------------------------------------------------------
# logrotate config generation
# ---------------------------------------------------------------------------

def _is_log_path(path: str) -> bool:
    """True if path looks like an actual log file path (not blank, not None)."""
    return bool(path) and isinstance(path, str) and path.strip() != ""


def generate_logrotate_config(
    log_paths: list[str],
    compress: bool = True,
    rotate_count: int = 7,
    rotate_size: str = "100M",
    postrotate_cmd: str = "squid -k rotate",
    prerotate_cmd: str = "",
    daily: bool = True,
    missingok: bool = True,
    notifempty: bool = True,
    create_mode: str = "0640",
    create_owner: str = "proxy",
    create_group: str = "proxy",
    sharedscripts: bool = True,
    delaycompress: bool = False,
) -> str:
    """Render a full /etc/logrotate.d/squid config block.

    Only non-empty paths from ``log_paths`` are included.
    ``rotate_count`` is the number of old logs to keep, ``rotate_size`` is the
    size threshold that triggers rotation (e.g. ``"100M"``, ``"500K"``).
    """
    paths = [p.strip() for p in (log_paths or []) if _is_log_path(p)]
    # de-dup, keep order
    seen = set()
    uniq_paths = []
    for p in paths:
        if p not in seen:
            seen.add(p)
            uniq_paths.append(p)

    lines = [MANAGED_HEADER, "{", ""]
    lines.append("    # rotate thresholds")
    lines.append(f"    size {rotate_size}")
    if daily:
        lines.append("    daily")
    if missingok:
        lines.append("    missingok")
    if notifempty:
        lines.append("    notifempty")
    lines.append(f"    rotate {int(rotate_count)}")
    if create_mode or create_owner or create_group:
        parts = []
        if create_mode:
            parts.append(create_mode)
        if create_owner:
            parts.append(create_owner)
        if create_group:
            parts.append(create_group)
        lines.append(f"    create {' '.join(parts)}".rstrip())
    if delaycompress:
        lines.append("    delaycompress")
    if compress:
        lines.append("    compress")
    if sharedscripts:
        lines.append("    sharedscripts")
    lines.append("")
    if uniq_paths:
        lines.append("    # log files managed by this block")
        for p in uniq_paths:
            lines.append(f"    {p}")
    lines.append("")
    if prerotate_cmd:
        lines.append(f"    prerotate")
        lines.append(f"        {prerotate_cmd}")
        lines.append(f"    endscript")
    if postrotate_cmd:
        lines.append(f"    postrotate")
        lines.append(f"        {postrotate_cmd}")
        lines.append(f"    endscript")
    lines.append("}")
    lines.append("")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# logrotate install / uninstall
# ---------------------------------------------------------------------------

def _try_write(path: str, content: str) -> tuple[bool, str]:
    """Attempt a direct write to ``path``. Returns (ok, msg)."""
    parent = os.path.dirname(path)
    try:
        if parent and not os.path.isdir(parent):
            os.makedirs(parent, exist_ok=True)
        # Write atomically: temp file then rename.
        tmp_path = f"{path}.squidmgr.tmp"
        with open(tmp_path, "w", encoding="utf-8") as fh:
            fh.write(content)
        os.replace(tmp_path, path)
        try:
            os.chmod(path, 0o644)
        except OSError:
            pass
        return True, f"wrote {path} ({len(content)} bytes)"
    except PermissionError as e:
        return False, f"permission denied: {e}"
    except OSError as e:
        return False, f"{type(e).__name__}: {e}"


def install_logrotate(config_content: str, path: str = LOGROTATE_PATH) -> tuple[bool, str]:
    """Write a logrotate config to /etc/logrotate.d/squid.

    Tries direct write first, then falls back to ``sudo tee`` if available.
    """
    ok, msg = _try_write(path, config_content)
    if ok:
        return True, msg
    # Fall back to sudo tee
    if shutil.which("sudo") is None:
        return False, msg
    try:
        proc = subprocess.run(
            ["sudo", "-n", "tee", path],
            input=config_content,
            text=True,
            capture_output=True,
            timeout=10,
        )
        if proc.returncode == 0:
            # sudo tee creates the file as root; chmod for readability
            try:
                subprocess.run(["sudo", "-n", "chmod", "644", path],
                               capture_output=True, timeout=5)
            except Exception:
                pass
            return True, f"wrote {path} via sudo ({len(config_content)} bytes)"
        return False, msg + f" | sudo failed: rc={proc.returncode} {proc.stderr.strip()}"
    except subprocess.TimeoutExpired:
        return False, msg + " | sudo timed out"
    except FileNotFoundError as e:
        return False, msg + f" | sudo not usable: {e}"


def uninstall_logrotate(path: str = LOGROTATE_PATH) -> tuple[bool, str]:
    """Remove the logrotate config. Returns (ok, msg)."""
    if not os.path.exists(path):
        return True, f"{path} does not exist (nothing to do)"
    try:
        os.remove(path)
        return True, f"removed {path}"
    except PermissionError:
        pass
    except OSError as e:
        return False, f"{type(e).__name__}: {e}"
    # Fallback: sudo rm
    if shutil.which("sudo") is None:
        return False, f"permission denied removing {path} (sudo unavailable)"
    try:
        proc = subprocess.run(
            ["sudo", "-n", "rm", "-f", path],
            capture_output=True, text=True, timeout=10,
        )
        if proc.returncode == 0:
            return True, f"removed {path} via sudo"
        return False, f"sudo rm failed: rc={proc.returncode} {proc.stderr.strip()}"
    except subprocess.TimeoutExpired:
        return False, "sudo rm timed out"


def read_logrotate_config(path: str = LOGROTATE_PATH) -> tuple[bool, str, str]:
    """Read the current logrotate config. Returns (ok, content, msg)."""
    if not os.path.exists(path):
        return False, "", f"{path} not found"
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            data = fh.read()
        return True, data, f"read {len(data)} bytes from {path}"
    except PermissionError as e:
        return False, "", f"permission denied: {e}"
    except OSError as e:
        return False, "", f"{type(e).__name__}: {e}"


# ---------------------------------------------------------------------------
# logrotate binary detection
# ---------------------------------------------------------------------------

def is_logrotate_available() -> bool:
    """Return True if the logrotate binary is on PATH."""
    for cand in LOGROTATE_BIN_CANDIDATES:
        if shutil.which(cand):
            return True
    return False


def get_logrotate_binary() -> str | None:
    """Return the absolute path of logrotate, or None."""
    for cand in LOGROTATE_BIN_CANDIDATES:
        p = shutil.which(cand)
        if p:
            return p
    return None


# ---------------------------------------------------------------------------
# Squid rotate trigger
# ---------------------------------------------------------------------------

def trigger_squid_rotate(binary: str = "squid", timeout: int = 15) -> tuple[bool, str]:
    """Ask Squid to rotate its logs via ``squid -k rotate``.

    This sends a signal to the running squid process asking it to close and
    reopen its log files. Combined with logrotate's "create" directive this
    lets logrotate manage file rotation without disrupting Squid.

    Returns (ok, msg). msg is a one-line summary suitable for flash / audit.
    """
    binary_path = shutil.which(binary) if os.sep not in binary else binary
    if not binary_path or not os.path.exists(binary_path):
        return False, f"squid binary not found: {binary}"
    try:
        proc = subprocess.run(
            [binary_path, "-k", "rotate"],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        out = (proc.stdout or "").strip()
        err = (proc.stderr or "").strip()
        if proc.returncode == 0:
            msg = f"squid -k rotate ok"
            if out:
                msg += f" | stdout={out[:200]}"
            return True, msg
        msg = f"squid -k rotate failed rc={proc.returncode}"
        if err:
            msg += f" | stderr={err[:200]}"
        elif out:
            msg += f" | stdout={out[:200]}"
        return False, msg
    except subprocess.TimeoutExpired:
        return False, f"squid -k rotate timed out after {timeout}s"
    except FileNotFoundError as e:
        return False, f"squid not executable: {e}"
    except Exception as e:
        return False, f"{type(e).__name__}: {e}"


# ---------------------------------------------------------------------------
# Threshold helpers for UI badges
# ---------------------------------------------------------------------------

# Bytes; UI uses these to colour-size badges.
SIZE_WARN_BYTES = 500 * 1024 * 1024     # 500 MB
SIZE_CRIT_BYTES = 2 * 1024 * 1024 * 1024  # 2 GB


def size_severity(size_bytes: int) -> str:
    """Return 'normal' / 'warn' / 'crit' for badge styling."""
    if size_bytes >= SIZE_CRIT_BYTES:
        return "crit"
    if size_bytes >= SIZE_WARN_BYTES:
        return "warn"
    return "normal"