1418 lines
53 KiB
Python
1418 lines
53 KiB
Python
"""CreatorHub native browser gateway."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import re
|
|
import secrets
|
|
import signal
|
|
import socket
|
|
import threading
|
|
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 (
|
|
ACCOUNT_KEY_RE,
|
|
ACTIONS,
|
|
COMMENTS_PATH,
|
|
IDENTITY_URL,
|
|
UID_RE,
|
|
WORKS_PATH,
|
|
DouyinBrowser,
|
|
DouyinError,
|
|
SubscriptionManager,
|
|
is_douyin_content_url,
|
|
is_douyin_share_url,
|
|
)
|
|
from ..platform.xiaohongshu import XiaohongshuBrowser, is_xiaohongshu_share_url
|
|
from ..proxy import ProxyExit
|
|
from ..runtime import (
|
|
BROWSER_VERSION_RE,
|
|
BrowserRuntimeError,
|
|
NativeRuntimeManager,
|
|
has_control,
|
|
validate_runtime_input,
|
|
)
|
|
from ..runtime import (
|
|
NETWORK_ID_RE as _NETWORK_ID_RE,
|
|
)
|
|
from ..runtime import (
|
|
RUNTIME_CLEANUP_SENTINEL as _RUNTIME_CLEANUP_SENTINEL,
|
|
)
|
|
from ..runtime import (
|
|
RUNTIME_ID_RE as _RUNTIME_ID_RE,
|
|
)
|
|
from ..runtime import (
|
|
parse_proxy_exit as _parse_proxy_exit,
|
|
)
|
|
from ..runtime import (
|
|
validate_proxy_exit as _validate_proxy_exit,
|
|
)
|
|
|
|
LOG = logging.getLogger("creatorhub.gateway")
|
|
RUNTIME_ID_RE = _RUNTIME_ID_RE
|
|
NETWORK_ID_RE = _NETWORK_ID_RE
|
|
RUNTIME_CLEANUP_SENTINEL = _RUNTIME_CLEANUP_SENTINEL
|
|
EXIT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$")
|
|
DOUYIN_ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,127}$")
|
|
DOUYIN_ORIGIN = "https://www.douyin.com"
|
|
DOUYIN_IDENTITY_PATH = "/aweme/v1/web/user/profile/self/"
|
|
DOUYIN_PROFILE_OTHER_PATH = "/aweme/v1/web/user/profile/other/"
|
|
DOUYIN_IDENTITY_URL = IDENTITY_URL
|
|
DOUYIN_WORKS_PATH = WORKS_PATH
|
|
DOUYIN_WORK_DETAIL_PATH = "/aweme/v1/web/aweme/detail/"
|
|
DOUYIN_COMMENTS_PATH = COMMENTS_PATH
|
|
XHS_ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$")
|
|
XHS_IDENTITY_PATH = "/api/sns/web/v2/user/me"
|
|
XHS_USER_POSTED_PATH = "/api/sns/web/v1/user_posted"
|
|
XHS_COMMENTS_PATH = "/api/sns/web/v2/comment/page"
|
|
XHS_SEARCH_PATH = "/api/sns/web/v2/search/notes"
|
|
XHS_FEED_PATH = "/api/sns/web/v1/feed"
|
|
|
|
|
|
def _noop() -> None:
|
|
return None
|
|
|
|
|
|
class RequestError(BrowserRuntimeError):
|
|
pass
|
|
|
|
|
|
class Gateway:
|
|
def __init__(
|
|
self,
|
|
runtimes: NativeRuntimeManager,
|
|
token: str,
|
|
node_id: str,
|
|
browser: DouyinBrowser | None = None,
|
|
xiaohongshu_browser: XiaohongshuBrowser | None = None,
|
|
) -> None:
|
|
self.runtimes = runtimes
|
|
self.token = token
|
|
self.node_id = node_id
|
|
self.browser = browser or DouyinBrowser(self._browser_endpoint)
|
|
self.xiaohongshu_browser = xiaohongshu_browser or XiaohongshuBrowser(
|
|
self._browser_endpoint
|
|
)
|
|
self.proxies = runtimes.proxies
|
|
self.subscriptions = SubscriptionManager(self.browser)
|
|
self._action_ownership_lock = threading.Lock()
|
|
self._uncertain_actions: dict[str, float] = {}
|
|
|
|
def _browser_endpoint(self, alias: str) -> str:
|
|
return self.runtimes.endpoint(alias)
|
|
|
|
def list_browsers(self) -> list[dict]:
|
|
# Released generations remain in the runtime journal for cleanup/audit,
|
|
# but are not live browser bindings and must not collide with a reused alias.
|
|
return [
|
|
browser
|
|
for browser in self.runtimes.list_public()
|
|
if browser.get("state") != "released"
|
|
]
|
|
|
|
def info(self) -> dict:
|
|
return {
|
|
"service": "browser-gateway",
|
|
"node_id": self.node_id,
|
|
"browser_versions": sorted(self.runtimes.browser_versions),
|
|
}
|
|
|
|
def create(self, input: dict) -> dict:
|
|
return self.runtimes.create(input)
|
|
|
|
def change_state(self, alias: str, action: str, input: dict) -> None:
|
|
generation = decode_generation(
|
|
input, require_runtime=True, require_network=True
|
|
)
|
|
self.runtimes.change_state(alias, action, generation)
|
|
|
|
def remove(self, alias: str, input: dict) -> None:
|
|
generation = decode_generation(
|
|
input,
|
|
require_runtime=False,
|
|
require_network=False,
|
|
allow_profile_purge=True,
|
|
)
|
|
purge_profile = input.get("purge_profile", False)
|
|
if type(purge_profile) is not bool:
|
|
raise RequestError("purge_profile must be boolean", 400)
|
|
self.runtimes.remove(alias, generation, purge_profile)
|
|
|
|
def restore_proxy(self, alias: str, input: dict) -> None:
|
|
validate_proxy_restore(input, alias)
|
|
generation = decode_generation(
|
|
{key: input.get(key) for key in ("binding_version", "runtime_id", "network_id")},
|
|
True,
|
|
True,
|
|
)
|
|
self.runtimes.restore_proxy(alias, generation, input["network_exit"])
|
|
|
|
def _require_generation(self, alias: str, generation: dict) -> tuple[str, bool]:
|
|
try:
|
|
record = self.runtimes.require_generation(alias, generation)
|
|
except FileNotFoundError:
|
|
return "", False
|
|
return record.runtime_id, True
|
|
|
|
def _require_douyin_generation(self, alias: str, input: dict) -> None:
|
|
record = self.runtimes.require_generation(alias, input)
|
|
if input.get("network_exit_id", "") != record.network_exit_id:
|
|
raise RequestError("runtime proxy generation does not match request", 409)
|
|
|
|
def _alias_lock(self, alias: str):
|
|
return self.runtimes.alias_lock(alias)
|
|
|
|
def get_douyin(self, alias: str, input: dict) -> dict:
|
|
if not valid_douyin_generation(input) or not valid_douyin_url(
|
|
input.get("url", "")
|
|
):
|
|
raise RequestError("invalid restricted browser request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
response = self.browser.get(alias, input["url"])
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin GET failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError("restricted browser operation failed") from exc
|
|
return {
|
|
"status": response.status,
|
|
"body": response.body,
|
|
"challenge": response.challenge,
|
|
}
|
|
|
|
def resolve_douyin(self, alias: str, input: dict) -> dict:
|
|
target = input.get("url", "")
|
|
if not valid_douyin_generation(input) or not valid_douyin_source_url(target):
|
|
raise RequestError("invalid restricted Douyin source URL", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
resolved = self.browser.resolve(alias, target)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin share resolution failed alias=%s reason=%s", alias, str(exc)
|
|
)
|
|
raise RequestError("restricted Douyin share resolution failed") from exc
|
|
if not is_douyin_content_url(resolved):
|
|
raise RequestError("Douyin share resolved to an unsupported URL", 502)
|
|
return {"url": resolved}
|
|
|
|
def get_douyin_media(self, alias: str, input: dict) -> dict:
|
|
target = input.get("url", "")
|
|
if not valid_douyin_generation(input) or not isinstance(target, str):
|
|
raise RequestError("invalid restricted browser request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
response = self.browser.get_media(alias, target)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin media download failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError("restricted browser media download failed") from exc
|
|
return {
|
|
"status": response.status,
|
|
"content_type": response.content_type,
|
|
"body_base64": response.body_base64,
|
|
}
|
|
|
|
def douyin_identity(self, alias: str, input: dict) -> dict:
|
|
expected_account_key = input.get("expected_account_key", "")
|
|
if (
|
|
not valid_douyin_generation(input)
|
|
or not isinstance(expected_account_key, str)
|
|
or not ACCOUNT_KEY_RE.fullmatch(expected_account_key)
|
|
):
|
|
raise RequestError("invalid Douyin identity request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
identity = self.browser.identity(alias)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin identity verification failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError(
|
|
"Douyin login identity could not be verified"
|
|
) from exc
|
|
if expected_account_key not in {
|
|
identity["uid"],
|
|
identity["sec_uid"],
|
|
identity["unique_id"],
|
|
}:
|
|
raise RequestError(
|
|
"Douyin identity does not match the expected account", 409
|
|
)
|
|
return identity
|
|
|
|
def douyin_login_qr(self, alias: str, input: dict) -> dict:
|
|
if not valid_douyin_generation(input):
|
|
raise RequestError("invalid Douyin login QR request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
screen = self.browser.login_qr(alias)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin login screen capture failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError("Douyin login screen could not be captured") from exc
|
|
return {
|
|
"content_type": screen.content_type,
|
|
"body_base64": screen.body_base64,
|
|
"qr_detected": screen.qr_detected,
|
|
}
|
|
|
|
def get_xiaohongshu(self, alias: str, input: dict) -> dict:
|
|
target = input.get("url", "")
|
|
if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_url(target):
|
|
raise RequestError("invalid restricted Xiaohongshu request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
response = self.xiaohongshu_browser.get(alias, target)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Xiaohongshu GET failed alias=%s reason=%s", alias, str(exc)
|
|
)
|
|
raise RequestError("restricted Xiaohongshu operation failed") from exc
|
|
return {
|
|
"status": response.status,
|
|
"body": response.body,
|
|
"challenge": response.challenge,
|
|
}
|
|
|
|
def post_xiaohongshu(self, alias: str, input: dict) -> dict:
|
|
target = input.get("url", "")
|
|
body = input.get("body")
|
|
if (
|
|
not valid_xiaohongshu_generation(input)
|
|
or not valid_xhs_post_url(target)
|
|
or not isinstance(body, dict)
|
|
):
|
|
raise RequestError("invalid restricted Xiaohongshu POST request", 400)
|
|
try:
|
|
encoded = json.dumps(
|
|
body, ensure_ascii=False, separators=(",", ":")
|
|
).encode()
|
|
except (TypeError, ValueError) as exc:
|
|
raise RequestError("invalid restricted Xiaohongshu POST body", 400) from exc
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
response = self.xiaohongshu_browser.post(alias, target, encoded)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Xiaohongshu POST failed alias=%s reason=%s", alias, str(exc)
|
|
)
|
|
raise RequestError("restricted Xiaohongshu operation failed") from exc
|
|
return {
|
|
"status": response.status,
|
|
"body": response.body,
|
|
"challenge": response.challenge,
|
|
}
|
|
|
|
def resolve_xiaohongshu(self, alias: str, input: dict) -> dict:
|
|
target = input.get("url", "")
|
|
if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_source_url(
|
|
target
|
|
):
|
|
raise RequestError("invalid restricted Xiaohongshu source URL", 400)
|
|
if not is_xiaohongshu_share_url(target):
|
|
return {"url": target}
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
resolved = self.xiaohongshu_browser.resolve(alias, target)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Xiaohongshu share resolution failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError(
|
|
"restricted Xiaohongshu share resolution failed"
|
|
) from exc
|
|
if not valid_xiaohongshu_page_url(resolved):
|
|
raise RequestError("Xiaohongshu share resolved to an unsupported URL", 502)
|
|
return {"url": resolved}
|
|
|
|
def get_xiaohongshu_media(self, alias: str, input: dict) -> dict:
|
|
target = input.get("url", "")
|
|
if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_media_url(
|
|
target
|
|
):
|
|
raise RequestError("invalid restricted Xiaohongshu media request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
response = self.xiaohongshu_browser.get_media(alias, target)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Xiaohongshu media download failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError(
|
|
"restricted Xiaohongshu media download failed"
|
|
) from exc
|
|
return {
|
|
"status": response.status,
|
|
"content_type": response.content_type,
|
|
"body_base64": response.body_base64,
|
|
}
|
|
|
|
def xiaohongshu_identity(self, alias: str, input: dict) -> dict:
|
|
expected_account_key = input.get("expected_account_key", "")
|
|
if (
|
|
not valid_xiaohongshu_generation(input)
|
|
or not isinstance(expected_account_key, str)
|
|
or not XHS_ACCOUNT_KEY_RE.fullmatch(expected_account_key)
|
|
):
|
|
raise RequestError("invalid Xiaohongshu identity request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
identity = self.xiaohongshu_browser.identity(alias)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Xiaohongshu identity verification failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError(
|
|
"Xiaohongshu login identity could not be verified"
|
|
) from exc
|
|
if identity.get("uid") != expected_account_key:
|
|
raise RequestError(
|
|
"Xiaohongshu identity does not match the expected account", 409
|
|
)
|
|
return identity
|
|
|
|
def douyin_message_history(self, alias: str, input: dict) -> dict:
|
|
expected_uid = input.get("expected_uid", "")
|
|
target_uid = input.get("target_uid", "")
|
|
limit = input.get("limit", 100)
|
|
cursor = input.get("cursor", "")
|
|
if (
|
|
not valid_douyin_generation(input)
|
|
or not isinstance(expected_uid, str)
|
|
or not isinstance(target_uid, str)
|
|
or not isinstance(cursor, str)
|
|
or len(cursor) > 500
|
|
or type(limit) is not int
|
|
or not UID_RE.fullmatch(expected_uid)
|
|
or not UID_RE.fullmatch(target_uid)
|
|
or expected_uid == target_uid
|
|
or not 1 <= limit <= 200
|
|
):
|
|
raise RequestError("invalid Douyin message history request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
if cursor:
|
|
result = self.browser.message_history(
|
|
alias, expected_uid, target_uid, limit, cursor=cursor
|
|
)
|
|
else:
|
|
result = self.browser.message_history(
|
|
alias, expected_uid, target_uid, limit
|
|
)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin message history failed alias=%s reason=%s",
|
|
alias,
|
|
str(exc),
|
|
)
|
|
raise RequestError("Douyin message history failed") from exc
|
|
return result
|
|
|
|
def douyin_action(self, alias: str, input: dict) -> dict:
|
|
expected_uid = input.get("expected_uid", "")
|
|
action = input.get("action", "")
|
|
target_uid = input.get("target_uid", "")
|
|
target_comment_id = input.get("target_comment_id", "")
|
|
target_work_id = input.get("target_work_id", "")
|
|
text = input.get("text", "")
|
|
confirm = input.get("confirm", False)
|
|
operation_id = input.get("operation_id", "")
|
|
if (
|
|
not valid_douyin_generation(input)
|
|
or not isinstance(expected_uid, str)
|
|
or not isinstance(action, str)
|
|
or not isinstance(target_uid, str)
|
|
or not isinstance(target_comment_id, str)
|
|
or not isinstance(target_work_id, str)
|
|
or not isinstance(text, str)
|
|
or not isinstance(operation_id, str)
|
|
or len(operation_id) > 200
|
|
or type(confirm) is not bool
|
|
or not UID_RE.fullmatch(expected_uid)
|
|
or action not in ACTIONS
|
|
):
|
|
raise RequestError("invalid Douyin action request", 400)
|
|
if action == "repost":
|
|
raise RequestError("ACTION_UNAVAILABLE", 409)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
self._claim_action(
|
|
alias,
|
|
str(input.get("runtime_id", "")),
|
|
operation_id,
|
|
)
|
|
try:
|
|
result = self.browser.action(
|
|
alias,
|
|
expected_uid,
|
|
action,
|
|
target_uid,
|
|
target_comment_id,
|
|
target_work_id,
|
|
text,
|
|
confirm,
|
|
)
|
|
self._require_douyin_generation(alias, input)
|
|
except DouyinError as exc:
|
|
self._handle_douyin_action_error(alias, action, exc, operation_id)
|
|
raise RequestError("Douyin action failed") from exc
|
|
except Exception:
|
|
self._release_action_ownership(alias, operation_id)
|
|
raise
|
|
else:
|
|
self._release_action_ownership(alias, operation_id)
|
|
return result
|
|
|
|
def _handle_douyin_action_error(
|
|
self, alias: str, action: str, error: DouyinError, operation_id: str = ""
|
|
) -> None:
|
|
if getattr(error, "uncertain", False) or "timed out" in str(error).lower():
|
|
self._retain_action_ownership(alias, operation_id)
|
|
else:
|
|
self._release_action_ownership(alias, operation_id)
|
|
LOG.warning(
|
|
"Douyin action failed alias=%s action=%s reason=%s",
|
|
alias,
|
|
action,
|
|
str(error),
|
|
)
|
|
|
|
def _claim_action(
|
|
self, alias: str, runtime_id: str = "", operation_id: str = ""
|
|
) -> None:
|
|
now = time.monotonic()
|
|
with self._action_ownership_lock:
|
|
until = self._uncertain_actions.get(alias, 0.0)
|
|
if until > now:
|
|
raise RequestError("previous Douyin action outcome is uncertain", 409)
|
|
if runtime_id:
|
|
marker = self.browser.action_ownership(alias)
|
|
if marker is not None:
|
|
raise RequestError("previous Douyin action outcome is uncertain", 409)
|
|
self.browser.set_action_ownership(
|
|
alias,
|
|
{
|
|
"runtime_id": runtime_id,
|
|
"operation_id": operation_id,
|
|
"started_at": time.time(),
|
|
},
|
|
)
|
|
with self._action_ownership_lock:
|
|
self._uncertain_actions.pop(alias, None)
|
|
self._uncertain_actions[alias] = 0.0
|
|
|
|
def _release_action_ownership(self, alias: str, operation_id: str = "") -> None:
|
|
if operation_id:
|
|
self.browser.clear_action_ownership(alias, operation_id)
|
|
with self._action_ownership_lock:
|
|
self._uncertain_actions.pop(alias, None)
|
|
|
|
def _retain_action_ownership(self, alias: str, operation_id: str = "") -> None:
|
|
with self._action_ownership_lock:
|
|
# A timed-out page script may still finish its network write. Keep the
|
|
# alias blocked until the browser generation is removed or replaced.
|
|
self._uncertain_actions[alias] = math.inf
|
|
|
|
def start_douyin_events(self, alias: str, input: dict) -> dict:
|
|
expected_uid = input.get("expected_uid", "")
|
|
if (
|
|
not valid_douyin_generation(input)
|
|
or not isinstance(expected_uid, str)
|
|
or not UID_RE.fullmatch(expected_uid)
|
|
):
|
|
raise RequestError("invalid Douyin event request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
self.browser.identity(alias, expected_uid)
|
|
return self.subscriptions.start(alias, expected_uid)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin event listener start failed alias=%s uid=%s reason=%s",
|
|
alias,
|
|
expected_uid,
|
|
exc,
|
|
)
|
|
raise RequestError("Douyin event listener could not start") from exc
|
|
|
|
def poll_douyin_events(self, alias: str, input: dict, query: dict) -> list[dict]:
|
|
if not valid_douyin_generation(input):
|
|
raise RequestError("invalid Douyin event request", 400)
|
|
try:
|
|
limit = int(query.get("limit", ["50"])[0])
|
|
wait = float(query.get("wait", ["25"])[0])
|
|
except (IndexError, TypeError, ValueError) as exc:
|
|
raise RequestError("invalid event poll options", 400) from exc
|
|
if not 1 <= limit <= 100 or not 0 <= wait <= 30:
|
|
raise RequestError("invalid event poll options", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
try:
|
|
acknowledgements = query.get("ack", [])
|
|
delivery_ids = [
|
|
value for raw in acknowledgements for value in raw.split(",")
|
|
]
|
|
self.subscriptions.ack(alias, delivery_ids)
|
|
return self.subscriptions.poll(alias, limit, wait)
|
|
except DouyinError as exc:
|
|
LOG.warning(
|
|
"Douyin event listener poll failed alias=%s reason=%s",
|
|
alias,
|
|
exc,
|
|
)
|
|
raise RequestError("Douyin event listener is unavailable") from exc
|
|
|
|
def stop_douyin_events(self, alias: str, input: dict) -> None:
|
|
if not valid_douyin_generation(input):
|
|
raise RequestError("invalid Douyin event request", 400)
|
|
with self._alias_lock(alias):
|
|
self._require_douyin_generation(alias, input)
|
|
self.subscriptions.stop(alias)
|
|
|
|
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()
|
|
|
|
|
|
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", extra={"request": 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:
|
|
needs_body = method in {"POST", "DELETE"} or (
|
|
method == "GET" and parsed.path.endswith("/douyin/events")
|
|
)
|
|
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 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):
|
|
gateway = self.server_as_gateway().gateway
|
|
if method == "GET" and path == "/v1/info":
|
|
return gateway.info()
|
|
if method == "GET" and path == "/v1/browsers":
|
|
return gateway.list_browsers()
|
|
if method == "POST" and path == "/v1/browsers":
|
|
return 201, gateway.create(body)
|
|
match = re.fullmatch(r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})", path)
|
|
if match and method == "DELETE":
|
|
gateway.remove(match.group(1), body)
|
|
return None
|
|
match = re.fullmatch(
|
|
r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/(start|stop|cancel|proxy)", path
|
|
)
|
|
if match:
|
|
alias, action = match.groups()
|
|
if method == "POST" and action in {"start", "stop", "cancel"}:
|
|
gateway.change_state(alias, action, body)
|
|
return None
|
|
if method == "POST" and action == "proxy":
|
|
gateway.restore_proxy(alias, body)
|
|
return None
|
|
match = re.fullmatch(
|
|
r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/xiaohongshu/(get|post|media|identity|resolve)",
|
|
path,
|
|
)
|
|
if match:
|
|
alias, action = match.groups()
|
|
if action == "get" and method == "POST":
|
|
return gateway.get_xiaohongshu(alias, body)
|
|
if action == "post" and method == "POST":
|
|
return gateway.post_xiaohongshu(alias, body)
|
|
if action == "media" and method == "POST":
|
|
return gateway.get_xiaohongshu_media(alias, body)
|
|
if action == "resolve" and method == "POST":
|
|
return gateway.resolve_xiaohongshu(alias, body)
|
|
if action == "identity" and method == "POST":
|
|
return gateway.xiaohongshu_identity(alias, body)
|
|
match = re.fullmatch(
|
|
r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/douyin/(get|media|identity|resolve|login-qr|action|messages|events)",
|
|
path,
|
|
)
|
|
if match:
|
|
alias, action = match.groups()
|
|
if action == "get" and method == "POST":
|
|
return gateway.get_douyin(alias, body)
|
|
if action == "resolve" and method == "POST":
|
|
return gateway.resolve_douyin(alias, body)
|
|
if action == "media" and method == "POST":
|
|
return gateway.get_douyin_media(alias, body)
|
|
if action == "identity" and method == "POST":
|
|
return gateway.douyin_identity(alias, body)
|
|
if action == "login-qr" and method == "POST":
|
|
return gateway.douyin_login_qr(alias, body)
|
|
if action == "action" and method == "POST":
|
|
return gateway.douyin_action(alias, body)
|
|
if action == "messages" and method == "POST":
|
|
return gateway.douyin_message_history(alias, body)
|
|
if action == "events":
|
|
if method == "POST":
|
|
return gateway.start_douyin_events(alias, body)
|
|
if method == "GET":
|
|
return gateway.poll_douyin_events(alias, body, query)
|
|
if method == "DELETE":
|
|
gateway.stop_douyin_events(alias, body)
|
|
return None
|
|
raise RequestError("not found", 404)
|
|
|
|
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:
|
|
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)
|
|
|
|
|
|
def json_bytes(value: object) -> bytes:
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
|
|
|
|
|
|
def valid_xhs_query(
|
|
query: object, allowed: set[str], required: set[str] | None = None
|
|
) -> bool:
|
|
if not isinstance(query, dict) or not isinstance(allowed, set):
|
|
return False
|
|
required = required or set()
|
|
if not required.issubset(query) or not set(query).issubset(allowed):
|
|
return False
|
|
for key, values in query.items():
|
|
if not isinstance(key, str) or not isinstance(values, list) or len(values) != 1:
|
|
return False
|
|
if (
|
|
not isinstance(values[0], str)
|
|
or len(values[0]) > 2048
|
|
or "\r" in values[0]
|
|
or "\n" in values[0]
|
|
):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _valid_xhs_host(parsed: object, host: str) -> bool:
|
|
return (
|
|
getattr(parsed, "scheme", "") == "https"
|
|
and getattr(parsed, "hostname", None) == host
|
|
and getattr(parsed, "port", None) is None
|
|
and getattr(parsed, "username", None) is None
|
|
and getattr(parsed, "password", None) is None
|
|
and getattr(parsed, "fragment", "") == ""
|
|
)
|
|
|
|
|
|
def valid_xhs_url(raw: object) -> bool:
|
|
if not isinstance(raw, str):
|
|
return False
|
|
try:
|
|
parsed = urlsplit(raw)
|
|
query = parse_qs(parsed.query, keep_blank_values=True)
|
|
except ValueError:
|
|
return False
|
|
if (
|
|
_valid_xhs_host(parsed, "edith.xiaohongshu.com")
|
|
and parsed.path == XHS_IDENTITY_PATH
|
|
):
|
|
return not query
|
|
if (
|
|
_valid_xhs_host(parsed, "edith.xiaohongshu.com")
|
|
and parsed.path == XHS_USER_POSTED_PATH
|
|
):
|
|
return (
|
|
valid_xhs_query(
|
|
query,
|
|
{
|
|
"user_id",
|
|
"cursor",
|
|
"num",
|
|
"image_formats",
|
|
"xsec_source",
|
|
"xsec_token",
|
|
},
|
|
{"user_id", "num"},
|
|
)
|
|
and bool(XHS_ACCOUNT_KEY_RE.fullmatch(query["user_id"][0]))
|
|
and query["num"] == ["30"]
|
|
)
|
|
if (
|
|
_valid_xhs_host(parsed, "edith.xiaohongshu.com")
|
|
and parsed.path == XHS_COMMENTS_PATH
|
|
):
|
|
return valid_xhs_query(
|
|
query,
|
|
{
|
|
"note_id",
|
|
"cursor",
|
|
"top_comment_id",
|
|
"image_formats",
|
|
"xsec_source",
|
|
"xsec_token",
|
|
},
|
|
{"note_id", "cursor", "top_comment_id"},
|
|
) and bool(XHS_ACCOUNT_KEY_RE.fullmatch(query["note_id"][0]))
|
|
return False
|
|
|
|
|
|
def valid_xiaohongshu_url(raw: object) -> bool:
|
|
return valid_xhs_url(raw)
|
|
|
|
|
|
def valid_xiaohongshu_page_url(raw: object) -> bool:
|
|
if not isinstance(raw, str):
|
|
return False
|
|
try:
|
|
parsed = urlsplit(raw)
|
|
query = parse_qs(parsed.query, keep_blank_values=True)
|
|
except ValueError:
|
|
return False
|
|
if not _valid_xhs_host(parsed, "www.xiaohongshu.com") or not valid_xhs_query(
|
|
query, {"xsec_source", "xsec_token"}
|
|
):
|
|
return False
|
|
parts = parsed.path.strip("/").split("/")
|
|
return (
|
|
len(parts) == 2
|
|
and parts[0] == "explore"
|
|
and bool(XHS_ACCOUNT_KEY_RE.fullmatch(parts[1]))
|
|
) or (
|
|
len(parts) == 3
|
|
and parts[:2] == ["user", "profile"]
|
|
and bool(XHS_ACCOUNT_KEY_RE.fullmatch(parts[2]))
|
|
)
|
|
|
|
|
|
def valid_xiaohongshu_source_url(raw: object) -> bool:
|
|
return valid_xiaohongshu_page_url(raw) or is_xiaohongshu_share_url(raw)
|
|
|
|
|
|
def valid_xhs_post_url(raw: object) -> bool:
|
|
if not isinstance(raw, str):
|
|
return False
|
|
try:
|
|
parsed = urlsplit(raw)
|
|
query = parse_qs(parsed.query, keep_blank_values=True)
|
|
except ValueError:
|
|
return False
|
|
return (
|
|
_valid_xhs_host(parsed, "so.xiaohongshu.com")
|
|
and parsed.path == XHS_SEARCH_PATH
|
|
and not query
|
|
) or (
|
|
_valid_xhs_host(parsed, "edith.xiaohongshu.com")
|
|
and parsed.path == XHS_FEED_PATH
|
|
and not query
|
|
)
|
|
|
|
|
|
def valid_xiaohongshu_media_url(raw: object) -> bool:
|
|
if not isinstance(raw, str):
|
|
return False
|
|
try:
|
|
parsed = urlsplit(raw)
|
|
query = parse_qs(parsed.query, keep_blank_values=True)
|
|
except ValueError:
|
|
return False
|
|
if not _valid_xhs_host(parsed, "www.xiaohongshu.com"):
|
|
return False
|
|
parts = parsed.path.strip("/").split("/")
|
|
return (
|
|
len(parts) == 2
|
|
and parts[0] == "explore"
|
|
and bool(XHS_ACCOUNT_KEY_RE.fullmatch(parts[1]))
|
|
and valid_xhs_query(query, {"xsec_source", "xsec_token"})
|
|
)
|
|
|
|
|
|
def valid_douyin_source_url(raw: object) -> bool:
|
|
return is_douyin_share_url(raw)
|
|
|
|
|
|
def valid_douyin_url(raw: object) -> bool:
|
|
if not isinstance(raw, str):
|
|
return False
|
|
try:
|
|
parsed = urlsplit(raw)
|
|
except ValueError:
|
|
return False
|
|
if (
|
|
parsed.scheme != "https"
|
|
or parsed.netloc != "www.douyin.com"
|
|
or parsed.username
|
|
or parsed.fragment
|
|
):
|
|
return False
|
|
query = parse_qs(parsed.query, keep_blank_values=True)
|
|
if parsed.path == DOUYIN_IDENTITY_PATH:
|
|
return query == {"aid": ["6383"], "device_platform": ["webapp"]}
|
|
if parsed.path == DOUYIN_PROFILE_OTHER_PATH:
|
|
return valid_douyin_profile_query(query)
|
|
if parsed.path == DOUYIN_WORKS_PATH:
|
|
return valid_douyin_api_query(query, "sec_user_id", "max_cursor")
|
|
if parsed.path == DOUYIN_WORK_DETAIL_PATH:
|
|
return valid_douyin_work_detail_query(query)
|
|
if parsed.path == DOUYIN_COMMENTS_PATH:
|
|
return valid_douyin_api_query(query, "aweme_id", "cursor")
|
|
return False
|
|
|
|
|
|
def valid_douyin_work_detail_query(query: dict[str, list[str]]) -> bool:
|
|
return (
|
|
len(query) == 3
|
|
and query.get("aid") == ["6383"]
|
|
and query.get("device_platform") == ["webapp"]
|
|
and len(query.get("aweme_id", [])) == 1
|
|
and bool(re.fullmatch(r"[1-9][0-9]{0,63}", query["aweme_id"][0]))
|
|
)
|
|
|
|
|
|
def valid_douyin_profile_query(query: dict[str, list[str]]) -> bool:
|
|
account_fields = [field for field in ("user_id", "sec_user_id") if field in query]
|
|
return (
|
|
len(query) == 3
|
|
and query.get("aid") == ["6383"]
|
|
and query.get("device_platform") == ["webapp"]
|
|
and len(account_fields) == 1
|
|
and valid_account_key_query(query, account_fields[0])
|
|
)
|
|
|
|
|
|
def valid_douyin_api_query(
|
|
query: dict[str, list[str]], account_field: str, cursor_field: str
|
|
) -> bool:
|
|
return (
|
|
len(query) == 5
|
|
and query.get("aid") == ["6383"]
|
|
and query.get("device_platform") == ["webapp"]
|
|
and valid_account_key_query(query, account_field)
|
|
and query.get("count") == ["20"]
|
|
and numeric_cursor(query.get(cursor_field))
|
|
)
|
|
|
|
|
|
def valid_account_key_query(query: dict[str, list[str]], key: str) -> bool:
|
|
return len(query.get(key, [])) == 1 and bool(
|
|
DOUYIN_ACCOUNT_KEY_RE.fullmatch(query[key][0])
|
|
)
|
|
|
|
|
|
def numeric_cursor(values: list[str] | None) -> bool:
|
|
if not values or len(values) != 1 or not values[0].isdigit():
|
|
return False
|
|
try:
|
|
return int(values[0]) >= 0
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def parse_proxy_exit(value: object) -> ProxyExit:
|
|
try:
|
|
return _parse_proxy_exit(value)
|
|
except BrowserRuntimeError as exc:
|
|
raise RequestError(str(exc), exc.status, exc.network_id) from exc
|
|
|
|
|
|
def validate_proxy_exit(exit: ProxyExit) -> None:
|
|
try:
|
|
_validate_proxy_exit(exit)
|
|
except BrowserRuntimeError as exc:
|
|
raise RequestError(str(exc), exc.status, exc.network_id) from exc
|
|
|
|
|
|
def validate_create(input: dict) -> None:
|
|
try:
|
|
validate_runtime_input(input)
|
|
except BrowserRuntimeError as exc:
|
|
raise RequestError(str(exc), exc.status, exc.network_id) from exc
|
|
|
|
|
|
def proxy_port(proxy_url: str) -> int:
|
|
return urlsplit(proxy_url).port or 0
|
|
|
|
|
|
def decode_generation(
|
|
value: dict,
|
|
require_runtime: bool,
|
|
require_network: bool,
|
|
allow_profile_purge: bool = False,
|
|
) -> dict:
|
|
allowed = {"binding_version", "runtime_id", "network_id"}
|
|
if allow_profile_purge:
|
|
allowed.add("purge_profile")
|
|
if not isinstance(value, dict) or set(value) - allowed:
|
|
raise RequestError(
|
|
"binding_version, runtime_id and network_id must identify the expected generation",
|
|
400,
|
|
)
|
|
binding = value.get("binding_version")
|
|
runtime = value.get("runtime_id", "")
|
|
network = value.get("network_id", "")
|
|
if (
|
|
type(binding) is not int
|
|
or binding < 1
|
|
or not isinstance(runtime, str)
|
|
or not isinstance(network, str)
|
|
or (
|
|
runtime
|
|
and runtime != RUNTIME_CLEANUP_SENTINEL
|
|
and not RUNTIME_ID_RE.fullmatch(runtime)
|
|
)
|
|
or (network and not NETWORK_ID_RE.fullmatch(network))
|
|
or (require_runtime and not runtime)
|
|
or (require_network and not network)
|
|
):
|
|
raise RequestError(
|
|
"binding_version, runtime_id and network_id must identify the expected generation",
|
|
400,
|
|
)
|
|
return {"binding_version": binding, "runtime_id": runtime, "network_id": network}
|
|
|
|
|
|
def validate_proxy_restore(value: dict, alias: str) -> None:
|
|
del alias
|
|
allowed = {
|
|
"binding_version",
|
|
"runtime_id",
|
|
"network_id",
|
|
"network_exit_id",
|
|
"network_exit",
|
|
}
|
|
if not isinstance(value, dict) or set(value) - allowed:
|
|
raise RequestError("invalid proxy recovery request", 400)
|
|
generation = decode_generation(
|
|
{
|
|
key: value.get(key)
|
|
for key in ("binding_version", "runtime_id", "network_id")
|
|
},
|
|
True,
|
|
True,
|
|
)
|
|
exit_id = value.get("network_exit_id", "")
|
|
if not isinstance(exit_id, str) or (exit_id and not EXIT_ID_RE.fullmatch(exit_id)):
|
|
raise RequestError("invalid proxy recovery request", 400)
|
|
exit = parse_proxy_exit(value.get("network_exit", {}))
|
|
direct = not exit_id and exit == ProxyExit("", "", 0)
|
|
if bool(exit_id) != (not direct):
|
|
raise RequestError("invalid proxy recovery request", 400)
|
|
if not direct:
|
|
validate_proxy_exit(exit)
|
|
value["network_exit"] = exit
|
|
value.update(generation)
|
|
value["network_exit_id"] = exit_id
|
|
|
|
|
|
def valid_douyin_generation(value: dict) -> bool:
|
|
if not isinstance(value, dict):
|
|
return False
|
|
binding = value.get("binding_version")
|
|
runtime = value.get("runtime_id", "")
|
|
network = value.get("network_id", "")
|
|
exit_id = value.get("network_exit_id", "")
|
|
return (
|
|
type(binding) is int
|
|
and binding > 0
|
|
and isinstance(runtime, str)
|
|
and isinstance(network, str)
|
|
and isinstance(exit_id, str)
|
|
and bool(RUNTIME_ID_RE.fullmatch(runtime))
|
|
and bool(NETWORK_ID_RE.fullmatch(network))
|
|
and (not exit_id or bool(EXIT_ID_RE.fullmatch(exit_id)))
|
|
)
|
|
|
|
|
|
def valid_xiaohongshu_generation(value: dict) -> bool:
|
|
return valid_douyin_generation(value)
|
|
|
|
|
|
def _positive_float(env: Mapping[str, str], key: str, default: float, maximum: float) -> float:
|
|
raw = env.get(key, str(default)).strip()
|
|
try:
|
|
value = float(raw)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"{key} must be a number") from exc
|
|
if not 0 < value <= maximum:
|
|
raise ValueError(f"{key} is out of range")
|
|
return value
|
|
|
|
|
|
def _optional_positive_int(env: Mapping[str, str], key: str) -> int | None:
|
|
raw = env.get(key, "").strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
value = int(raw)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"{key} must be an integer") from exc
|
|
if value < 1:
|
|
raise ValueError(f"{key} is out of range")
|
|
return value
|
|
|
|
|
|
def _integer(env: Mapping[str, str], key: str, default: int) -> int:
|
|
raw = env.get(key, str(default)).strip()
|
|
try:
|
|
return int(raw)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"{key} must be an integer") from exc
|
|
|
|
|
|
def _browser_versions(env: Mapping[str, str], default_version: str, default_path: str) -> dict[str, str]:
|
|
result = {default_version: default_path}
|
|
raw = env.get("BROWSER_VERSION_PATHS", "").strip()
|
|
if raw:
|
|
try:
|
|
decoded = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError("BROWSER_VERSION_PATHS must be a JSON object") from exc
|
|
if not isinstance(decoded, dict) or not decoded:
|
|
raise ValueError("BROWSER_VERSION_PATHS must be a non-empty JSON object")
|
|
result = {}
|
|
for version, path in decoded.items():
|
|
if not isinstance(version, str) or not BROWSER_VERSION_RE.fullmatch(version):
|
|
raise ValueError("BROWSER_VERSION_PATHS contains an invalid version")
|
|
if not isinstance(path, str) or not os.path.isabs(os.path.expanduser(path)):
|
|
raise ValueError("BROWSER_VERSION_PATHS contains a non-absolute path")
|
|
result[version] = os.path.abspath(os.path.expanduser(path))
|
|
if default_version not in result:
|
|
result[default_version] = default_path
|
|
return result
|
|
|
|
|
|
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()
|
|
state_dir = os.path.abspath(os.path.expanduser(env.get(
|
|
"BROWSER_STATE_DIR", "~/.local/state/creatorhub/browser-gateway"
|
|
).strip()))
|
|
profile_root = os.path.abspath(os.path.expanduser(env.get(
|
|
"BROWSER_PROFILE_ROOT", "~/.local/share/creatorhub/browser-profiles"
|
|
).strip()))
|
|
browser_version = env.get("BROWSER_VERSION", "148.0.7778.215").strip()
|
|
browser_path = os.path.abspath(os.path.expanduser(env.get(
|
|
"BROWSER_PATH",
|
|
"~/.local/share/creatorhub/browsers/fingerprint-chromium/148.0.7778.215/chrome",
|
|
).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 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 BROWSER_VERSION_RE.fullmatch(browser_version):
|
|
raise ValueError("BROWSER_VERSION is invalid")
|
|
if not node_id:
|
|
node_id = ""
|
|
elif not re.fullmatch(r"^[a-z0-9][a-z0-9._-]{0,63}$", node_id):
|
|
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")
|
|
versions = _browser_versions(env, browser_version, browser_path)
|
|
external_display = _optional_positive_int(env, "RUNTIME_EXTERNAL_DISPLAY")
|
|
for version, path in versions.items():
|
|
try:
|
|
available = os.path.isfile(path) and os.access(path, os.X_OK)
|
|
except (OSError, TypeError) as exc:
|
|
raise ValueError(f"browser version {version} is unavailable") from exc
|
|
if not available:
|
|
raise ValueError(f"browser version {version} is unavailable")
|
|
return {
|
|
"listen": (host, port),
|
|
"state_dir": state_dir,
|
|
"profile_root": profile_root,
|
|
"browser_versions": versions,
|
|
"browser_version": browser_version,
|
|
"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", 15.0, 300.0),
|
|
"min_free_bytes": _integer(env, "RUNTIME_MIN_FREE_BYTES", 20 * 1024**3),
|
|
"log_max_bytes": _integer(env, "RUNTIME_LOG_MAX_BYTES", 1 * 1024**3),
|
|
"profile_cache_max_bytes": _integer(env, "PROFILE_CACHE_MAX_BYTES", 20 * 1024**3),
|
|
"external_display": external_display,
|
|
}
|
|
|
|
|
|
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
|
|
path = Path(state_dir) / "node-id"
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
if path.exists():
|
|
value = path.read_text(encoding="utf-8").strip()
|
|
if not re.fullmatch(r"^[a-z0-9][a-z0-9._-]{0,63}$", value):
|
|
raise ValueError("persisted NODE_ID is invalid")
|
|
return value
|
|
value = "node-" + secrets.token_hex(16)
|
|
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
|
fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as output:
|
|
output.write(value + "\n")
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
with suppress(FileNotFoundError):
|
|
temporary.unlink()
|
|
return value
|
|
|
|
|
|
def run() -> None:
|
|
config = load_config()
|
|
if os.geteuid() == 0:
|
|
raise RuntimeError("browser gateway must run as a non-root user")
|
|
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
|
node_id = load_stable_node_id(config["state_dir"], config["node_id"])
|
|
runtimes = NativeRuntimeManager(
|
|
state_dir=config["state_dir"],
|
|
profile_root=config["profile_root"],
|
|
node_id=node_id,
|
|
browser_versions=config["browser_versions"],
|
|
cleanup_timeout=config["cleanup_timeout"],
|
|
ready_timeout=config["ready_timeout"],
|
|
min_free_bytes=config["min_free_bytes"],
|
|
log_max_bytes=config["log_max_bytes"],
|
|
profile_cache_max_bytes=config["profile_cache_max_bytes"],
|
|
external_display=config["external_display"],
|
|
)
|
|
gateway = Gateway(runtimes, config["token"], node_id)
|
|
server = GatewayHTTPServer(config["listen"], gateway)
|
|
LOG.info(
|
|
json.dumps(
|
|
{
|
|
"service": "browser-gateway",
|
|
"node_id": node_id,
|
|
"node_name": config["node_name"],
|
|
"listen_addr": f"{config['listen'][0]}:{config['listen'][1]}",
|
|
"browser_version": config["browser_version"],
|
|
"external_display": config["external_display"],
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
shutdown_requested = threading.Event()
|
|
|
|
def request_shutdown(_signum, _frame) -> None:
|
|
if shutdown_requested.is_set():
|
|
return
|
|
shutdown_requested.set()
|
|
threading.Thread(
|
|
target=server.shutdown,
|
|
name="gateway-shutdown",
|
|
daemon=True,
|
|
).start()
|
|
|
|
signal.signal(signal.SIGINT, request_shutdown)
|
|
signal.signal(signal.SIGTERM, request_shutdown)
|
|
try:
|
|
server.serve_forever()
|
|
finally:
|
|
server.wait_for_requests(30.0)
|
|
gateway.subscriptions.close()
|
|
runtimes.close()
|
|
server.server_close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|