svrnty-hermes-webui-plugin/routes/bte_proxy.py

"""GET|POST /api/bte/proxy?path=<bte-endpoint> — same-origin proxy to BTE REST.

The browser hits the plugin (same origin as webui :8787), the plugin forwards
to BTE REST (:6001 by default). This avoids cross-origin CORS, keeps BTE
free of webui-origin allowances, and lets us whitelist exactly which BTE
endpoints the panel can reach.

Configuration (read at call time, never persisted):
  BTE_BASE_URL    BTE REST base (default http://localhost:6001)
  BTE_TENANT_ID   default X-Tenant-Id forwarded to BTE (Plan B tenant uuid)

Public API surface used: api.register_route, api.logger.
Stdlib only — no urllib3, no requests.
"""
import json
import os
import re
import urllib.error
import urllib.parse
import urllib.request

_DEFAULT_BTE_BASE = "http://localhost:6001"
_DEFAULT_TENANT = "00000000-0000-0000-0000-000000000001"  # Plan B tenant

# Whitelist of allowed BTE paths. Static prefixes match exact path; pattern
# entries match a literal `/api/assets/<id>/<suffix>` form. Keep tight — the
# proxy is a privilege amplifier, only the panel's needs go here.
_ALLOWED_EXACT = frozenset({
    "/api/command/requestPhotoshoot",
    "/api/command/rateAsset",
    "/api/query/assetGrid",
    "/api/query/recipeStats",
})
# Pattern: /api/assets/<uuid-or-id>/(thumb|status). id-segment may not contain '/' or '?'.
_ALLOWED_PATTERN = re.compile(r"^/api/assets/[A-Za-z0-9_\-]+/(thumb|status)$")


def register(api):
    """Wire the GET + POST /api/bte/proxy routes."""
    log = api.logger("svrnty.routes.bte_proxy")
    api.register_route("/api/bte/proxy", "GET", _handle_proxy)
    api.register_route("/api/bte/proxy", "POST", _handle_proxy)
    log.info("bte proxy endpoint registered (GET+POST)")


def _is_allowed(path: str) -> bool:
    if path in _ALLOWED_EXACT:
        return True
    if _ALLOWED_PATTERN.match(path):
        return True
    return False


def _handle_proxy(handler, parsed):
    """Forward a single request to BTE. Returns BTE response body verbatim."""
    qs = urllib.parse.parse_qs(parsed.query or "")
    target_path = (qs.get("path") or [""])[0].strip()
    if not target_path or not target_path.startswith("/api/"):
        return _send_json(handler, {"ok": False, "error": "missing or invalid path"}, 400)
    if not _is_allowed(target_path):
        return _send_json(handler, {"ok": False, "error": f"path not allowed: {target_path}"}, 403)

    base = os.environ.get("BTE_BASE_URL", _DEFAULT_BTE_BASE).rstrip("/")
    tenant = (handler.headers.get("X-Tenant-Id")
              or os.environ.get("BTE_TENANT_ID", _DEFAULT_TENANT))
    target_url = base + target_path

    method = handler.command  # 'GET' / 'POST'
    body = b""
    content_type = handler.headers.get("Content-Type", "application/json")
    if method in ("POST", "PUT", "PATCH"):
        length = int(handler.headers.get("Content-Length", "0") or 0)
        if length > 0:
            body = handler.rfile.read(length)

    req = urllib.request.Request(target_url, data=body if body else None, method=method)
    req.add_header("X-Tenant-Id", tenant)
    if body:
        req.add_header("Content-Type", content_type)
    auth = handler.headers.get("Authorization")
    if auth:
        req.add_header("Authorization", auth)

    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            resp_body = resp.read()
            resp_ctype = resp.headers.get("Content-Type", "application/octet-stream")
            resp_status = resp.status
    except urllib.error.HTTPError as e:
        # Forward BTE's own error status + body — panel renders "endpoint coming
        # soon" placeholders when it sees 404/501.
        resp_body = e.read() or b""
        resp_ctype = e.headers.get("Content-Type", "application/json") if e.headers else "application/json"
        resp_status = e.code
    except urllib.error.URLError as e:
        return _send_json(handler, {"ok": False, "error": f"bte unreachable: {e.reason}"}, 502)
    except Exception as e:
        return _send_json(handler, {"ok": False, "error": f"proxy error: {e}"}, 500)

    handler.send_response(resp_status)
    handler.send_header("Content-Type", resp_ctype)
    handler.send_header("Content-Length", str(len(resp_body)))
    handler.send_header("Cache-Control", "no-store")
    handler.end_headers()
    handler.wfile.write(resp_body)
    return True


def _send_json(handler, payload: dict, status: int) -> bool:
    body = json.dumps(payload).encode("utf-8")
    handler.send_response(status)
    handler.send_header("Content-Type", "application/json; charset=utf-8")
    handler.send_header("Content-Length", str(len(body)))
    handler.send_header("Cache-Control", "no-store")
    handler.end_headers()
    handler.wfile.write(body)
    return True