Open raw ↗"""The reference web application — one process, no dependencies beyond the Python stdlib.
Pattern of GOV-B7.4 (Standard v3.9): deploy-anywhere, named-user login, pluggable
AI-engine connector, keys server-side only. Static file serving is containment-guarded
(resolve + relative_to — no traversal); responses carry a conservative CSP; the chat
route refuses anyone without a session cookie.
"""
import json
import secrets
import threading
from http import cookies as http_cookies
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit
from adapters.glm import GLMAdapter
from adapters.stub import StubAdapter
from users import UserStore
BASE = Path(__file__).resolve().parent
STATIC = BASE / "static"
CSP = "default-src 'self'; style-src 'self'; script-src 'self'; connect-src 'self'; img-src 'self' data:; frame-ancestors 'none'"
sessions: dict[str, str] = {} # token -> username
sessions_lock = threading.Lock()
store = UserStore()
ADAPTERS = {"glm": GLMAdapter, "stub": StubAdapter}
adapter = None # set by create_app()
def _contained(p: Path, root: Path) -> Path:
p = p.resolve()
p.relative_to(root) # raises ValueError on any escape attempt
return p
def create_app(engine: str = "glm") -> None:
global adapter
adapter = ADAPTERS[engine]()
class Handler(BaseHTTPRequestHandler):
server_version = "govref/1.0"
# ---------- helpers ----------
def _send(self, code: int, body: bytes, ctype: str, extra: dict | None = None) -> None:
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Content-Security-Policy", CSP)
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("Cache-Control", "no-store")
for k, v in (extra or {}).items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(body)
def _json(self, code: int, obj: dict, extra: dict | None = None) -> None:
self._send(code, json.dumps(obj).encode("utf-8"), "application/json; charset=utf-8", extra)
def _session_user(self) -> str | None:
raw = self.headers.get("Cookie", "")
jar = http_cookies.SimpleCookie()
try:
jar.load(raw)
except http_cookies.CookieError:
return None
token = jar.get("session")
if not token:
return None
with sessions_lock:
return sessions.get(token.value)
def _read_json(self) -> dict:
length = int(self.headers.get("Content-Length", 0) or 0)
if length <= 0 or length > 65536:
return {}
try:
return json.loads(self.rfile.read(length).decode("utf-8"))
except (ValueError, UnicodeDecodeError):
return {}
def log_message(self, fmt, *args): # keep the console quiet and key-free
print(f"{self.address_string()} {fmt % args}")
# ---------- routes ----------
def do_GET(self) -> None: # noqa: N802
path = urlsplit(self.path).path
if path in ("/", "/index.html"):
body = _contained(STATIC / "index.html", STATIC).read_bytes()
self._send(200, body, "text/html; charset=utf-8")
return
if path.startswith("/static/"):
name = path[len("/static/"):]
root = STATIC.resolve()
candidate = root / name
try:
file = _contained(candidate, root)
except ValueError:
self._json(404, {"error": "not found"})
return
if not file.is_file():
self._json(404, {"error": "not found"})
return
ctype = ("text/css" if file.suffix == ".css"
else "application/javascript" if file.suffix == ".js"
else "application/octet-stream")
self._send(200, file.read_bytes(), ctype + "; charset=utf-8")
return
if path == "/api/whoami":
user = self._session_user()
self._json(200, {"user": user, "engine": adapter.NAME} if user else {"user": None})
return
self._json(404, {"error": "not found"})
def do_POST(self) -> None: # noqa: N802
path = urlsplit(self.path).path
if path == "/api/login":
data = self._read_json()
username = str(data.get("username", ""))[:64]
password = str(data.get("password", ""))[:256]
if username and password and store.check(username, password):
token = secrets.token_urlsafe(32)
with sessions_lock:
sessions[token] = username
cookie = f"session={token}; HttpOnly; SameSite=Strict; Path=/; Max-Age=86400"
self._json(200, {"ok": True, "user": username}, {"Set-Cookie": cookie})
return
self._json(401, {"error": "invalid credentials"})
return
if path == "/api/logout":
raw = self.headers.get("Cookie", "")
jar = http_cookies.SimpleCookie()
try:
jar.load(raw)
except http_cookies.CookieError:
pass
token = jar.get("session")
if token:
with sessions_lock:
sessions.pop(token.value, None)
self._json(200, {"ok": True}, {"Set-Cookie": "session=; HttpOnly; SameSite=Strict; Path=/; Max-Age=0"})
return
if path == "/api/chat":
user = self._session_user()
if not user:
self._json(401, {"error": "login required"})
return
data = self._read_json()
text = str(data.get("message", "")).strip()[:8000]
if not text:
self._json(400, {"error": "empty message"})
return
history = [{"role": "system", "content": "You are a helpful assistant inside a governed project web app."},
{"role": "user", "content": text}]
try:
reply = adapter.chat(history)
except Exception as exc: # engine errors never leak keys or internals
self._json(502, {"error": f"engine error: {type(exc).__name__}"})
return
self._json(200, {"reply": reply})
return
self._json(404, {"error": "not found"})
def serve(host: str, port: int, engine: str) -> None:
create_app(engine)
httpd = ThreadingHTTPServer((host, port), Handler)
print(f"reference web app listening on http://{host}:{port} · engine: {engine} · login: the admin credentials printed at first run")
httpd.serve_forever()