server.py

#!/usr/bin/env python3
"""Low-footprint WSGI host for the dashboard and its stale-only refresh API."""
from __future__ import annotations

import fcntl
import gzip
import json
import logging
import mimetypes
import os
import subprocess
import sys
import tempfile
from datetime import datetime, timedelta, timezone
from email.utils import formatdate
from pathlib import Path
from typing import Callable, Iterable
from urllib.parse import urlsplit

MAX_AGE_SECONDS = 24 * 60 * 60
REFRESH_TIMEOUT_SECONDS = 180
BACKOFF_SECONDS = (15 * 60, 60 * 60, 6 * 60 * 60)
MAX_REQUEST_BYTES = 1024
DATASET_FILES = {"data.json", "cve-data.json"}
STATIC_FILES = {
    "index.html", "app.js", "app-utils.js", "refresh-client.js", "styles.css",
    "demo.html", "demo.js", "mschangeintelligence.png",
    "favicon-16x16.png", "favicon-32x32.png", "favicon-96x96.png", "apple-touch-icon.png",
    "cves.html", "cves.css", "cves.js",
}
SECURITY_HEADERS = [
    ("X-Content-Type-Options", "nosniff"),
    ("X-Frame-Options", "DENY"),
    ("Referrer-Policy", "no-referrer"),
    ("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self'; object-src 'none'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'"),
]
LOG = logging.getLogger("release-dashboard")


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


def utc_text(value: datetime) -> str:
    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def parse_utc(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)


def read_dataset(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def dataset_is_stale(path: Path, now: datetime, max_age_seconds: int = MAX_AGE_SECONDS) -> bool:
    try:
        generated = parse_utc(read_dataset(path)["generatedAt"])
    except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError):
        return True
    return (now.astimezone(timezone.utc) - generated).total_seconds() >= max_age_seconds


def backoff_seconds(failures: int) -> int:
    return BACKOFF_SECONDS[min(max(failures, 1), len(BACKOFF_SECONDS)) - 1]


def atomic_state(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp_path = None
    try:
        with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, prefix=".refresh-state-", delete=False) as tmp:
            json.dump(payload, tmp, separators=(",", ":"))
            tmp.write("\n")
            temp_path = Path(tmp.name)
        temp_path.chmod(0o600)
        temp_path.replace(path)
    except Exception:
        if temp_path:
            temp_path.unlink(missing_ok=True)
        raise


def read_state(path: Path) -> dict:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
        return value if isinstance(value, dict) else {}
    except (OSError, ValueError, json.JSONDecodeError):
        return {}


def json_response(start_response, status: str, payload: dict, extra_headers=()) -> Iterable[bytes]:
    raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
    headers = [
        ("Content-Type", "application/json; charset=utf-8"),
        ("Content-Length", str(len(raw))),
        ("Cache-Control", "no-store"),
        *SECURITY_HEADERS,
        *extra_headers,
    ]
    start_response(status, headers)
    return [raw]


def expected_origin(environ: dict) -> str:
    forwarded = environ.get("HTTP_X_FORWARDED_PROTO", "").split(",", 1)[0].strip()
    scheme = forwarded or environ.get("wsgi.url_scheme", "https")
    return f"{scheme}://{environ.get('HTTP_HOST', '')}"


def origin_is_same(environ: dict, allowed_origin: str | None = None) -> bool:
    supplied = environ.get("HTTP_ORIGIN", "")
    if not supplied:
        return False
    expected = allowed_origin or expected_origin(environ)
    actual_parts = urlsplit(supplied)
    expected_parts = urlsplit(expected)
    return (
        actual_parts.scheme.lower(), actual_parts.netloc.lower(), actual_parts.path
    ) == (
        expected_parts.scheme.lower(), expected_parts.netloc.lower(), ""
    )


def valid_json_request(environ: dict) -> tuple[bool, str]:
    content_type = environ.get("CONTENT_TYPE", "").split(";", 1)[0].strip().lower()
    if content_type != "application/json":
        return False, "unsupported"
    try:
        length = int(environ.get("CONTENT_LENGTH") or "0")
    except ValueError:
        return False, "invalid"
    if length < 0 or length > MAX_REQUEST_BYTES:
        return False, "oversized"
    try:
        payload = json.loads(environ["wsgi.input"].read(length) or b"{}")
    except (KeyError, TypeError, json.JSONDecodeError):
        return False, "invalid"
    return (isinstance(payload, dict), "invalid")


def _subprocess_runner(root: Path, collector_name: str) -> Callable[[Path, int, int], None]:
    def run(output_path: Path, timeout: int, lock_fd: int) -> None:
        completed = subprocess.run(
            [sys.executable, str(root / collector_name), "--output", str(output_path)],
            cwd=root,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
            pass_fds=(lock_fd,),
        )
        if completed.returncode != 0:
            detail = (completed.stderr or completed.stdout or "collector failed").strip()
            raise RuntimeError(detail[-2000:])
    return run


def subprocess_runner(root: Path) -> Callable[[Path, int, int], None]:
    return _subprocess_runner(root, "collector.py")


def cve_subprocess_runner(root: Path) -> Callable[[Path, int, int], None]:
    return _subprocess_runner(root, "cve_collector.py")


def seed_persistent_dataset(root: Path, data_dir: Path, filename: str = "data.json") -> None:
    """Seed or repair writable last-known-good data without overwriting valid data."""
    root = Path(root)
    data_dir = Path(data_dir)
    data_dir.mkdir(parents=True, exist_ok=True)
    destination = data_dir / filename
    packaged = root / filename
    packaged_payload = None
    if packaged.is_file() and destination.resolve() != packaged.resolve():
        try:
            packaged_payload = read_dataset(packaged)
        except (OSError, TypeError, ValueError, json.JSONDecodeError):
            packaged_payload = None
    raw = None
    try:
        current = read_dataset(destination)
        parse_utc(current["generatedAt"])
        if isinstance(current.get("items"), list):
            needs_cve_migration = (
                filename == "cve-data.json"
                and isinstance(packaged_payload, dict)
                and (
                    current.get("schemaVersion") != packaged_payload.get("schemaVersion")
                    or current.get("productFilter") != packaged_payload.get("productFilter")
                )
            )
            if not needs_cve_migration:
                raw = destination.read_bytes()
    except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError):
        pass

    if raw is None:
        if not packaged.is_file() or destination.resolve() == packaged.resolve():
            return
        raw = packaged.read_bytes()
        payload = json.loads(raw)
        parse_utc(payload["generatedAt"])
        if not isinstance(payload.get("items"), list):
            raise ValueError("packaged dataset is invalid")
        with tempfile.NamedTemporaryFile("wb", dir=data_dir, prefix=".seed-data-", delete=False) as tmp:
            tmp.write(raw)
            temp_path = Path(tmp.name)
        try:
            temp_path.chmod(0o640)
            temp_path.replace(destination)
        except Exception:
            temp_path.unlink(missing_ok=True)
            raise

    gzip_destination = data_dir / f"{filename}.gz"
    try:
        if gzip.decompress(gzip_destination.read_bytes()) == raw:
            return
    except (OSError, EOFError, gzip.BadGzipFile):
        pass
    with tempfile.NamedTemporaryFile("wb", dir=data_dir, prefix=".seed-gzip-", delete=False) as tmp:
        tmp.write(gzip.compress(raw, mtime=0))
        temp_path = Path(tmp.name)
    try:
        temp_path.chmod(0o640)
        temp_path.replace(gzip_destination)
    except Exception:
        temp_path.unlink(missing_ok=True)
        raise


def static_response(environ: dict, start_response, root: Path, data_dir: Path):
    method = environ.get("REQUEST_METHOD", "GET").upper()
    if method not in {"GET", "HEAD"}:
        return json_response(start_response, "405 Method Not Allowed", {"error": "method not allowed"}, [("Allow", "GET, HEAD")])

    request_path = environ.get("PATH_INFO", "/")
    if request_path == "/microsoft":
        start_response("308 Permanent Redirect", [("Location", "/microsoft/"), ("Content-Length", "0"), *SECURITY_HEADERS])
        return [b""]
    if request_path == "/microsoft/cves":
        start_response("308 Permanent Redirect", [("Location", "/microsoft/cves/"), ("Content-Length", "0"), *SECURITY_HEADERS])
        return [b""]
    if request_path in {"/", "/microsoft/"}:
        file_path = root / "index.html"
    elif request_path == "/microsoft/cves/":
        file_path = root / "cves.html"
    elif request_path in {"/data.json", "/microsoft/data.json"}:
        file_path = data_dir / "data.json"
    elif request_path == "/microsoft/cve-data.json":
        file_path = data_dir / "cve-data.json"
    else:
        name = request_path.removeprefix("/microsoft/")
        if name not in STATIC_FILES:
            return json_response(start_response, "404 Not Found", {"error": "not found"})
        file_path = root / name

    gzip_path = file_path.with_suffix(file_path.suffix + ".gz")
    use_gzip = False
    if file_path.name in DATASET_FILES and "gzip" in environ.get("HTTP_ACCEPT_ENCODING", "").lower():
        try:
            use_gzip = gzip_path.stat().st_mtime_ns >= file_path.stat().st_mtime_ns
        except OSError:
            pass
    served_path = gzip_path if use_gzip else file_path
    try:
        stat = served_path.stat()
    except OSError:
        return json_response(start_response, "404 Not Found", {"error": "not found"})

    etag = f'"{stat.st_mtime_ns:x}-{stat.st_size:x}"'
    content_type = "application/json; charset=utf-8" if file_path.name in DATASET_FILES else (mimetypes.guess_type(file_path.name)[0] or "application/octet-stream")
    headers = [
        ("Content-Type", content_type),
        ("ETag", etag),
        ("Last-Modified", formatdate(stat.st_mtime, usegmt=True)),
        ("Cache-Control", "no-cache" if file_path.name in DATASET_FILES else "public, max-age=300"),
        *SECURITY_HEADERS,
    ]
    if file_path.name in DATASET_FILES:
        headers.append(("Vary", "Accept-Encoding"))
    if use_gzip:
        headers.append(("Content-Encoding", "gzip"))
    if environ.get("HTTP_IF_NONE_MATCH") == etag:
        start_response("304 Not Modified", [("Content-Length", "0"), *headers])
        return [b""]
    try:
        raw = served_path.read_bytes()
    except OSError:
        return json_response(start_response, "404 Not Found", {"error": "not found"})
    headers.append(("Content-Length", str(len(raw))))
    start_response("200 OK", headers)
    return [b"" if method == "HEAD" else raw]


def refresh_response(
    environ: dict,
    start_response,
    *,
    data_path: Path,
    state_path: Path,
    lock_path: Path,
    runner: Callable[[Path, int, int], None],
    clock: Callable[[], datetime],
    allowed_origin: str | None,
    max_age_seconds: int = MAX_AGE_SECONDS,
):
    method = environ.get("REQUEST_METHOD", "GET").upper()
    if method != "POST":
        return json_response(start_response, "405 Method Not Allowed", {"error": "method not allowed"}, [("Allow", "POST")])
    valid, reason = valid_json_request(environ)
    if not valid:
        code = {
            "unsupported": "415 Unsupported Media Type",
            "oversized": "413 Content Too Large",
        }.get(reason, "400 Bad Request")
        return json_response(start_response, code, {"error": reason})
    if not origin_is_same(environ, allowed_origin):
        return json_response(start_response, "403 Forbidden", {"error": "same-origin request required"})

    lock_path.touch(exist_ok=True)
    with lock_path.open("a+") as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return json_response(start_response, "202 Accepted", {"status": "running"}, [("Retry-After", "3")])
        try:
            now = clock().astimezone(timezone.utc)
            if not dataset_is_stale(data_path, now, max_age_seconds):
                dataset = read_dataset(data_path)
                return json_response(start_response, "200 OK", {"status": "fresh", "generatedAt": dataset["generatedAt"]})

            state = read_state(state_path)
            try:
                next_allowed = parse_utc(state.get("nextAllowedAt", ""))
            except (TypeError, ValueError):
                next_allowed = datetime.min.replace(tzinfo=timezone.utc)
            latest_reasonable = now + timedelta(seconds=BACKOFF_SECONDS[-1])
            if next_allowed > latest_reasonable:
                next_allowed = latest_reasonable
            if now < next_allowed:
                retry = max(1, int((next_allowed - now).total_seconds()))
                return json_response(start_response, "503 Service Unavailable", {
                    "status": "backoff",
                    "retryAfter": retry,
                    "nextAllowedAt": utc_text(next_allowed),
                }, [("Retry-After", str(retry))])

            refresh_error = None
            try:
                runner(data_path, REFRESH_TIMEOUT_SECONDS, lock.fileno())
                dataset = read_dataset(data_path)
                if dataset_is_stale(data_path, clock(), max_age_seconds):
                    raise RuntimeError("collector output remained stale")
            except Exception as error:
                refresh_error = error
                try:
                    dataset = read_dataset(data_path)
                    if dataset_is_stale(data_path, clock(), max_age_seconds):
                        raise RuntimeError("collector did not publish fresh data")
                except Exception:
                    dataset = None

            if dataset is None:
                try:
                    previous_failures = int(state.get("failures") or 0)
                except (TypeError, ValueError):
                    previous_failures = 0
                failures = max(0, previous_failures) + 1
                delay = backoff_seconds(failures)
                finished = clock().astimezone(timezone.utc)
                next_allowed = finished + timedelta(seconds=delay)
                atomic_state(state_path, {
                    "failures": failures,
                    "lastAttemptAt": utc_text(finished),
                    "nextAllowedAt": utc_text(next_allowed),
                    "status": "failed",
                })
                LOG.error("Dashboard refresh failed; next attempt after %s", utc_text(next_allowed), exc_info=refresh_error)
                return json_response(start_response, "503 Service Unavailable", {
                    "status": "failed",
                    "retryAfter": delay,
                    "nextAllowedAt": utc_text(next_allowed),
                }, [("Retry-After", str(delay))])

            finished = clock().astimezone(timezone.utc)
            atomic_state(state_path, {
                "failures": 0,
                "lastAttemptAt": utc_text(finished),
                "nextAllowedAt": "",
                "status": "success",
                "generatedAt": dataset["generatedAt"],
            })
            return json_response(start_response, "200 OK", {
                "status": "updated",
                "generatedAt": dataset["generatedAt"],
                "summary": dataset.get("summary", {}),
            })
        finally:
            fcntl.flock(lock, fcntl.LOCK_UN)


def create_app(
    *,
    root: Path | None = None,
    data_dir: Path | None = None,
    runner: Callable[[Path, int, int], None] | None = None,
    cve_runner: Callable[[Path, int, int], None] | None = None,
    clock: Callable[[], datetime] = utc_now,
    allowed_origin: str | None = None,
):
    root = Path(root or os.environ.get("DASHBOARD_ROOT") or Path(__file__).parent)
    data_dir = Path(data_dir or os.environ.get("DASHBOARD_DATA_DIR") or root)
    data_dir.mkdir(parents=True, exist_ok=True)
    seed_persistent_dataset(root, data_dir, "data.json")
    seed_persistent_dataset(root, data_dir, "cve-data.json")
    data_path = data_dir / "data.json"
    state_path = data_dir / ".refresh-state.json"
    lock_path = data_dir / ".refresh.lock"
    cve_data_path = data_dir / "cve-data.json"
    cve_state_path = data_dir / ".cve-refresh-state.json"
    cve_lock_path = data_dir / ".cve-refresh.lock"
    runner = runner or subprocess_runner(root)
    cve_runner = cve_runner or cve_subprocess_runner(root)
    allowed_origin = allowed_origin or os.environ.get("DASHBOARD_ORIGIN")

    def app(environ, start_response):
        path = environ.get("PATH_INFO", "")
        if path == "/api/refresh-if-stale":
            return refresh_response(
                environ, start_response,
                data_path=data_path,
                state_path=state_path,
                lock_path=lock_path,
                runner=runner,
                clock=clock,
                allowed_origin=allowed_origin,
            )
        if path == "/api/refresh-cves-if-stale":
            return refresh_response(
                environ, start_response,
                data_path=cve_data_path,
                state_path=cve_state_path,
                lock_path=cve_lock_path,
                runner=cve_runner,
                clock=clock,
                allowed_origin=allowed_origin,
            )
        return static_response(environ, start_response, root, data_dir)

    return app


app = create_app()