122 lines
4.6 KiB
Python
122 lines
4.6 KiB
Python
"""本地配对服务：接收浏览器发来的 OMS Token。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import Callable
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_PAIR_PORT = 19333
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PairPayload:
|
|
token: str
|
|
oms_url: str
|
|
web_origin: str = ""
|
|
|
|
|
|
class PairServer:
|
|
"""127.0.0.1 本地 HTTP 服务，供 OMS 网页推送 JWT。"""
|
|
|
|
def __init__(self, port: int = DEFAULT_PAIR_PORT) -> None:
|
|
self.port = port
|
|
self._server: ThreadingHTTPServer | None = None
|
|
self._thread: threading.Thread | None = None
|
|
self._waiting = False
|
|
self._on_pair: Callable[[PairPayload], None] | None = None
|
|
|
|
@property
|
|
def is_waiting(self) -> bool:
|
|
return self._waiting
|
|
|
|
def start(self, on_pair: Callable[[PairPayload], None]) -> None:
|
|
self.stop()
|
|
self._on_pair = on_pair
|
|
self._waiting = True
|
|
server_ref = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, fmt: str, *args) -> None:
|
|
logger.debug(fmt, *args)
|
|
|
|
def _write_cors(self) -> None:
|
|
origin = self.headers.get("Origin", "")
|
|
if origin.startswith(("http://localhost", "http://127.0.0.1", "https://")):
|
|
self.send_header("Access-Control-Allow-Origin", origin)
|
|
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
|
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
|
|
|
def do_OPTIONS(self) -> None:
|
|
self.send_response(204)
|
|
self._write_cors()
|
|
self.end_headers()
|
|
|
|
def do_GET(self) -> None:
|
|
if self.path.split("?", 1)[0].rstrip("/") != "/status":
|
|
self.send_error(404)
|
|
return
|
|
body = json.dumps({"waiting": server_ref._waiting})
|
|
self.send_response(200)
|
|
self._write_cors()
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(body.encode("utf-8"))
|
|
|
|
def do_POST(self) -> None:
|
|
if self.path.split("?", 1)[0].rstrip("/") != "/pair":
|
|
self.send_error(404)
|
|
return
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
raw = self.rfile.read(length) if length else b"{}"
|
|
try:
|
|
data = json.loads(raw.decode("utf-8"))
|
|
except json.JSONDecodeError:
|
|
self.send_error(400)
|
|
return
|
|
if not isinstance(data, dict):
|
|
self.send_error(400)
|
|
return
|
|
|
|
token = str(data.get("token") or "").strip()
|
|
oms_url = str(data.get("omsUrl") or data.get("oms_url") or "").strip().rstrip("/")
|
|
web_origin = str(data.get("webOrigin") or data.get("web_origin") or "").strip().rstrip("/")
|
|
if not token or not oms_url:
|
|
payload = json.dumps({"ok": False, "message": "token 与 omsUrl 必填"})
|
|
self.send_response(400)
|
|
self._write_cors()
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(payload.encode("utf-8"))
|
|
return
|
|
|
|
server_ref._waiting = False
|
|
if server_ref._on_pair:
|
|
server_ref._on_pair(
|
|
PairPayload(token=token, oms_url=oms_url, web_origin=web_origin)
|
|
)
|
|
payload = json.dumps({"ok": True})
|
|
self.send_response(200)
|
|
self._write_cors()
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(payload.encode("utf-8"))
|
|
|
|
self._server = ThreadingHTTPServer(("127.0.0.1", self.port), Handler)
|
|
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
|
|
self._thread.start()
|
|
logger.info("配对服务已启动 127.0.0.1:%s", self.port)
|
|
|
|
def stop(self) -> None:
|
|
self._waiting = False
|
|
if self._server:
|
|
self._server.shutdown()
|
|
self._server.server_close()
|
|
self._server = None
|
|
self._thread = None
|