feat(基础设施): 增加旧库隧道和受控 HTTP 配置

将旧库直连地址与 Host 实际使用的本地隧道地址分开建模，增加 SOCKS4A 隧道的启动、就绪探测、异常诊断和关闭逻辑。

补充旧版解码索引、外部平台、模型与嵌入参数等显式配置，并保持凭据使用 SecretStr 隔离。

扩展出站策略：HTTPS 继续默认允许，HTTP 只允许命中独立主机白名单且仍执行 DNS、端口、重定向和私网地址校验。

忽略本机 .env 与 .local 运行数据，防止数据库凭据、索引和持久化产物进入版本库。

script_build_host/.gitignore

@@ -1,4 +1,6 @@
 .venv/
+.env
+.local/
 .pytest_cache/
 .mypy_cache/
 .ruff_cache/

script_build_host/src/script_build_host/infrastructure/config.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from pathlib import Path
 from typing import Self
+from urllib.parse import urlsplit
 
 from pydantic import Field, SecretStr, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -29,12 +30,16 @@ class ScriptBuildSettings(BaseSettings):
 
     environment: str = "production"
     read_database_url: SecretStr = Field(repr=False)
+    read_database_socks_proxy: str | None = None
+    read_database_tunnel_host: str = "127.0.0.1"
+    read_database_tunnel_port: int = Field(default=13306, ge=1, le=65535)
     write_database_url: SecretStr = Field(repr=False)
     final_database_url: SecretStr | None = Field(default=None, repr=False)
     agent_data_root: Path
     persona_root: Path
     section_pattern_root: Path
     decode_index_root: Path
+    decode_raw_root: Path | None = None
     knowledge_root: Path
     prompt_fallback_root: Path
     prompts_from_file: bool = False
@@ -44,9 +49,20 @@ class ScriptBuildSettings(BaseSettings):
     decode_endpoint: str | None = None
     embedding_endpoint: str | None = None
     external_endpoint: str | None = None
+    xhs_search_endpoint: str | None = None
+    xhs_detail_endpoint: str | None = None
+    zhihu_search_endpoint: str | None = None
     image_endpoint: str | None = None
+    openrouter_api_key: SecretStr | None = Field(default=None, repr=False)
+    openrouter_chat_endpoint: str = "https://openrouter.ai/api/v1/chat/completions"
+    embedding_model: str = "qwen/qwen3-embedding-8b"
+    embedding_dimension: int = Field(default=512, ge=1, le=8192)
+    knowledge_model: str = "google/gemini-3-flash-preview"
+    external_image_model: str = "google/gemini-3-flash-preview"
+    decode_min_score: float = Field(default=0.7, ge=-1, le=1)
     outbound_allowed_hosts: tuple[str, ...] = ()
     outbound_allowed_ports: tuple[int, ...] = (443,)
+    outbound_allowed_http_hosts: tuple[str, ...] = ()
     websocket_allowed_origins: tuple[str, ...] = ()
     runtime_factory: str | None = None
     bind_host: str = "127.0.0.1"
@@ -110,8 +126,25 @@ class ScriptBuildSettings(BaseSettings):
         return self
 
     def read_dsn(self) -> str:
+        direct = make_url(self.direct_read_dsn())
+        if self.read_database_socks_proxy:
+            direct = direct.set(
+                host=self.read_database_tunnel_host,
+                port=self.read_database_tunnel_port,
+            )
+        return direct.render_as_string(hide_password=False)
+
+    def direct_read_dsn(self) -> str:
         return self.read_database_url.get_secret_value()
 
+    def read_socks_proxy(self) -> tuple[str, int] | None:
+        if not self.read_database_socks_proxy:
+            return None
+        parsed = urlsplit("//" + self.read_database_socks_proxy)
+        if not parsed.hostname or parsed.port is None:
+            raise ValueError("read_database_socks_proxy must be host:port")
+        return parsed.hostname, parsed.port
+
     def write_dsn(self) -> str:
         return self.write_database_url.get_secret_value()
 

script_build_host/src/script_build_host/infrastructure/outbound.py

@@ -31,20 +31,23 @@ class OutboundPolicy:
     allowed_hosts: frozenset[str]
     allowed_ports: frozenset[int] = frozenset({443})
     resolver: AddressResolver = _system_resolver
+    allowed_http_hosts: frozenset[str] = frozenset()
 
     async def validate_url(self, url: str) -> str:
         try:
             parsed = urlsplit(url)
-            port = parsed.port or 443
+            port = parsed.port or (80 if parsed.scheme.lower() == "http" else 443)
         except ValueError as error:
             raise UnsafeOutboundTarget("the outbound URL is malformed") from error
-        if parsed.scheme.lower() != "https":
-            raise UnsafeOutboundTarget("only HTTPS outbound requests are permitted")
+        scheme = parsed.scheme.lower()
+        host = (parsed.hostname or "").rstrip(".").lower()
+        http_hosts = {item.rstrip(".").lower() for item in self.allowed_http_hosts}
+        if scheme not in {"http", "https"} or (scheme == "http" and host not in http_hosts):
+            raise UnsafeOutboundTarget("only HTTPS or explicitly allowlisted HTTP is permitted")
         if parsed.username is not None or parsed.password is not None:
             raise UnsafeOutboundTarget("outbound URLs must not contain credentials")
         if parsed.fragment:
             raise UnsafeOutboundTarget("outbound URLs must not contain fragments")
-        host = (parsed.hostname or "").rstrip(".").lower()
         if not host or host not in {item.rstrip(".").lower() for item in self.allowed_hosts}:
             raise UnsafeOutboundTarget("the outbound host is not allowlisted")
         if port not in self.allowed_ports:

script_build_host/src/script_build_host/infrastructure/socks_tunnel.py

@@ -0,0 +1,67 @@
+"""Lifecycle-managed SOCKS4A TCP tunnel for the legacy read database."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass, field
+
+
+@dataclass(slots=True)
+class ManagedSocks4aTunnel:
+    proxy_host: str
+    proxy_port: int
+    target_host: str
+    target_port: int
+    listen_host: str = "127.0.0.1"
+    listen_port: int = 13306
+    executable: str = "socat"
+    _process: asyncio.subprocess.Process | None = field(default=None, init=False, repr=False)
+
+    async def start(self) -> None:
+        if self._process is not None and self._process.returncode is None:
+            return
+        self._process = await asyncio.create_subprocess_exec(
+            self.executable,
+            f"TCP-LISTEN:{self.listen_port},bind={self.listen_host},reuseaddr,fork",
+            (
+                f"SOCKS4A:{self.proxy_host}:{self.target_host}:{self.target_port},"
+                f"socksport={self.proxy_port}"
+            ),
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        for _ in range(50):
+            if self._process.returncode is not None:
+                detail = await self._bounded_stderr()
+                raise RuntimeError(f"read database tunnel exited during startup: {detail}")
+            try:
+                _reader, writer = await asyncio.open_connection(self.listen_host, self.listen_port)
+            except OSError:
+                await asyncio.sleep(0.1)
+                continue
+            writer.close()
+            await writer.wait_closed()
+            return
+        await self.stop()
+        raise RuntimeError("read database tunnel did not become ready")
+
+    async def stop(self) -> None:
+        process = self._process
+        self._process = None
+        if process is None or process.returncode is not None:
+            return
+        process.terminate()
+        try:
+            await asyncio.wait_for(process.wait(), timeout=5)
+        except TimeoutError:
+            process.kill()
+            await process.wait()
+
+    async def _bounded_stderr(self) -> str:
+        if self._process is None or self._process.stderr is None:
+            return "no diagnostic"
+        value = await self._process.stderr.read(2_000)
+        return value.decode("utf-8", errors="replace").strip() or "no diagnostic"
+
+
+__all__ = ["ManagedSocks4aTunnel"]