296 lines
11 KiB
Python
Executable File
296 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Loopback-only HTTP server for the CASAN Core Local Assurance Viewer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import threading
|
|
from urllib.parse import parse_qs, unquote, urlparse
|
|
|
|
from local_report import (
|
|
SAFE_ID,
|
|
h6_report,
|
|
project_profile,
|
|
report_html,
|
|
run_report,
|
|
runs,
|
|
trace_graph,
|
|
)
|
|
|
|
|
|
ASSET_ROOT = Path(__file__).resolve().parents[2] / "assets" / "local-viewer"
|
|
MAX_RESPONSE_BYTES = 24 * 1024 * 1024
|
|
|
|
|
|
def atomic_json(path: Path, payload: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
|
temporary.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
|
try:
|
|
os.chmod(temporary, 0o600)
|
|
except OSError:
|
|
if os.name != "nt":
|
|
raise
|
|
os.replace(temporary, path)
|
|
|
|
|
|
class ViewerServer(ThreadingHTTPServer):
|
|
daemon_threads = True
|
|
allow_reuse_address = False
|
|
|
|
def __init__(self, address, root: Path, token: str):
|
|
self.root = root.resolve()
|
|
self.token = token
|
|
super().__init__(address, ViewerHandler)
|
|
|
|
|
|
class ViewerHandler(BaseHTTPRequestHandler):
|
|
server_version = "CASAN-Core-Viewer/1"
|
|
sys_version = ""
|
|
|
|
@property
|
|
def viewer(self) -> ViewerServer:
|
|
return self.server # type: ignore[return-value]
|
|
|
|
def log_message(self, fmt: str, *args) -> None:
|
|
# Never persist the session token from the initial/query download URL.
|
|
print(
|
|
"%s - - [%s] %s %s" % (
|
|
self.client_address[0],
|
|
self.log_date_time_string(),
|
|
self.command,
|
|
urlparse(self.path).path,
|
|
),
|
|
flush=True,
|
|
)
|
|
|
|
def _host_allowed(self) -> bool:
|
|
host = self.headers.get("Host", "").split(":", 1)[0].strip("[]").lower()
|
|
return host in {"127.0.0.1", "localhost", "::1"}
|
|
|
|
def _authorized(self, query: dict[str, list[str]]) -> bool:
|
|
supplied = self.headers.get("X-CASAN-Viewer-Token", "")
|
|
if not supplied:
|
|
supplied = query.get("token", [""])[0]
|
|
return bool(supplied) and supplied == self.viewer.token
|
|
|
|
def _headers(self, status: int, content_type: str, length: int) -> None:
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(length))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("Pragma", "no-cache")
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
self.send_header("X-Frame-Options", "DENY")
|
|
self.send_header("Referrer-Policy", "no-referrer")
|
|
self.send_header("Cross-Origin-Resource-Policy", "same-origin")
|
|
self.send_header(
|
|
"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'; frame-ancestors 'none'; form-action 'self'",
|
|
)
|
|
self.end_headers()
|
|
|
|
def _send(self, status: int, body: bytes, content_type: str) -> None:
|
|
if len(body) > MAX_RESPONSE_BYTES:
|
|
self._json(413, {"error": "CASAN_VIEW_RESPONSE_TOO_LARGE"})
|
|
return
|
|
self._headers(status, content_type, len(body))
|
|
if self.command != "HEAD":
|
|
self.wfile.write(body)
|
|
|
|
def _json(self, status: int, payload) -> None:
|
|
self._send(
|
|
status,
|
|
(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8"),
|
|
"application/json; charset=utf-8",
|
|
)
|
|
|
|
def _asset(self, name: str) -> None:
|
|
safe_name = "index.html" if name in {"", "/"} else name.lstrip("/")
|
|
if safe_name not in {"index.html", "app.css", "app.js", "favicon.svg"}:
|
|
self._json(404, {"error": "CASAN_VIEW_NOT_FOUND"})
|
|
return
|
|
path = ASSET_ROOT / safe_name
|
|
try:
|
|
body = path.read_bytes()
|
|
except OSError:
|
|
self._json(503, {"error": "CASAN_VIEW_ASSET_MISSING"})
|
|
return
|
|
content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
|
|
if content_type.startswith("text/") or content_type == "application/javascript":
|
|
content_type += "; charset=utf-8"
|
|
self._send(200, body, content_type)
|
|
|
|
def _download(self, report: dict, fmt: str, filename: str) -> None:
|
|
if fmt == "json":
|
|
body = (json.dumps(report, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
|
|
content_type = "application/json; charset=utf-8"
|
|
elif fmt == "html":
|
|
body = report_html(report, "h6" if report.get("harness") == "H6" else "run").encode("utf-8")
|
|
content_type = "text/html; charset=utf-8"
|
|
else:
|
|
self._json(400, {"error": "CASAN_REPORT_FORMAT_INVALID"})
|
|
return
|
|
if len(body) > MAX_RESPONSE_BYTES:
|
|
self._json(413, {"error": "CASAN_REPORT_TOO_LARGE"})
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Disposition", f'attachment; filename="{filename}.{fmt}"')
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
self.send_header("Referrer-Policy", "no-referrer")
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(body)
|
|
|
|
def do_HEAD(self) -> None:
|
|
self.do_GET()
|
|
|
|
def do_GET(self) -> None:
|
|
if self.client_address[0] not in {"127.0.0.1", "::1"} or not self._host_allowed():
|
|
self._json(403, {"error": "CASAN_VIEW_LOOPBACK_ONLY"})
|
|
return
|
|
parsed = urlparse(self.path)
|
|
path = unquote(parsed.path)
|
|
query = parse_qs(parsed.query, keep_blank_values=False)
|
|
if path in {"/", "/index.html", "/app.css", "/app.js"}:
|
|
self._asset(path)
|
|
return
|
|
if not self._authorized(query):
|
|
self._json(401, {"error": "CASAN_VIEW_TOKEN_REQUIRED"})
|
|
return
|
|
try:
|
|
if path == "/healthz":
|
|
self._json(200, {
|
|
"status": "ok",
|
|
"mode": "local_core",
|
|
"project_id": project_profile(self.viewer.root)["project_id"],
|
|
"pid": os.getpid(),
|
|
})
|
|
elif path == "/api/v1/project":
|
|
self._json(200, project_profile(self.viewer.root))
|
|
elif path == "/api/v1/runs":
|
|
self._json(200, runs(self.viewer.root, int(query.get("limit", ["100"])[0])))
|
|
elif path == "/api/v1/reports/h6":
|
|
report = self._h6(query)
|
|
self._json(200, report)
|
|
elif path == "/api/v1/reports/h6/export":
|
|
report = self._h6(query)
|
|
self._download(report, query.get("format", ["html"])[0], report["report_id"])
|
|
elif path == "/__shutdown":
|
|
self._json(200, {"status": "stopping"})
|
|
threading.Thread(target=self.viewer.shutdown, daemon=True).start()
|
|
elif path.startswith("/api/v1/runs/") and path.endswith("/graph"):
|
|
trace_id = path[len("/api/v1/runs/"):-len("/graph")].strip("/")
|
|
self._trace_required(trace_id)
|
|
self._json(200, trace_graph(self.viewer.root, trace_id))
|
|
elif path.startswith("/api/v1/reports/run/"):
|
|
suffix = path[len("/api/v1/reports/run/"):].strip("/")
|
|
exporting = suffix.endswith("/export")
|
|
trace_id = suffix[:-len("/export")].strip("/") if exporting else suffix
|
|
self._trace_required(trace_id)
|
|
report = run_report(self.viewer.root, trace_id)
|
|
if report["verdict"] == "not_found":
|
|
self._json(404, {"error": "CASAN_RUN_NOT_FOUND"})
|
|
elif exporting:
|
|
self._download(
|
|
report,
|
|
query.get("format", ["html"])[0],
|
|
report["report_id"],
|
|
)
|
|
else:
|
|
self._json(200, report)
|
|
else:
|
|
self._json(404, {"error": "CASAN_VIEW_NOT_FOUND"})
|
|
except (TypeError, ValueError):
|
|
self._json(400, {"error": "CASAN_VIEW_INVALID_QUERY"})
|
|
except BrokenPipeError:
|
|
return
|
|
except Exception as error: # fail closed without leaking project paths
|
|
self.log_error("request failed: %s", type(error).__name__)
|
|
self._json(500, {"error": "CASAN_VIEW_INTERNAL_ERROR"})
|
|
|
|
def _trace_required(self, trace_id: str) -> None:
|
|
if not SAFE_ID.fullmatch(trace_id):
|
|
raise ValueError("unsafe trace id")
|
|
|
|
def _h6(self, query: dict[str, list[str]]) -> dict:
|
|
return h6_report(
|
|
self.viewer.root,
|
|
project=query.get("project", [None])[0],
|
|
run=query.get("run", [None])[0],
|
|
from_date=query.get("from", [None])[0],
|
|
to_date=query.get("to", [None])[0],
|
|
limit=int(query.get("limit", ["50"])[0]),
|
|
)
|
|
|
|
def do_POST(self) -> None:
|
|
self._json(405, {"error": "CASAN_VIEW_READ_ONLY"})
|
|
|
|
do_PUT = do_POST
|
|
do_PATCH = do_POST
|
|
do_DELETE = do_POST
|
|
|
|
|
|
def serve(root: Path, port: int, token: str, state_file: Path) -> int:
|
|
if not token or len(token) < 32:
|
|
raise ValueError("viewer token is too short")
|
|
server = ViewerServer(("127.0.0.1", port), root, token)
|
|
actual_port = server.server_address[1]
|
|
atomic_json(state_file, {
|
|
"schema_version": 1,
|
|
"pid": os.getpid(),
|
|
"port": actual_port,
|
|
"token": token,
|
|
"project_root": str(root.resolve()),
|
|
"started_at": __import__("datetime").datetime.now(
|
|
__import__("datetime").timezone.utc
|
|
).isoformat().replace("+00:00", "Z"),
|
|
})
|
|
|
|
def stop(_signum, _frame):
|
|
threading.Thread(target=server.shutdown, daemon=True).start()
|
|
|
|
signal.signal(signal.SIGTERM, stop)
|
|
signal.signal(signal.SIGINT, stop)
|
|
try:
|
|
server.serve_forever(poll_interval=0.25)
|
|
finally:
|
|
server.server_close()
|
|
try:
|
|
current = json.loads(state_file.read_text(encoding="utf-8"))
|
|
if current.get("pid") == os.getpid():
|
|
state_file.unlink()
|
|
except (OSError, ValueError):
|
|
pass
|
|
return 0
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--root", required=True)
|
|
parser.add_argument("--port", type=int, default=0)
|
|
parser.add_argument("--token", required=True)
|
|
parser.add_argument("--state-file", required=True)
|
|
args = parser.parse_args(argv)
|
|
return serve(
|
|
Path(args.root).expanduser().resolve(),
|
|
max(0, min(args.port, 65535)),
|
|
args.token,
|
|
Path(args.state_file).expanduser().resolve(),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|