virtuality/web/core.py

"""Shared primitives of the Virtuality panel: settings, auth, command runner,
background operations store, templates. Feature modules import from here;
app.py re-exports the same names for backwards compatibility."""
import json
import os
import re
import subprocess
import threading
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Callable
from urllib.parse import quote, urlsplit

from fastapi import Request
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from itsdangerous import BadSignature, URLSafeSerializer, URLSafeTimedSerializer

import auth
import presenters
import update_core

BASE_DIR = Path(__file__).resolve().parent
APP_NAME = "Virtuality"
ENV_FILE = BASE_DIR / ".env"
# virsh, ip and qemu-img print translated text under a Russian locale; the panel parses their output.
os.environ["LC_ALL"] = "C.UTF-8"
os.environ["LANG"] = "C.UTF-8"

STORAGE_DIR = Path("/var/lib/virtuality")
ISO_DIR = Path("/var/lib/virtuality/iso")
IMAGES_DIR = Path("/var/lib/virtuality/images")
DISK_IMAGES_DIR = Path("/var/lib/virtuality/disk-images")
BACKUPS_DIR = Path("/var/lib/virtuality/backups")
CONFIG_DIR = Path("/var/lib/virtuality/config")
OPERATIONS_DIR = Path("/var/log/virtuality/operations")
DEFAULT_BRIDGE = "br0"
IMAGES_POOL = "virtuality-images"
ISO_POOL = "virtuality-iso"

OP_LOCK = threading.Lock()


def load_env() -> dict[str, str]:
    data: dict[str, str] = {}
    if ENV_FILE.exists():
        for raw_line in ENV_FILE.read_text().splitlines():
            line = raw_line.strip()
            if line and not line.startswith("#") and "=" in line:
                key, value = line.split("=", 1)
                data[key.strip()] = value.strip().strip('"').strip("'")
    return data


CONFIG = load_env()
AUTH_USER = CONFIG.get("VIRTUALITY_AUTH_USER", os.environ.get("VIRTUALITY_AUTH_USER", "root"))
SESSION_SECRET = CONFIG.get("VIRTUALITY_SESSION_SECRET", os.environ.get("VIRTUALITY_SESSION_SECRET", "dev-secret-change-me"))
COOKIE_SECURE = CONFIG.get("VIRTUALITY_COOKIE_SECURE", os.environ.get("VIRTUALITY_COOKIE_SECURE", "0")) == "1"
SESSION_MAX_AGE = 60 * 60 * 12
login_throttle = auth.LoginThrottle()
serializer = URLSafeTimedSerializer(SESSION_SECRET, salt="virtuality-session")
console_serializer = URLSafeSerializer(SESSION_SECRET, salt="virtuality-console")


def is_configured() -> bool:
    return bool(AUTH_USER and SESSION_SECRET != "dev-secret-change-me")


def read_version() -> str:
    for path in (BASE_DIR / "VERSION", BASE_DIR.parent / "VERSION"):
        try:
            return path.read_text().strip() or "unknown"
        except OSError:
            continue
    return "unknown"


APP_VERSION = read_version()


# ---------------------------------------------------------------- templates
def update_notice() -> dict[str, Any]:
    try:
        data = json.loads((update_core.STATE_DIR / "last_check.json").read_text())
    except Exception:
        return {"has_update": False}
    return {"has_update": bool(data.get("has_update")), "latest_version": data.get("latest_version", "")}


templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
templates.env.globals.update(
    update_notice=update_notice,
    vm_presets=presenters.VM_PRESETS,
    port_presets=presenters.PORT_PRESETS,
    plural=presenters.plural,
    app_version=APP_VERSION,
    app_name=APP_NAME,
)
templates.env.filters.update(
    vm_state=presenters.vm_state,
    operation_state=presenters.operation_state,
    service_state=presenters.service_state,
    format_mb=presenters.format_mb,
    format_bytes=presenters.format_bytes,
    level_tone=presenters.level_tone,
    port_service=presenters.port_service,
)


def render(request: Request, template: str, context: dict[str, Any] | None = None, status_code: int = 200, **extra: Any):
    data = {"request": request, "app_name": APP_NAME, "user": AUTH_USER}
    data.update(context or {})
    data.update(extra)
    return templates.TemplateResponse(request, template, data, status_code=status_code)


# ---------------------------------------------------------------- auth
def verify_linux_password(username: str, password: str) -> bool:
    if username != AUTH_USER:
        return False
    return auth.verify_password(username, password)


def user_from_session_token(token: str | None) -> str | None:
    if not token:
        return None
    try:
        data = serializer.loads(token, max_age=SESSION_MAX_AGE)
    except BadSignature:
        return None
    if not isinstance(data, dict):
        return None
    return AUTH_USER if data.get("user") == AUTH_USER else None


def get_current_user(request: Request) -> str | None:
    return user_from_session_token(request.cookies.get("virtuality_session"))


def require_auth(request: Request):
    """Return a redirect to /login for anonymous requests, otherwise None."""
    if not get_current_user(request):
        return RedirectResponse(url="/login", status_code=303)
    return None


def require_api_auth(request: Request):
    """JSON variant of require_auth for /api and /live endpoints."""
    if not get_current_user(request):
        return JSONResponse({"ok": False, "error": "Unauthorized"}, status_code=401)
    return None


def client_key(request: Request) -> str:
    return request.client.host if request.client else "unknown"


def same_origin(request: Request) -> bool:
    origin = request.headers.get("origin") or request.headers.get("referer")
    if not origin or origin == "null":
        # Browsers always send Origin on cross-site form posts; tools like curl do not.
        return origin != "null"
    origin_host = urlsplit(origin).netloc.lower()
    allowed = {request.headers.get("host", "").lower()}
    forwarded = request.headers.get("x-forwarded-host")
    if forwarded:
        allowed.update(item.strip().lower() for item in forwarded.split(","))
    return origin_host in allowed


def redirect_with_message(path: str, key: str, message: str) -> RedirectResponse:
    return RedirectResponse(url=f"{path}?{key}={quote(message)}", status_code=303)


# ---------------------------------------------------------------- commands
def run_cmd(cmd: list[str], timeout: int = 12) -> dict[str, Any]:
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False)
        return {"ok": result.returncode == 0, "code": result.returncode, "stdout": result.stdout.strip(), "stderr": result.stderr.strip(), "cmd": " ".join(cmd)}
    except Exception as exc:
        return {"ok": False, "code": -1, "stdout": "", "stderr": str(exc), "cmd": " ".join(cmd)}


def cmd_error(result: dict[str, Any], fallback: str) -> str:
    """Best human-readable error text from a run_cmd result."""
    text = (result.get("stderr") or result.get("stdout") or "").strip()
    text = re.sub(r"^error:\s*", "", text, flags=re.IGNORECASE)
    return text.splitlines()[0] if text else fallback


def utc_now() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def tail_text(path: Path, max_lines: int = 220) -> str:
    if not path.exists():
        return ""
    try:
        return "\n".join(path.read_text(errors="replace").splitlines()[-max_lines:])
    except OSError:
        return ""


def valid_vm_name(name: str) -> bool:
    """Latin letters, digits, dot, dash, underscore; never a bare number or a UUID — virsh would read those as an ID/UUID of another machine."""
    if not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{1,62}", name or ""):
        return False
    return not name.isdigit() and not re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", name)


def valid_label(name: str) -> bool:
    """Snapshot, backup and similar user-chosen names."""
    return bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,62}", name or ""))


def vm_exists(name: str) -> bool:
    return run_cmd(["virsh", "dominfo", name], timeout=8)["ok"]


def vm_state(name: str) -> str:
    result = run_cmd(["virsh", "domstate", name], timeout=8)
    return (result.get("stdout") or "unknown").strip().lower()


def vm_is_running(name: str) -> bool:
    return vm_state(name) in ("running", "idle", "blocked", "paused", "pmsuspended", "in shutdown")


# ---------------------------------------------------------------- operations
def operation_meta_path(operation_id: str) -> Path:
    return OPERATIONS_DIR / f"{operation_id}.json"


def operation_log_path(operation_id: str) -> Path:
    return OPERATIONS_DIR / f"{operation_id}.log"


def ensure_operations_dir() -> None:
    OPERATIONS_DIR.mkdir(parents=True, exist_ok=True)


def write_operation(operation: dict[str, Any]) -> None:
    ensure_operations_dir()
    path = operation_meta_path(operation["id"])
    tmp_path = path.with_suffix(".json.tmp")
    with OP_LOCK:
        tmp_path.write_text(json.dumps(operation, ensure_ascii=False, indent=2))
        tmp_path.replace(path)


def read_operation(operation_id: str) -> dict[str, Any] | None:
    if not re.fullmatch(r"[a-f0-9-]{36}", operation_id or ""):
        return None
    path = operation_meta_path(operation_id)
    if not path.exists():
        return None
    try:
        data = json.loads(path.read_text())
    except Exception:
        return None
    data["log_tail"] = tail_text(operation_log_path(operation_id))
    return data


def list_operations(limit: int = 25) -> list[dict[str, Any]]:
    ensure_operations_dir()
    operations: list[dict[str, Any]] = []
    for path in sorted(OPERATIONS_DIR.glob("*.json"), key=lambda item: item.stat().st_mtime, reverse=True):
        try:
            data = json.loads(path.read_text())
            data["log_tail"] = tail_text(operation_log_path(data["id"]), max_lines=20)
            operations.append(data)
        except Exception:
            continue
        if len(operations) >= limit:
            break
    return operations


def process_owner(pid: int | None = None) -> dict[str, Any] | None:
    """Who runs an operation: pid plus its start time and boot id, so a reused pid is not mistaken for it."""
    pid = pid or os.getpid()
    try:
        stat = Path(f"/proc/{pid}/stat").read_text()
        boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip()
    except OSError:
        return None
    return {"pid": pid, "start": stat.rsplit(")", 1)[1].split()[19], "boot": boot}


def owner_alive(owner: Any) -> bool:
    if not isinstance(owner, dict) or not isinstance(owner.get("pid"), int):
        return False
    return process_owner(owner["pid"]) == owner


def interrupt_orphaned_operations() -> int:
    """Operations run in threads of the panel process: after a restart the ones still marked
    queued/running will never finish. Close them so they do not block the next attempt.
    Operations of another live process (scheduled backups run from a timer) are left alone."""
    if not OPERATIONS_DIR.exists():
        return 0
    count = 0
    for path in OPERATIONS_DIR.glob("*.json"):
        try:
            data = json.loads(path.read_text())
        except (OSError, ValueError):
            continue
        if data.get("status") not in ("queued", "running"):
            continue
        owner = data.get("owner")
        if owner_alive(owner) and owner["pid"] != os.getpid():
            continue
        data.pop("log_tail", None)
        message = "Прервано: панель перезапускалась во время операции. Проверьте состояние машины и запустите действие снова."
        update_operation(data, status="error", progress=100, message=message, finished_at=utc_now(), interrupted=True)
        append_operation_log(data["id"], message)
        count += 1
    return count


def active_operations_for(vm_name: str, kinds: tuple[str, ...] | set[str] | None = None) -> list[dict[str, Any]]:
    """Unfinished operations of one machine, over the whole operations directory (not only the latest N)."""
    ensure_operations_dir()
    found = []
    for path in OPERATIONS_DIR.glob("*.json"):
        try:
            data = json.loads(path.read_text())
        except (OSError, ValueError):
            continue
        if data.get("vm_name") == vm_name and data.get("status") in ("queued", "running") and (kinds is None or data.get("type") in kinds):
            found.append(data)
    return found


def running_operations(kind: str | None = None) -> list[dict[str, Any]]:
    return [op for op in list_operations(50) if op.get("status") in ("queued", "running") and (kind is None or op.get("type") == kind)]


def append_operation_log(operation_id: str, message: str) -> None:
    ensure_operations_dir()
    with operation_log_path(operation_id).open("a", encoding="utf-8") as handle:
        handle.write(f"[{utc_now()}] {message.rstrip()}\n")


def operation_css(status: str) -> str:
    return "ok" if status == "success" else "err" if status == "error" else "warn"


def update_operation(operation: dict[str, Any], **changes: Any) -> None:
    operation.update(changes)
    operation["updated_at"] = utc_now()
    write_operation(operation)


def new_operation(kind: str, title: str, message: str = "Операция поставлена в очередь", **meta: Any) -> dict[str, Any]:
    operation = {"id": str(uuid.uuid4()), "type": kind, "title": title, "status": "queued", "progress": 0, "message": message, "created_at": utc_now(), "updated_at": utc_now(), "created_by": AUTH_USER, "owner": process_owner()}
    operation.update(meta)
    write_operation(operation)
    append_operation_log(operation["id"], "Операция поставлена в очередь.")
    return operation


def finish_operation(operation: dict[str, Any], ok: bool, message: str, **meta: Any) -> None:
    fresh = read_operation(operation["id"]) or operation
    fresh.pop("log_tail", None)
    update_operation(fresh, status="success" if ok else "error", progress=100, message=message, finished_at=utc_now(), **meta)
    append_operation_log(operation["id"], message)


def run_operation(operation: dict[str, Any], worker: Callable[[dict[str, Any]], None]) -> None:
    """Run worker(operation) in a daemon thread; the worker reports progress via
    update_operation/append_operation_log and must call finish_operation."""

    def runner() -> None:
        try:
            update_operation(operation, status="running", started_at=utc_now())
            worker(operation)
        except Exception as exc:  # noqa: BLE001 — surface any failure to the user
            append_operation_log(operation["id"], f"Ошибка: {exc}")
            finish_operation(operation, False, str(exc)[:240])

    threading.Thread(target=runner, daemon=True).start()