portal-dev

#!/usr/bin/env python3
"""Local portal preview with a same-origin API proxy.

Serves Hexo `docs/` and proxies `/machine-api/*` → machine-api so the browser
never hits cross-origin CORS (useful before the API CORS change is deployed).

  ./portal-dev              # http://127.0.0.1:4000
  ./portal-dev --port 4010
  TF_MACHINE_API_URL=https://machine-api.testflows.com ./portal-dev
"""

from __future__ import annotations

import argparse
import os
import sys
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

ROOT = Path(__file__).resolve().parent
DOCS = ROOT / "docs"
DEFAULT_API = "https://machine-api.testflows.com"
PROXY_PREFIX = "/machine-api"


def _dev_cookie(value: str) -> str:
    """Rewrite an upstream Set-Cookie so it stores over http://localhost: drop
    `Secure` (never stored on http) and any `Domain` (must bind to localhost), and
    relax `SameSite` to Lax. HttpOnly / Path / Max-Age are preserved, so the
    HttpOnly session cookie behaves the same as in production, just locally."""
    keep = []
    for attr in value.split(";"):
        a = attr.strip()
        low = a.lower()
        if low == "secure" or low.startswith("domain="):
            continue
        if low.startswith("samesite="):
            a = "SameSite=Lax"
        keep.append(a)
    return "; ".join(keep)


class PortalHandler(SimpleHTTPRequestHandler):
    """Static docs + reverse proxy for Machine API."""

    api_base: str = DEFAULT_API

    def __init__(self, *args, **kwargs):
        super().__init__(*args, directory=str(DOCS), **kwargs)

    def log_message(self, fmt: str, *args) -> None:
        sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args))

    def send_error(self, code, message=None, explain=None) -> None:
        """Serve the site's 404 page (as GitHub Pages does) for missing GET/HEAD paths."""
        page = DOCS / "404.html"
        if code == 404 and self.command in ("GET", "HEAD") and page.is_file():
            data = page.read_bytes()
            self.send_response(404)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            if self.command == "GET":
                self.wfile.write(data)
            return
        super().send_error(code, message, explain)

    def do_OPTIONS(self) -> None:
        if self.path.startswith(PROXY_PREFIX):
            self._proxy()
            return
        self.send_error(404)

    def do_GET(self) -> None:
        if self.path.startswith(PROXY_PREFIX):
            self._proxy()
            return
        super().do_GET()

    def do_POST(self) -> None:
        if self.path.startswith(PROXY_PREFIX):
            self._proxy()
            return
        self.send_error(405, "Method Not Allowed")

    def do_PUT(self) -> None:
        if self.path.startswith(PROXY_PREFIX):
            self._proxy()
            return
        self.send_error(405, "Method Not Allowed")

    def do_PATCH(self) -> None:
        if self.path.startswith(PROXY_PREFIX):
            self._proxy()
            return
        self.send_error(405, "Method Not Allowed")

    def do_DELETE(self) -> None:
        if self.path.startswith(PROXY_PREFIX):
            self._proxy()
            return
        self.send_error(405, "Method Not Allowed")

    def _proxy(self) -> None:
        suffix = self.path[len(PROXY_PREFIX) :] or "/"
        if not suffix.startswith("/"):
            suffix = "/" + suffix
        url = self.api_base.rstrip("/") + suffix
        length = int(self.headers.get("Content-Length", "0") or "0")
        body = self.rfile.read(length) if length > 0 else None
        headers = {}
        for key in (
            "Content-Type",
            "Authorization",
            "Cookie",
            # Forward the browser's User-Agent so a login through this proxy is captured
            # with the real device label (else urllib sends "Python-urllib" and the
            # signed-in-devices list shows a raw "python-urllib/…" row for a browser).
            "User-Agent",
            "X-Tf-Client",
            "X-Tf-Pow",
            "X-Tf-Code",
            "Idempotency-Key",
            "Accept",
        ):
            val = self.headers.get(key)
            if val:
                headers[key] = val
        req = Request(url, data=body, headers=headers, method=self.command)
        try:
            with urlopen(req, timeout=60) as resp:
                data = resp.read()
                self.send_response(resp.status)
                for hop in (
                    "Content-Type",
                    "Cache-Control",
                    "Content-Length",
                ):
                    value = resp.headers.get(hop)
                    if value:
                        self.send_header(hop, value)
                for cookie in resp.headers.get_all("Set-Cookie") or []:
                    self.send_header("Set-Cookie", _dev_cookie(cookie))
                # Same-origin proxy — no CORS headers required.
                self.end_headers()
                self.wfile.write(data)
        except HTTPError as exc:
            data = exc.read()
            self.send_response(exc.code)
            ctype = exc.headers.get("Content-Type")
            if ctype:
                self.send_header("Content-Type", ctype)
            for cookie in exc.headers.get_all("Set-Cookie") or []:
                self.send_header("Set-Cookie", _dev_cookie(cookie))
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)
        except URLError as exc:
            msg = f"proxy upstream error: {exc.reason}".encode()
            self.send_response(502)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.send_header("Content-Length", str(len(msg)))
            self.end_headers()
            self.wfile.write(msg)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--port", type=int, default=4000)
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument(
        "--api",
        default=os.environ.get("TF_MACHINE_API_URL", DEFAULT_API),
        help="Upstream Machine API base URL",
    )
    args = parser.parse_args()
    if not DOCS.is_dir():
        print(f"missing {DOCS} — run: npx hexo generate", file=sys.stderr)
        return 1
    PortalHandler.api_base = args.api.rstrip("/")
    server = ThreadingHTTPServer((args.host, args.port), PortalHandler)
    print(
        f"portal-dev  http://{args.host}:{args.port}/machine/portal/login/\n"
        f"api proxy   {PROXY_PREFIX}/ → {PortalHandler.api_base}/\n"
        f"(same-origin — no CORS needed)"
    )
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())