From d9e57c49007ce8e764b857022aab99ab86ee395f Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 9 Oct 2026 08:25:27 +0800 Subject: [PATCH] feat(gateway)!: drop HTTP listener, run outbound agent only - delete GatewayHTTPServer/GatewayHandler, create_server, /healthz and /v1/channel; event_channel module removed - load_config now requires CREATOR_PLATFORM_URL + CREATOR_GATEWAY_KEY - run() wires Gateway into OutboundAgent; SIGINT/SIGTERM stop the agent - deploy env example, install script, systemd unit docs and deployment docs updated to the outbound model; NATIVE_GATEWAY_ENDPOINT dead dev config removed --- browser_gateway/server/event_channel.py | 135 ----------- browser_gateway/server/http.py | 264 ++-------------------- browser_gateway/test_agent.py | 2 +- browser_gateway/test_event_channel.py | 95 -------- browser_gateway/test_event_listener.py | 12 +- browser_gateway/test_gateway.py | 154 ++++--------- browser_gateway/test_identity_binding.py | 2 +- browser_gateway/test_login_pending.py | 2 +- browser_gateway/test_private_messages.py | 13 +- browser_gateway/test_server_entry.py | 106 +++++++++ deploy/browser-gateway.env.example | 7 +- docs/deployment.md | 15 +- docs/e2e-test-plan.md | 2 +- docs/native-browser-verification.md | 8 +- scripts/dev-backend.mjs | 1 - scripts/dev-backend.test.mjs | 2 +- scripts/install-native-browser-gateway.sh | 2 +- 17 files changed, 199 insertions(+), 623 deletions(-) delete mode 100644 browser_gateway/server/event_channel.py delete mode 100644 browser_gateway/test_event_channel.py create mode 100644 browser_gateway/test_server_entry.py diff --git a/browser_gateway/server/event_channel.py b/browser_gateway/server/event_channel.py deleted file mode 100644 index 209a89c..0000000 --- a/browser_gateway/server/event_channel.py +++ /dev/null @@ -1,135 +0,0 @@ -"""One shared-port WebSocket carrying event subscriptions and acknowledgements.""" -import json -import logging -import threading -import time -from simple_websocket import Server, ConnectionClosed -from ..platform.douyin import DouyinError - -LOG = logging.getLogger('browser-gateway') - - -def serve_channel(handler): - from .http import RequestError - if handler.headers.get('Upgrade', '').lower() != 'websocket': - raise RequestError('event channel requires WebSocket upgrade', 400) - environ = {'werkzeug.socket': handler.connection} - # simple-websocket's documented socket environment adapter owns framing and handshake. - for name, value in handler.headers.items(): - environ['HTTP_' + name.upper().replace('-', '_')] = value - handler.connection.settimeout(None) - handler.close_connection = True - handler.websocket_handoff = True - ws = Server(environ, ping_interval=15, max_message_size=1024 * 1024) - handler.close_connection = True - manager = handler.server.gateway.subscriptions - gateway = handler.server.gateway - stopped = threading.Event() - lock = threading.RLock() - writing = threading.Lock() - subscriptions = {} - - def send(value): - with writing: - ws.send(json.dumps(value, ensure_ascii=False, separators=(',', ':'))) - - def close(): - try: - ws.close() - except ConnectionClosed: - LOG.debug('event WS already closed') - - def sender(): - try: - while not stopped.is_set(): - with manager.changed: - version = manager.version - with lock: - for alias, state in list(subscriptions.items()): - sub, item, sent = state - try: - current = manager._get(alias) - error = item.error or ('event subscription replaced or stopped' if current is not item or item.stopped else '') - except DouyinError as exc: - error = str(exc) - if error: - send({'type': 'error', 'subscription': sub, 'alias': alias, 'error': error}) - subscriptions.pop(alias) - continue - if sent: - continue - pending = [d for d in item.pending() if d['delivery_id'] not in sent][:100] - if pending: - started = time.monotonic() - send({'type': 'deliveries', 'subscription': sub, 'alias': alias, 'deliveries': pending}) - sent.update({d['delivery_id']: started for d in pending}) - LOG.info('event WS delivered alias=%s subscription=%s count=%s', alias, sub, len(pending)) - with manager.changed: - manager.changed.wait_for(lambda: stopped.is_set() or manager.version != version, timeout=15) - except Exception: - if not stopped.is_set(): - LOG.exception('event WS delivery failed') - try: - send({'type': 'error', 'error': 'event channel delivery failed'}) - except ConnectionClosed: - LOG.info('event WS disconnected during error reporting') - close() - stopped.set() - - thread = threading.Thread(target=sender, name='event-ws-delivery', daemon=True) - thread.start() - try: - while not stopped.is_set(): - raw = ws.receive() - if not isinstance(raw, str): - raise DouyinError('event channel requires text JSON') - value = json.loads(raw) - if not isinstance(value, dict) or value.get('type') not in ('subscribe', 'ack', 'unsubscribe'): - raise DouyinError('event channel message invalid') - alias, sub = value.get('alias'), value.get('subscription') - if not isinstance(alias, str) or not isinstance(sub, str) or not sub or len(sub) > 128: - raise DouyinError('event channel subscription invalid') - try: - # This is a metadata snapshot check, not a browser operation. - # Do not wait on the browser-operation lock in the shared receiver. - if value['type'] == 'subscribe': - gateway._require_douyin_generation(alias, value) - with lock: - if value['type'] == 'subscribe': - item = manager._get(alias) - if item.uid != value.get('uid') or item.stopped or getattr(item, 'generation', None) != value.get('listener_generation') or item.session_id != value.get('session_id'): - raise DouyinError('event channel identity changed') - subscriptions[alias] = (sub, item, {}) - send({'type': 'subscribed', 'alias': alias, 'subscription': sub}) - else: - state = subscriptions.get(alias) - if not state or state[0] != sub or manager._get(alias) is not state[1]: - raise DouyinError('stale event channel subscription') - if value['type'] == 'unsubscribe': - subscriptions.pop(alias) - else: - ids = value.get('delivery_ids') - if not isinstance(ids, list) or not ids or any(not isinstance(i, str) or i not in state[2] for i in ids): - raise DouyinError('event acknowledgement invalid') - state[1].ack(ids) - elapsed_ms = round((time.monotonic() - min(state[2][i] for i in ids)) * 1000, 3) - for delivery_id in ids: - state[2].pop(delivery_id) - LOG.info('event WS acknowledged alias=%s count=%s send_to_persist_ack_ms=%s', alias, len(ids), elapsed_ms) - except (DouyinError, RequestError) as exc: - LOG.warning('event WS subscription rejected alias=%s subscription=%s reason=%s', alias, sub, exc) - send({'type': 'error', 'subscription': sub, 'alias': alias, 'error': str(exc)}) - manager.notify() - except ConnectionClosed: - LOG.info('event WS disconnected subscriptions=%s', len(subscriptions)) - except Exception as exc: - LOG.exception('event WS request failed') - try: - send({'type': 'error', 'error': str(exc)}) - except ConnectionClosed: - LOG.info('event WS disconnected during error reporting') - finally: - stopped.set() - manager.notify() - thread.join(timeout=2) - close() diff --git a/browser_gateway/server/http.py b/browser_gateway/server/http.py index 904ad5f..43bcb4e 100644 --- a/browser_gateway/server/http.py +++ b/browser_gateway/server/http.py @@ -2,8 +2,6 @@ from __future__ import annotations -import errno -import hmac import json import logging import math @@ -17,9 +15,7 @@ from contextlib import contextmanager import time from collections.abc import Mapping from contextlib import suppress -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from typing import cast from urllib.parse import parse_qs, urlsplit from ..platform.douyin import ( @@ -35,6 +31,7 @@ from ..platform.douyin import ( is_douyin_share_url, ) from ..platform.notice_polling import SubscriptionManager, validate_checkpoint +from ..agent import AgentConfig, OutboundAgent from ..proxy import ProxyExit from ..runtime import ( NETWORK_ID_RE as _NETWORK_ID_RE, @@ -87,12 +84,10 @@ class Gateway: def __init__( self, runtimes: NativeRuntimeManager, - token: str, node_id: str, browser: DouyinBrowser | None = None, ) -> None: self.runtimes = runtimes - self.token = token self.node_id = node_id self.browser = browser or DouyinBrowser(self._browser_endpoint) self.proxies = runtimes.proxies @@ -535,56 +530,6 @@ class Gateway: raise RequestError('event listener session_id required', 400) self.subscriptions.stop(alias, input.get('listener_generation'), session_id) -class GatewayHTTPServer(ThreadingHTTPServer): - daemon_threads = True - allow_reuse_address = True - - gateway: Gateway - - def __init__(self, address, gateway: Gateway): - super().__init__(address, GatewayHandler) - self.gateway = gateway - self._connections: set[socket.socket] = set() - self._connections_lock = threading.Lock() - self._connections_changed = threading.Condition(self._connections_lock) - - def process_request(self, request, client_address): - with self._connections_changed: - self._connections.add(cast(socket.socket, request)) - try: - super().process_request(request, client_address) - except Exception: - with self._connections_changed: - self._connections.discard(cast(socket.socket, request)) - self._connections_changed.notify_all() - raise - - def process_request_thread(self, request, client_address): - try: - super().process_request_thread(request, client_address) - finally: - with self._connections_changed: - self._connections.discard(cast(socket.socket, request)) - self._connections_changed.notify_all() - - def wait_for_requests(self, timeout: float) -> None: - deadline = time.monotonic() + timeout - with self._connections_changed: - while self._connections: - remaining = deadline - time.monotonic() - if remaining <= 0: - break - self._connections_changed.wait(remaining) - if self._connections: - connections = list(self._connections) - else: - connections = [] - for connection in connections: - with suppress(OSError): - connection.shutdown(socket.SHUT_RDWR) - connection.close() - - def route_gateway_request(gateway: "Gateway", method: str, path: str, query: dict, body: dict): """统一任务路由:HTTP handler 与出站通道 task 分发共用同一套 method/path 语义。""" if method == "GET" and path == "/v1/info": @@ -641,127 +586,6 @@ def route_gateway_request(gateway: "Gateway", method: str, path: str, query: dic raise RequestError("not found", 404) -class GatewayHandler(BaseHTTPRequestHandler): - protocol_version = "HTTP/1.1" - - def setup(self) -> None: - super().setup() - self.request.settimeout(5.0) - - def do_GET(self) -> None: - self._dispatch("GET") - - def do_POST(self) -> None: - self._dispatch("POST") - - def do_DELETE(self) -> None: - self._dispatch("DELETE") - - def log_message(self, format: str, *args) -> None: - LOG.info("http_request %s", format % args) - - def _dispatch(self, method: str) -> None: - parsed = urlsplit(self.path) - if parsed.path == "/healthz": - self._respond(204, b"") - return - if not parsed.path.startswith("/v1/"): - self._respond(404, json_bytes({"error": "not found"})) - return - if not self._authorized(): - self._respond(401, json_bytes({"error": "gateway token rejected"})) - return - try: - if parsed.path == '/v1/channel' and method == 'GET': - from .event_channel import serve_channel - serve_channel(self) - return - needs_body = method in {"POST", "DELETE"} - body = self._body() if needs_body else {} - result = self._route(method, parsed.path, parse_qs(parsed.query), body) - if result is None: - self._respond(204, b"") - elif isinstance(result, tuple): - status, value = result - if type(status) is not int: - raise RuntimeError("gateway route returned an invalid status") - self._respond(status, json_bytes(value)) - else: - self._respond(200, json_bytes(result)) - except (RuntimeError, OSError, ValueError, TypeError, KeyError) as exc: - self._handle_exception(parsed.path, exc) - - def _handle_exception(self, path: str, exc: Exception) -> None: - if getattr(self, 'websocket_handoff', False): - LOG.exception('gateway WebSocket failed after handoff', extra={'path': path}) - self.close_connection = True - return - if isinstance(exc, RequestError): - payload = {"error": str(exc)} - if exc.network_id: - payload["network_id"] = exc.network_id - self._respond(exc.status, json_bytes(payload)) - elif isinstance(exc, FileNotFoundError): - self._respond(404, json_bytes({"error": str(exc)})) - elif isinstance(exc, BrowserRuntimeError): - payload = {"error": str(exc)} - if exc.network_id: - payload["network_id"] = exc.network_id - self._respond(exc.status, json_bytes(payload)) - elif isinstance(exc, ValueError): - self._respond(400, json_bytes({"error": str(exc)})) - else: - LOG.exception("gateway request failed", extra={"path": path}) - self._respond(500, json_bytes({"error": "gateway operation failed"})) - - def _route(self, method: str, path: str, query: dict, body: dict): - return route_gateway_request(self.server_as_gateway().gateway, method, path, query, body) - - def server_as_gateway(self) -> GatewayHTTPServer: - if not isinstance(self.server, GatewayHTTPServer): - raise TypeError("gateway HTTP server type is invalid") - return self.server - - def _authorized(self) -> bool: - supplied = self.headers.get("Authorization", "") - return hmac.compare_digest( - supplied, "Bearer " + self.server_as_gateway().gateway.token - ) - - def _body(self) -> dict: - length_text = self.headers.get("Content-Length") - if length_text is None: - raise RequestError("request body is required", 400) - try: - length = int(length_text) - except ValueError as exc: - raise RequestError("invalid request body length", 400) from exc - if length < 0 or length > 1 << 20: - raise RequestError("request body is too large", 400) - raw = self.rfile.read(length) - try: - value = json.loads(raw) - except json.JSONDecodeError as exc: - raise RequestError("request body must be one JSON object", 400) from exc - if not isinstance(value, dict): - raise RequestError("request body must be one JSON object", 400) - return value - - def _respond(self, status: int, body: bytes) -> None: - try: - self.send_response(status) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - if body: - self.wfile.write(body) - except BrokenPipeError: - LOG.info( - "gateway client disconnected before response", - extra={"status": status}, - ) - - def json_bytes(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode() @@ -1034,8 +858,8 @@ def _integer(env: Mapping[str, str], key: str, default: int) -> int: def load_config(env: Mapping[str, str] | None = None) -> dict: env = os.environ if env is None else env - listen = env.get("LISTEN_ADDR", "0.0.0.0:8081").strip() - token = env.get("GATEWAY_TOKEN", "").strip() + platform_url = env.get("CREATOR_PLATFORM_URL", "").strip().rstrip("/") + access_key = env.get("CREATOR_GATEWAY_KEY", "").strip() state_dir = os.path.abspath(os.path.expanduser(env.get( "BROWSER_STATE_DIR", "~/.local/state/creatorhub/browser-gateway" ).strip())) @@ -1048,9 +872,12 @@ def load_config(env: Mapping[str, str] | None = None) -> dict: ).strip())) node_id = env.get("NODE_ID", "").strip() node_name = env.get("NODE_NAME", socket.gethostname()).strip() - host, port = split_listen_address(listen) - if len(token) < 16: - raise ValueError("GATEWAY_TOKEN must be at least 16 characters") + if not platform_url.startswith(("http://", "https://")): + raise ValueError("CREATOR_PLATFORM_URL must be an http(s) platform base URL") + if platform_url.endswith("/v1/agent") or platform_url.endswith("/v1"): + raise ValueError("CREATOR_PLATFORM_URL must be the platform base URL without the agent path") + if len(access_key) < 16: + raise ValueError("CREATOR_GATEWAY_KEY must be at least 16 characters") if not state_dir or not profile_root or not os.path.isabs(state_dir) or not os.path.isabs(profile_root): raise ValueError("BROWSER_STATE_DIR and BROWSER_PROFILE_ROOT must be absolute") if not node_id: @@ -1059,8 +886,6 @@ def load_config(env: Mapping[str, str] | None = None) -> dict: raise ValueError("NODE_ID is invalid") if not node_name or has_control(node_name) or len(node_name) > 128: raise ValueError("NODE_NAME is invalid") - if not 1 <= port <= 65535: - raise ValueError("LISTEN_ADDR port must be 1..65535") external_display = _optional_positive_int(env, "RUNTIME_EXTERNAL_DISPLAY") try: available = os.path.isfile(browser_path) and os.access(browser_path, os.X_OK) @@ -1069,13 +894,13 @@ def load_config(env: Mapping[str, str] | None = None) -> dict: if not available: raise ValueError("browser executable is unavailable") return { - "listen": (host, port), + "platform_url": platform_url, + "access_key": access_key, "state_dir": state_dir, "profile_root": profile_root, "browser_path": browser_path, "node_id": node_id, "node_name": node_name, - "token": token, "cleanup_timeout": _positive_float(env, "RUNTIME_CLEANUP_TIMEOUT", 30.0, 300.0), "ready_timeout": _positive_float(env, "RUNTIME_READY_TIMEOUT", 60.0, 300.0), "min_free_bytes": _integer(env, "RUNTIME_MIN_FREE_BYTES", 20 * 1024**3), @@ -1085,25 +910,6 @@ def load_config(env: Mapping[str, str] | None = None) -> dict: } -def split_listen_address(value: str) -> tuple[str, int]: - if value.startswith(":"): - host, port_text = "", value[1:] - elif value.startswith("["): - closing = value.find("]:" ) - if closing <= 1: - raise ValueError("LISTEN_ADDR must be host:port") - host, port_text = value[1:closing], value[closing + 2:] - else: - if ":" not in value: - raise ValueError("LISTEN_ADDR must be host:port") - host, port_text = value.rsplit(":", 1) - try: - port = int(port_text) - except ValueError as exc: - raise ValueError("LISTEN_ADDR port must be an integer") from exc - return host, port - - def load_stable_node_id(state_dir: str, configured: str) -> str: if configured: return configured @@ -1129,29 +935,8 @@ def load_stable_node_id(state_dir: str, configured: str) -> str: return value -def format_listen_address(address: tuple[str, int]) -> str: - host, port = address - shown = host or "0.0.0.0" - if ":" in shown: - shown = f"[{shown}]" - return f"{shown}:{port}" - - -def create_server(config: dict, gateway: Gateway) -> GatewayHTTPServer: - try: - return GatewayHTTPServer(config["listen"], gateway) - except OSError as exc: - if exc.errno != errno.EADDRINUSE: - raise - raise SystemExit( - "browser gateway cannot listen on " - f"{format_listen_address(config['listen'])}: address already in use; " - "stop the process holding the port or change LISTEN_ADDR" - ) from exc - - -def run() -> None: - config = load_config() +def run(config: dict | None = None) -> int: + config = load_config() if config is None else config if os.geteuid() == 0: raise RuntimeError("browser gateway must run as a non-root user") logging.basicConfig(level=logging.INFO, format="%(message)s") @@ -1168,15 +953,19 @@ def run() -> None: profile_cache_max_bytes=config["profile_cache_max_bytes"], external_display=config["external_display"], ) - gateway = Gateway(runtimes, config["token"], node_id) - server = create_server(config, gateway) + gateway = Gateway(runtimes, node_id) + agent = OutboundAgent(gateway, AgentConfig( + platform_url=config["platform_url"], + access_key=config["access_key"], + )) LOG.info( json.dumps( { "service": "browser-gateway", + "mode": "outbound-channel", "node_id": node_id, "node_name": config["node_name"], - "listen_addr": f"{config['listen'][0]}:{config['listen'][1]}", + "platform_url": config["platform_url"], "browser_path": config["browser_path"], "external_display": config["external_display"], }, @@ -1189,22 +978,17 @@ def run() -> None: if shutdown_requested.is_set(): return shutdown_requested.set() - threading.Thread( - target=server.shutdown, - name="gateway-shutdown", - daemon=True, - ).start() + agent.stop() signal.signal(signal.SIGINT, request_shutdown) signal.signal(signal.SIGTERM, request_shutdown) try: - server.serve_forever() + agent.run() finally: - server.wait_for_requests(30.0) gateway.subscriptions.close() runtimes.close() - server.server_close() + return 0 if __name__ == "__main__": - run() + raise SystemExit(run()) diff --git a/browser_gateway/test_agent.py b/browser_gateway/test_agent.py index 1e97e6c..6058f6e 100644 --- a/browser_gateway/test_agent.py +++ b/browser_gateway/test_agent.py @@ -64,7 +64,7 @@ class FakePlatform: if scripted is not None: ws.send(json.dumps({"type": "result", "id": task_id, **scripted})) except Exception: - pass + pass # 连接关闭/超时即结束 handler;websockets 连接线程非 daemon,必须自然退出 def wait_for(self, predicate, timeout: float = 3.0) -> dict | None: deadline = time.monotonic() + timeout diff --git a/browser_gateway/test_event_channel.py b/browser_gateway/test_event_channel.py deleted file mode 100644 index 2b3e80b..0000000 --- a/browser_gateway/test_event_channel.py +++ /dev/null @@ -1,95 +0,0 @@ -import json -import threading -import unittest -import urllib.request -import urllib.error -from types import SimpleNamespace -from unittest.mock import Mock -import websocket -from browser_gateway.server.http import GatewayHTTPServer, GatewayHandler -from browser_gateway.platform.notice_polling import SubscriptionManager, NoticePollingSession -from browser_gateway.test_notice_polling import BOUNDARY, notice - -class EventChannelTest(unittest.TestCase): - def setUp(self): - self.manager = SubscriptionManager(Mock()) - self.sessions = {} - for alias in ('one','two'): - item = NoticePollingSession(Mock(), alias, '123', BOUNDARY, {}) - self.sessions[alias] = item - self.manager._items[alias] = item - item._changed = self.manager.notify - gateway = SimpleNamespace(token='test', subscriptions=self.manager, - _alias_lock=lambda alias:threading.RLock(), _require_douyin_generation=lambda alias,value:None, - runtimes=SimpleNamespace(close=lambda:None)) - self.server = GatewayHTTPServer(('127.0.0.1',0), gateway) - self.server.gateway = gateway - self.thread = threading.Thread(target=self.server.serve_forever,daemon=True); self.thread.start() - self.ws = websocket.create_connection('ws://127.0.0.1:%s/v1/channel'%self.server.server_port, - header=['Authorization: Bearer test'],timeout=2) - def tearDown(self): - self.ws.close();self.server.shutdown();self.server.server_close();self.thread.join(2) - def send(self,kind,alias='one',**values): - self.ws.send(json.dumps({'type':kind,'alias':alias,'subscription':alias,'uid':'123','session_id':self.sessions[alias].session_id,**values})) - def read(self):return json.loads(self.ws.recv()) - def test_shared_port_two_accounts_live_delivery_and_ack(self): - for alias in ('one','two'): - self.send('subscribe',alias);self.assertEqual(self.read()['type'],'subscribed') - self.sessions['two'].publish_notices([notice('9007199254740993123')]) - message=self.read();self.assertEqual(message['alias'],'two') - delivery=message['deliveries'][0] - self.assertEqual(delivery['notice']['event_key'],'9007199254740993123') - self.send('ack','two',delivery_ids=[delivery['delivery_id']]) - self.send('unsubscribe','one');self.send('subscribe','one');self.assertEqual(self.read()['type'],'subscribed') - self.assertEqual(self.sessions['two'].pending(),[]) - def test_unconfirmed_batch_is_replayed_after_reconnect(self): - self.send('subscribe');self.read();self.sessions['one'].publish_notices([notice('12')]);first=self.read() - self.ws.close() - self.ws=websocket.create_connection('ws://127.0.0.1:%s/v1/channel'%self.server.server_port, - header=['Authorization: Bearer test'],timeout=2) - self.send('subscribe');self.read();second=self.read() - self.assertEqual(first['deliveries'],second['deliveries']) - def test_plain_http_request_requires_upgrade(self): - request=urllib.request.Request('http://127.0.0.1:%s/v1/channel'%self.server.server_port,headers={'Authorization':'Bearer test'}) - with self.assertRaises(urllib.error.HTTPError) as failure: - urllib.request.urlopen(request,timeout=2) - self.assertEqual(failure.exception.code,400) - - def test_busy_browser_operation_does_not_delay_channel_subscription(self): - lock = threading.Lock();lock.acquire() - self.server.gateway._alias_lock = lambda alias:lock - try: - self.send('subscribe');self.assertEqual(self.read()['type'],'subscribed') - finally: - lock.release() - - def test_waiting_channel_does_not_hold_alias_lock(self): - lock = threading.Lock() - self.server.gateway._alias_lock = lambda alias:lock - self.send('subscribe');self.read() - self.assertTrue(lock.acquire(timeout=.1));lock.release() - - def test_one_stopped_account_does_not_disconnect_another(self): - self.send('subscribe','one');self.read();self.send('subscribe','two');self.read() - self.sessions['one'].stop() - self.assertEqual(self.read()['type'],'error') - self.sessions['two'].publish_notices([notice('22')]) - self.assertEqual(self.read()['alias'],'two') - - def test_old_session_is_rejected_after_same_generation_restart(self): - self.send('subscribe',session_id='previous-session') - self.assertEqual(self.read()['type'],'error') - - def test_uid_mismatch_is_rejected_without_disconnect_for_other_accounts(self): - self.send('subscribe','two');self.assertEqual(self.read()['type'],'subscribed') - self.send('subscribe',uid='999') - self.assertEqual(self.read()['type'],'error') - self.sessions['two'].publish_notices([notice('23')]) - self.assertEqual(self.read()['alias'],'two') - - def test_invalid_ack_is_reported_not_silently_removed(self): - self.send('subscribe');self.read();self.sessions['one'].publish_notices([notice('1')]);self.read() - self.send('ack',delivery_ids=['not-delivered']);self.assertEqual(self.read()['type'],'error') - self.assertEqual(len(self.sessions['one'].pending()),1) - -if __name__=='__main__':unittest.main() diff --git a/browser_gateway/test_event_listener.py b/browser_gateway/test_event_listener.py index c2285a3..43379f6 100644 --- a/browser_gateway/test_event_listener.py +++ b/browser_gateway/test_event_listener.py @@ -7,7 +7,7 @@ from browser_gateway.platform.douyin import ( normalize_notice, ) from browser_gateway.platform.notice_polling import SubscriptionManager -from browser_gateway.server.http import GatewayHandler, GatewayHTTPServer +from browser_gateway.server.http import route_gateway_request class EventListenerTests(unittest.TestCase): @@ -61,18 +61,14 @@ class EventListenerTests(unittest.TestCase): gateway.start_douyin_events.return_value = {"connected": True, "alias": "browser", "uid": "123"} gateway.poll_douyin_events.return_value = [{"kind": "open", "delivery_id": "1"}] gateway.stop_douyin_events.return_value = None - server = GatewayHTTPServer(("127.0.0.1", 0), gateway) - self.addCleanup(server.server_close) - handler = GatewayHandler.__new__(GatewayHandler) - handler.server = server route = "/v1/browsers/browser/douyin/events" payload = {"expected_uid": "123", "binding_version": 7, "runtime_id": "runtime", "network_id": "network", "network_exit_id": ""} - self.assertTrue(handler._route("POST", route, {}, payload)["connected"]) + self.assertTrue(route_gateway_request(gateway, "POST", route, {}, payload)["connected"]) query = {"ack": ["1"], "wait": ["5"], "limit": ["100"]} with self.assertRaisesRegex(Exception, 'not found'): - handler._route("GET", route, query, payload) + route_gateway_request(gateway, "GET", route, query, payload) gateway.poll_douyin_events.assert_not_called() - self.assertIsNone(handler._route("DELETE", route, {}, payload)) + self.assertIsNone(route_gateway_request(gateway, "DELETE", route, {}, payload)) if __name__ == "__main__": diff --git a/browser_gateway/test_gateway.py b/browser_gateway/test_gateway.py index 5e14f26..94e3663 100644 --- a/browser_gateway/test_gateway.py +++ b/browser_gateway/test_gateway.py @@ -51,14 +51,12 @@ from .proxy import ( gateway_module = import_module(f"{__package__}.server.http") douyin_module = import_module(f"{__package__}.platform.douyin") proxy_module = import_module(f"{__package__}.proxy") -create_server = gateway_module.create_server -format_listen_address = gateway_module.format_listen_address Gateway = gateway_module.Gateway RequestError = gateway_module.RequestError decode_generation = gateway_module.decode_generation json_bytes = gateway_module.json_bytes load_config = gateway_module.load_config -split_listen_address = gateway_module.split_listen_address +route_gateway_request = gateway_module.route_gateway_request valid_douyin_url = gateway_module.valid_douyin_url valid_douyin_source_url = gateway_module.valid_douyin_source_url validate_create = gateway_module.validate_create @@ -143,25 +141,22 @@ class GatewayValidationTests(unittest.TestCase): ) def test_config_and_urls(self) -> None: - self.assertEqual(split_listen_address(":8081"), ("", 8081)) - self.assertEqual(split_listen_address("[::1]:8081"), ("::1", 8081)) - with self.assertRaises(ValueError): - split_listen_address("missing-port") config = load_config( { - "LISTEN_ADDR": ":8081", + "CREATOR_PLATFORM_URL": "http://platform.example:8080", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", "BROWSER_PATH": "/bin/true", - "GATEWAY_TOKEN": "0123456789abcdef", } ) - self.assertEqual(config["listen"], ("", 8081)) + self.assertEqual(config["platform_url"], "http://platform.example:8080") self.assertEqual(config["browser_path"], "/bin/true") self.assertEqual(config["ready_timeout"], 60.0) self.assertIsNone(config["external_display"]) external = load_config( { + "CREATOR_PLATFORM_URL": "http://platform.example:8080", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", "BROWSER_PATH": "/bin/true", - "GATEWAY_TOKEN": "0123456789abcdef", "RUNTIME_EXTERNAL_DISPLAY": "99", } ) @@ -169,7 +164,8 @@ class GatewayValidationTests(unittest.TestCase): with self.assertRaises(ValueError): load_config( { - "GATEWAY_TOKEN": "0123456789abcdef", + "CREATOR_PLATFORM_URL": "http://platform.example:8080", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", "BROWSER_PATH": "/not/a/browser", } ) @@ -221,35 +217,7 @@ class GatewayValidationTests(unittest.TestCase): ) - def test_create_server_reports_conflicting_listen_address(self) -> None: - gateway = Gateway.__new__(Gateway) - with patch.object( - gateway_module, - "GatewayHTTPServer", - side_effect=OSError(errno.EADDRINUSE, "Address already in use"), - ): - with self.assertRaises(SystemExit) as ctx: - create_server({"listen": ("", 8081)}, gateway) - message = str(ctx.exception) - self.assertIn("0.0.0.0:8081", message) - self.assertIn("address already in use", message) - self.assertIn("LISTEN_ADDR", message) - with patch.object( - gateway_module, - "GatewayHTTPServer", - side_effect=OSError(errno.EACCES, "Permission denied"), - ): - with self.assertRaises(OSError) as denied: - create_server({"listen": ("127.0.0.1", 8081)}, gateway) - self.assertEqual(denied.exception.errno, errno.EACCES) - - def test_format_listen_address(self) -> None: - self.assertEqual(format_listen_address(("", 8081)), "0.0.0.0:8081") - self.assertEqual(format_listen_address(("127.0.0.1", 28187)), "127.0.0.1:28187") - self.assertEqual(format_listen_address(("::1", 8081)), "[::1]:8081") - def test_http_routes_and_body_validation(self) -> None: - handler = gateway_module.GatewayHandler.__new__(gateway_module.GatewayHandler) gateway = Mock() gateway.list_browsers.return_value = [] gateway.douyin_identity.return_value = {"uid": "123"} @@ -261,16 +229,14 @@ class GatewayValidationTests(unittest.TestCase): gateway.douyin_action.return_value = {"status": "succeeded"} gateway.douyin_message_history.return_value = {"status": "succeeded"} gateway.poll_douyin_events.return_value = [] - server = Mock() - server.gateway = gateway - cast(Any, handler).server = server - cast(Any, handler).server_as_gateway = lambda: server - self.assertEqual(handler._route("GET", "/v1/browsers", {}, {}), []) + route = route_gateway_request + self.assertEqual(route(gateway, "GET", "/v1/browsers", {}, {}), []) self.assertEqual( - handler._route("POST", "/v1/browsers", {}, {}), + route(gateway, "POST", "/v1/browsers", {}, {}), (201, gateway.create.return_value), ) - handler._route( + route( + gateway, "DELETE", "/v1/browsers/safe", {}, @@ -279,62 +245,36 @@ class GatewayValidationTests(unittest.TestCase): gateway.remove.assert_called_once_with( "safe", {"binding_version": 1, "runtime_id": "a" * 64, "network_id": ""} ) - handler._route("POST", "/v1/browsers/safe/start", {}, {}) - handler._route("POST", "/v1/browsers/safe/stop", {}, {}) - handler._route("POST", "/v1/browsers/safe/proxy", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/start", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/stop", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/proxy", {}, {}) with self.assertRaises(RequestError): - handler._route("POST", "/v1/browsers/safe/douyin/cookies", {}, {}) - handler._route("POST", "/v1/browsers/safe/douyin/get", {}, {}) - handler._route("POST", "/v1/browsers/safe/douyin/identity", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/douyin/cookies", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/douyin/get", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/douyin/identity", {}, {}) self.assertEqual( - handler._route("POST", "/v1/browsers/safe/douyin/login-qr", {}, {}), + route(gateway, "POST", "/v1/browsers/safe/douyin/login-qr", {}, {}), gateway.douyin_login_qr.return_value, ) gateway.douyin_login_qr.assert_called_once_with("safe", {}) - handler._route("POST", "/v1/browsers/safe/douyin/action", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/douyin/action", {}, {}) self.assertEqual( - handler._route("POST", "/v1/browsers/safe/douyin/messages", {}, {}), + route(gateway, "POST", "/v1/browsers/safe/douyin/messages", {}, {}), {"status": "succeeded"}, ) gateway.douyin_message_history.assert_called_once_with("safe", {}) with self.assertRaises(RequestError): - handler._route("GET", "/v1/browsers/safe/douyin/events", {}, {}) - handler._route("POST", "/v1/browsers/safe/douyin/events", {}, {}) - handler._route("DELETE", "/v1/browsers/safe/douyin/events", {}, {}) + route(gateway, "GET", "/v1/browsers/safe/douyin/events", {}, {}) + route(gateway, "POST", "/v1/browsers/safe/douyin/events", {}, {}) + route(gateway, "DELETE", "/v1/browsers/safe/douyin/events", {}, {}) with self.assertRaises(RequestError): - handler._route("GET", "/v1/unknown", {}, {}) - cast(Any, handler).headers = {"Content-Length": "7"} - cast(Any, handler).rfile = io.BytesIO(b'{"x":1}') - self.assertEqual(handler._body(), {"x": 1}) - cast(Any, handler).headers = {} - with self.assertRaises(RequestError): - handler._body() - test_token = "x" * 16 - cast(Any, handler).headers = {"Authorization": f"Bearer {test_token}"} - server.gateway.token = test_token - self.assertTrue(handler._authorized()) - handler._respond = Mock() - handler._handle_exception("/v1", RequestError("bad", 400)) - handler._handle_exception("/v1", ValueError("bad")) - self.assertEqual(handler._respond.call_count, 2) + route(gateway, "GET", "/v1/unknown", {}, {}) - def test_removed_get_event_route_does_not_read_generation_body(self) -> None: - handler = gateway_module.GatewayHandler.__new__(gateway_module.GatewayHandler) - cast(Any, handler).path = "/v1/browsers/safe/douyin/events?wait=1" - cast(Any, handler).headers = {"Content-Length": "67"} - cast(Any, handler).rfile = io.BytesIO( - b'{"binding_version":1,"runtime_id":"runtime","network_id":"network"}' - ) - cast(Any, handler)._authorized = lambda: True - cast(Any, handler)._route = Mock(return_value=[]) - cast(Any, handler)._respond = Mock() - handler._dispatch("GET") - cast(Any, handler)._route.assert_called_once_with( - "GET", - "/v1/browsers/safe/douyin/events", - {"wait": ["1"]}, - {}, - ) + def test_removed_get_event_route_is_not_dispatchable(self) -> None: + gateway = Mock() + with self.assertRaises(RequestError): + route_gateway_request(gateway, "GET", "/v1/browsers/safe/douyin/events", {}, {}) + gateway.poll_douyin_events.assert_not_called() def test_validation_boundaries(self) -> None: self.assertEqual(proxy_port("http://127.0.0.1:1234"), 1234) @@ -378,12 +318,13 @@ class GatewayValidationTests(unittest.TestCase): with self.assertRaises(RequestError): validate_create({"alias": "safe", "unknown": True}) with self.assertRaises(ValueError): - load_config({"GATEWAY_TOKEN": "short"}) + load_config({"CREATOR_GATEWAY_KEY": "short"}) with self.assertRaises(ValueError): load_config( { + "CREATOR_PLATFORM_URL": "http://platform.example:8080", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", "BROWSER_PATH": "/bin/true", - "GATEWAY_TOKEN": "0123456789abcdef", "RUNTIME_EXTERNAL_DISPLAY": "0", } ) @@ -436,7 +377,6 @@ class GatewayBusinessMethodTests(unittest.TestCase): self.browser.action_ownership.return_value = None self.gateway = Gateway( self.runtimes, - "gateway-token-123456", "node-a", browser=self.browser, ) @@ -547,17 +487,10 @@ class GatewayBusinessMethodTests(unittest.TestCase): self.gateway.stop_douyin_events("safe", {**self.generation, 'session_id': 'test-session'}) def test_http_info_route_and_action_error_ownership(self) -> None: - handler = gateway_module.GatewayHandler.__new__(gateway_module.GatewayHandler) - server = Mock() - server.gateway = self.gateway - cast(Any, handler).server = server - cast(Any, handler).server_as_gateway = lambda: server - self.assertEqual(handler._route("GET", "/v1/info", {}, {}), self.gateway.info()) - cast(Any, handler).path = "/v1/info" - cast(Any, handler).headers = {"Authorization": "Bearer gateway-token-123456"} - cast(Any, handler)._respond = Mock() - handler._dispatch("GET") - self.assertEqual(handler._respond.call_args.args[0], 200) + self.assertEqual( + route_gateway_request(self.gateway, "GET", "/v1/info", {}, {}), + self.gateway.info(), + ) self.gateway._handle_douyin_action_error("safe", "follow", DouyinError("timed out"), "operation-1") self.assertEqual(self.gateway._uncertain_actions["safe"], float("inf")) self.gateway._release_action_ownership("safe", "operation-1") @@ -1205,17 +1138,6 @@ class ChunkSocket: class GatewayLifecycleTests(unittest.TestCase): - def test_server_tracks_daemon_request_threads_and_timeout(self) -> None: - server = gateway_module.GatewayHTTPServer(("127.0.0.1", 0), Mock()) - try: - self.assertTrue(server.daemon_threads) - self.assertEqual(gateway_module.GatewayHandler.protocol_version, "HTTP/1.1") - finally: - server.server_close() - - - - def _input(self, stopped: bool = True) -> dict: return { "alias": "safe", diff --git a/browser_gateway/test_identity_binding.py b/browser_gateway/test_identity_binding.py index e1f4385..27e2a08 100644 --- a/browser_gateway/test_identity_binding.py +++ b/browser_gateway/test_identity_binding.py @@ -10,7 +10,7 @@ class IdentityBindingTests(unittest.TestCase): def setUp(self): self.browser = Mock() self.browser.identity.return_value = {"uid": "123456789", "sec_uid": "MS4wLjAB-test", "unique_id": "douyin-name"} - self.gateway = Gateway(Mock(), "token", "node", browser=self.browser) + self.gateway = Gateway(Mock(), "node", browser=self.browser) self.generation = {"binding_version": 1, "runtime_id": "a" * 64, "network_id": "native-" + "b" * 32} self.lock = patch.object(self.gateway, "_alias_lock", return_value=nullcontext()) self.check = patch.object(self.gateway, "_require_douyin_generation") diff --git a/browser_gateway/test_login_pending.py b/browser_gateway/test_login_pending.py index 6eed92e..251fe62 100644 --- a/browser_gateway/test_login_pending.py +++ b/browser_gateway/test_login_pending.py @@ -36,7 +36,7 @@ class LoginPendingTests(unittest.TestCase): runtimes.alias_lock.return_value = nullcontext() runtimes.require_generation.return_value = SimpleNamespace(runtime_id='a' * 64, network_exit_id='exit-1') browser = Mock() - gateway = Gateway(runtimes, 'gateway-token-123456', 'node-a', browser=browser) + gateway = Gateway(runtimes, 'node-a', browser=browser) payload = {'runtime_id': 'a' * 64, 'binding_version': 1, 'network_id': 'native-' + 'b' * 32, 'network_exit_id': 'exit-1', 'expected_account_key': '123'} return gateway, browser, payload diff --git a/browser_gateway/test_private_messages.py b/browser_gateway/test_private_messages.py index 1245baa..de60a60 100644 --- a/browser_gateway/test_private_messages.py +++ b/browser_gateway/test_private_messages.py @@ -6,7 +6,7 @@ from contextlib import nullcontext, contextmanager from unittest.mock import Mock, patch from .platform.douyin import DouyinBrowser, DouyinError, inbox_expression, im_expression, private_identity_expression -from .server.http import Gateway, GatewayHandler, RequestError +from .server.http import Gateway, RequestError, route_gateway_request class PrivateMessageTests(unittest.TestCase): @@ -55,7 +55,7 @@ assert.equal((await vm.runInNewContext(SOURCE,context)).ready,true);assert.equal try:yield finally:active.release() runtimes=Mock();runtimes.alias_lock.side_effect=native_lock - gateway=Gateway(runtimes,'token','node',Mock()) + gateway=Gateway(runtimes,'node',Mock()) def first(): try: with gateway._alias_lock('account'):entered.set();release.wait(3) @@ -70,9 +70,9 @@ assert.equal((await vm.runInNewContext(SOURCE,context)).ready,true);assert.equal self.assertTrue(second.is_set());self.assertEqual(errors,[]) def test_extra_private_login_routes_are_removed(self): - handler=object.__new__(GatewayHandler);gateway=Mock();handler.server_as_gateway=Mock(return_value=Mock(gateway=gateway)) + gateway=Mock() for action in ['private-login-qr','private-login-verify']: - with self.assertRaises(RequestError):handler._route('POST','/v1/browsers/account/douyin/'+action,{}, {'expected_uid':'123'}) + with self.assertRaises(RequestError):route_gateway_request(gateway,'POST','/v1/browsers/account/douyin/'+action,{}, {'expected_uid':'123'}) def test_listener_restart_validates_checkpoint_before_touching_browser(self): gateway = object.__new__(Gateway) @@ -82,12 +82,9 @@ assert.equal((await vm.runInNewContext(SOURCE,context)).ready,true);assert.equal gateway.subscriptions.start.assert_not_called() def test_inbox_route_is_dispatched(self): - handler = object.__new__(GatewayHandler) gateway = Mock() - server = Mock(gateway=gateway) - handler.server_as_gateway = Mock(return_value=server) body = {'expected_uid':'123'} - handler._route('POST', '/v1/browsers/account/douyin/inbox', {}, body) + route_gateway_request(gateway, 'POST', '/v1/browsers/account/douyin/inbox', {}, body) gateway.douyin_inbox.assert_called_once_with('account',body) def test_inbox_script_uses_identity_and_server_messages(self): diff --git a/browser_gateway/test_server_entry.py b/browser_gateway/test_server_entry.py new file mode 100644 index 0000000..aecc0e4 --- /dev/null +++ b/browser_gateway/test_server_entry.py @@ -0,0 +1,106 @@ +"""Server entry tests: outbound-only config and run() wiring.""" +from __future__ import annotations + +import unittest +from importlib import import_module +from unittest.mock import Mock, patch + +gateway_module = import_module(f"{__package__}.server.http") + +load_config = gateway_module.load_config + + +class OutboundConfigTests(unittest.TestCase): + def test_platform_url_and_key_are_required(self) -> None: + config = load_config({ + "CREATOR_PLATFORM_URL": "http://platform.example:8080", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", + "BROWSER_PATH": "/bin/true", + }) + self.assertEqual(config["platform_url"], "http://platform.example:8080") + self.assertEqual(config["access_key"], "0123456789abcdef") + self.assertNotIn("listen", config) + self.assertNotIn("token", config) + + def test_platform_url_must_be_http(self) -> None: + with self.assertRaises(ValueError): + load_config({ + "CREATOR_PLATFORM_URL": "ftp://platform.example", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", + "BROWSER_PATH": "/bin/true", + }) + + def test_platform_url_must_not_embed_agent_path(self) -> None: + with self.assertRaises(ValueError): + load_config({ + "CREATOR_PLATFORM_URL": "http://platform.example:8080/v1/agent", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", + "BROWSER_PATH": "/bin/true", + }) + + def test_access_key_minimum_length(self) -> None: + with self.assertRaises(ValueError): + load_config({ + "CREATOR_PLATFORM_URL": "http://platform.example:8080", + "CREATOR_GATEWAY_KEY": "short", + "BROWSER_PATH": "/bin/true", + }) + + def test_trailing_slash_platform_url_is_normalized(self) -> None: + config = load_config({ + "CREATOR_PLATFORM_URL": "http://platform.example:8080/", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", + "BROWSER_PATH": "/bin/true", + }) + self.assertEqual(config["platform_url"], "http://platform.example:8080") + + +class RunWiringTests(unittest.TestCase): + def test_run_builds_agent_from_config_and_stops_on_signal(self) -> None: + config = load_config({ + "CREATOR_PLATFORM_URL": "http://platform.example:8080", + "CREATOR_GATEWAY_KEY": "0123456789abcdef", + "BROWSER_PATH": "/bin/true", + }) + agent_instances = [] + + class FakeAgent: + def __init__(self, gateway, agent_config) -> None: + self.gateway = gateway + self.config = agent_config + self.ran = False + self.stopped = False + agent_instances.append(self) + + def run(self) -> None: + self.ran = True + + def stop(self) -> None: + self.stopped = True + + registered = {} + + def record_signal(signum, handler): + registered[signum] = handler + + with patch.object(gateway_module, "OutboundAgent", FakeAgent), \ + patch.object(gateway_module, "NativeRuntimeManager", Mock()), \ + patch.object(gateway_module.signal, "signal", record_signal): + exit_code = gateway_module.run(config) + self.assertEqual(exit_code, 0) + self.assertEqual(len(agent_instances), 1) + self.assertEqual(agent_instances[0].config.platform_url, config["platform_url"]) + self.assertEqual(agent_instances[0].config.access_key, config["access_key"]) + self.assertTrue(agent_instances[0].ran) + # SIGINT/SIGTERM 都注册了停止回调,触发后 agent 收到 stop + self.assertIn(gateway_module.signal.SIGINT, registered) + self.assertIn(gateway_module.signal.SIGTERM, registered) + registered[gateway_module.signal.SIGTERM](15, None) + registered[gateway_module.signal.SIGINT](2, None) + self.assertTrue(agent_instances[0].stopped) + # Gateway 的业务方法必须仍挂在 agent 上(agent.gateway) + self.assertIsNotNone(agent_instances[0].gateway.subscriptions) + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/browser-gateway.env.example b/deploy/browser-gateway.env.example index 471efc1..932b235 100644 --- a/deploy/browser-gateway.env.example +++ b/deploy/browser-gateway.env.example @@ -1,6 +1,7 @@ -# Copy to ~/.config/creatorhub/browser-gateway.env and replace the token. -LISTEN_ADDR=0.0.0.0:8081 -GATEWAY_TOKEN=replace-with-at-least-16-random-characters +# Copy to ~/.config/creatorhub/browser-gateway.env and replace the key. +# 平台地址与网关 key 在控制面「网关管理」页生成;网关主动出站连接平台。 +CREATOR_PLATFORM_URL=http://platform.example:8080 +CREATOR_GATEWAY_KEY=replace-with-at-least-16-random-characters BROWSER_STATE_DIR=~/.local/state/creatorhub/browser-gateway BROWSER_PROFILE_ROOT=~/.local/share/creatorhub/browser-profiles BROWSER_PATH=~/.local/share/creatorhub/browsers/fingerprint-chromium/148.0.7778.215/chrome diff --git a/docs/deployment.md b/docs/deployment.md index 6e2f4c0..842cb22 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -47,8 +47,8 @@ $EDITOR ~/.config/creatorhub/browser-gateway.env 关键配置: -- `LISTEN_ADDR=0.0.0.0:8081`:允许局域网 control-plane 联调;防火墙只放行受信网络; -- `GATEWAY_TOKEN`:至少 16 个字符,和控制面登记值完全一致; +- `CREATOR_PLATFORM_URL`:control-plane 的 http(s) 根地址(不含 `/v1/agent` 路径);网关主动出站建立 WebSocket 连接,无需公网入站端口; +- `CREATOR_GATEWAY_KEY`:至少 16 个字符,与控制面「网关管理」为该网关签发的 key 完全一致; - gateway 宿主机必须预安装并自检浏览器运行时;control-plane 不提供浏览器版本登记、路径配置或升级入口,相关运行时参数只由 gateway 本机服务读取; - `BROWSER_STATE_DIR`:runtime 清单、锁和日志目录; - `BROWSER_PROFILE_ROOT`:持久账号 Profile 根目录;不得指向临时目录; @@ -58,9 +58,10 @@ $EDITOR ~/.config/creatorhub/browser-gateway.env ```bash scripts/install-native-browser-gateway.sh -curl --fail --silent --show-error http://127.0.0.1:8081/healthz -curl --fail --silent --show-error -H "Authorization: Bearer ${GATEWAY_TOKEN}" http://127.0.0.1:8081/v1/info systemctl --user --no-pager status creatorhub-browser-gateway.service +journalctl --user -u creatorhub-browser-gateway.service -n 50 --no-pager + +网关启动后主动连入 `CREATOR_PLATFORM_URL` 的 `/v1/agent` WebSocket;连接成功时控制面「网关管理」列表该网关显示在线,不再提供本地 `/healthz` 或入站 HTTP 接口。 ``` 若 user service 不随登录启动,按主机运行规范启用 lingering;不要把 gateway 改成 root service: @@ -89,7 +90,7 @@ docker compose ps docker compose logs --tail=200 creator-hub postgres ``` -Compose 中的 control-plane 访问宿主机 gateway 时,网关 Endpoint 登记为 `http://host.docker.internal:8081`。裸机运行 control-plane 时登记为 `http://127.0.0.1:8081`。两种情况下都必须使用同一个 `GATEWAY_TOKEN`。 +网关不再登记 Endpoint;在控制面「网关管理」创建网关条目,将生成的 access key 配置到网关的 `CREATOR_GATEWAY_KEY`。平台与网关的全部通信复用网关发起的那条 WebSocket 连接。 ### 裸机开发 @@ -158,8 +159,8 @@ docker compose config --quiet systemctl --user --no-pager --full status creatorhub-browser-gateway.service journalctl --user -u creatorhub-browser-gateway.service --since=-30m find ~/.local/state/creatorhub/browser-gateway -maxdepth 3 -type f -name runtime.json -print -curl --fail --silent --show-error http://127.0.0.1:8081/v1/browsers \ - -H "Authorization: Bearer ${GATEWAY_TOKEN}" + +网关本地无 HTTP 接口;浏览器列表与任务执行记录在控制面「网关管理」页查看。 ``` 强杀或网络中断后,不要重复发送真实平台写操作。先查询 control-plane 的运行、结果未知和 cleanup 状态,再按 generation 执行显式恢复或清理。 diff --git a/docs/e2e-test-plan.md b/docs/e2e-test-plan.md index 2984bc7..8d2bc84 100644 --- a/docs/e2e-test-plan.md +++ b/docs/e2e-test-plan.md @@ -33,7 +33,7 @@ 1. 使用专用、可销毁的 Linux 主机;gateway 以非 root `systemctl --user` 服务运行。禁止与现有业务共用固定 browser alias、Profile 目录或数据库 schema;不得删除现有 Docker PostgreSQL 之外的业务服务。 2. 由执行负责人按 `docs/deployment.md` 准备控制面/PostgreSQL Compose 与 host-native gateway,记录构建提交、gateway 运行时配置、systemd/Xvfb/操作系统、数据库版本、时区、测试起止 UTC;另用独立时段验证 `compose.dev.yaml` 的开发入口。不得一边测试一边自动升级源码。 3. 准备 `curl`、`jq`、必要时 Docker CLI(仅用于 PostgreSQL Compose)、`systemctl --user`、浏览器 DevTools;媒体核验使用宿主机 `ffprobe`/`ffmpeg`。gateway 宿主机必须有可执行的默认浏览器配置;缺失配置由 gateway 明确拒绝。 -4. 配置必须来自测试凭据保管渠道:`CONTROL_PLANE_USERNAME`、`CONTROL_PLANE_PASSWORD`、32 字节 Base64 `CREATORHUB_CREDENTIAL_MASTER_KEY`、`GATEWAY_TOKEN`。gateway 还需合法 `BROWSER_STATE_DIR`、`BROWSER_PROFILE_ROOT` 和 `NODE_ID`。不得将实际密码、Cookie、token、证件号贴入文档/缺陷/日志截图。 +4. 配置必须来自测试凭据保管渠道:`CONTROL_PLANE_USERNAME`、`CONTROL_PLANE_PASSWORD`、32 字节 Base64 `CREATORHUB_CREDENTIAL_MASTER_KEY`、`CREATOR_GATEWAY_KEY`(控制面网关管理签发)。gateway 还需合法 `CREATOR_PLATFORM_URL`、`BROWSER_STATE_DIR`、`BROWSER_PROFILE_ROOT` 和 `NODE_ID`。不得将实际密码、Cookie、token、证件号贴入文档/缺陷/日志截图。 5. 记录控制面 `LISTEN_ADDR`、`WEB_DIR`、`DATABASE_URL`、`CREATORHUB_CREDENTIAL_STORE_DIR`、`LOG_LEVEL`;gateway 的 `LISTEN_ADDR`、`BROWSER_STATE_DIR`、`BROWSER_PROFILE_ROOT`、`NODE_ID` 和宿主机运行时配置。媒体目录 `CREATOR_MEDIA_DIR` 可写;如测转写,`CREATOR_TRANSCRIPTION_BIN` 必须是已安装的真实供应商适配程序,接受音轨路径参数、stdout 输出正文;不能用 `echo` 冒充真实转写。 6. 如测 AI:准备已审批的 Bailian 账号、配额、`BAILIAN_API_KEY`,可选 `BAILIAN_BASE_URL`;页面 provider 填 `bailian`,model 填实际批准模型。勾选“已完成审批”并不验证密钥。转写页面元数据也不替代可执行程序配置。 7. 对所有代理协议 http/https/socks4/socks5 分别准备真实可控测试端点;至少一个无认证端点、一个带认证端点、两个不同公网 IP、一个不可达端点,保存实际地区与供应商定义。没有某协议服务时该行阻塞,不能用另一协议代替。 diff --git a/docs/native-browser-verification.md b/docs/native-browser-verification.md index 5643b3a..bb7b321 100644 --- a/docs/native-browser-verification.md +++ b/docs/native-browser-verification.md @@ -88,14 +88,14 @@ docker compose config --quiet - Go 使用 `LISTEN_ADDR`;联调须设置 `0.0.0.0:8082`。 - Vite 当前 `/api` 代理到 `127.0.0.1:8082`,端口 5100。Vite 必须加 `--host 0.0.0.0`,不能只给 localhost 地址。 -- native Python gateway 使用 `LISTEN_ADDR=0.0.0.0:8081`,由非 root systemd user service 运行;稳定 `node_id` 从配置或主机身份加载。 -- `scripts/dev-backend.mjs` 只启动 PostgreSQL 并检查 `NATIVE_GATEWAY_ENDPOINT`,不会启动浏览器容器。 +- native Python gateway 由非 root systemd user service 运行,通过 `CREATOR_PLATFORM_URL` 与 `CREATOR_GATEWAY_KEY` 主动出站连接控制面 `/v1/agent`;稳定 `node_id` 从配置或主机身份加载。 +- `scripts/dev-backend.mjs` 只启动 PostgreSQL 与热重载控制面,不会启动浏览器容器。 ### 单节点执行步骤 1. 按部署说明在单节点安装已确认浏览器、Xvfb 和 `creatorhub-browser-gateway.service`。 2. 使用已有、经验证的 PostgreSQL 与凭据配置。不要重新生成现有主密钥;不要把秘密值粘贴到验收报告。 -3. 按交付脚本启动单节点 native gateway;在控制面登记真实地址,并核对 `/v1/info` 返回的稳定节点身份。 +3. 按交付脚本启动单节点 native gateway;在控制面「网关管理」创建条目并把签发的 access key 配置到网关,核对列表在线状态与节点版本。 4. 启动控制面及前端: ```bash @@ -105,7 +105,7 @@ LISTEN_ADDR=0.0.0.0:8082 go run ./cmd/control-plane npm --prefix web run dev -- --host 0.0.0.0 ``` -命令启动后,用 `ip -brief -4 addr` 记录当次实际局域网 IP。用户在另一台设备访问 `http://:5100`;开发者可查看 `http://:8082/healthz`。报告列出单节点 gateway 的实际 Endpoint。 +命令启动后,用 `ip -brief -4 addr` 记录当次实际局域网 IP。用户在另一台设备访问 `http://:5100`;开发者可查看 `http://:8082/healthz`。报告记录单节点网关的在线状态与版本。 历史观察到的地址不构成可用性证据;实施当天必须重新核实 IP、监听地址和防火墙。 diff --git a/scripts/dev-backend.mjs b/scripts/dev-backend.mjs index d7004e9..4c3b0c2 100755 --- a/scripts/dev-backend.mjs +++ b/scripts/dev-backend.mjs @@ -38,7 +38,6 @@ if (!overrides.WEB_DIR) overrides.WEB_DIR = path.join(root, "web", "dist"); if (!overrides.CREATOR_COVER_DIR) overrides.CREATOR_COVER_DIR = path.join(root, ".data", "covers"); if (!overrides.CREATORHUB_CREDENTIAL_STORE_DIR) overrides.CREATORHUB_CREDENTIAL_STORE_DIR = path.join(root, ".dev-credentials"); if (!overrides.LOG_LEVEL) overrides.LOG_LEVEL = "debug"; -if (!overrides.NATIVE_GATEWAY_ENDPOINT) overrides.NATIVE_GATEWAY_ENDPOINT = "http://127.0.0.1:28187"; const env = { ...process.env, ...overrides }; const lockDir = path.join(root, ".data"); diff --git a/scripts/dev-backend.test.mjs b/scripts/dev-backend.test.mjs index dc0f145..d233895 100644 --- a/scripts/dev-backend.test.mjs +++ b/scripts/dev-backend.test.mjs @@ -103,7 +103,7 @@ test('reports backend startup failure and allows a retry', { timeout: 5000 }, as test('preserves environment configuration when launching through the lock', { timeout: 5000 }, async (t) => { const start = fixture(t, "console.log(process.env.LISTEN_ADDR); console.log(process.env.CREATORHUB_CREDENTIAL_MASTER_KEY);", { - envFile: `\n# ignored comment\ninvalid line\nLISTEN_ADDR=:19091\nLISTEN_ADDR=:19092\nCREATORHUB_CREDENTIAL_MASTER_KEY=${Buffer.alloc(32, 1).toString('base64')}\nCONTROL_PLANE_USERNAME=test\nCONTROL_PLANE_PASSWORD=test\nDATABASE_URL=postgres://test\nWEB_DIR=/tmp/web\nCREATOR_COVER_DIR=/tmp/covers\nCREATORHUB_CREDENTIAL_STORE_DIR=/tmp/credentials\nLOG_LEVEL=info\nNATIVE_GATEWAY_ENDPOINT=http://127.0.0.1:28187\n`, + envFile: `\n# ignored comment\ninvalid line\nLISTEN_ADDR=:19091\nLISTEN_ADDR=:19092\nCREATORHUB_CREDENTIAL_MASTER_KEY=${Buffer.alloc(32, 1).toString('base64')}\nCONTROL_PLANE_USERNAME=test\nCONTROL_PLANE_PASSWORD=test\nDATABASE_URL=postgres://test\nWEB_DIR=/tmp/web\nCREATOR_COVER_DIR=/tmp/covers\nCREATORHUB_CREDENTIAL_STORE_DIR=/tmp/credentials\nLOG_LEVEL=info\n`, }); const child = start(); assert.equal((await exited(child))[0], 0); diff --git a/scripts/install-native-browser-gateway.sh b/scripts/install-native-browser-gateway.sh index 7c2c076..6e00228 100755 --- a/scripts/install-native-browser-gateway.sh +++ b/scripts/install-native-browser-gateway.sh @@ -12,7 +12,7 @@ unit_dir="${HOME}/.config/systemd/user" unit_file="${unit_dir}/creatorhub-browser-gateway.service" if [[ ! -f "${env_file}" ]]; then - echo "missing ${env_file}; copy deploy/browser-gateway.env.example and set GATEWAY_TOKEN and BROWSER_PATH" >&2 + echo "missing ${env_file}; copy deploy/browser-gateway.env.example and set CREATOR_PLATFORM_URL, CREATOR_GATEWAY_KEY and BROWSER_PATH" >&2 exit 1 fi