agent-exec transport 507e19cd3dbf43d3bb3646b3b9a2bcfb/fb1caddea8fe

2026-10-18 04:23:58 +00:00
parent 0e58b6331b
commit e5ba5ffa8a
3 changed files with 102 additions and 19 deletions
@@ -86,10 +86,8 @@
  API 前缀 `/api/v1`；偶发 broken pipe / 超时，重试即可）。
 - **Git over SSH**：`ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/<Repo>.git`
  （.gitmodules 与各仓 origin 均为此格式）。
- **API token**：`~/.secrets/gitea`，两行键值 `URL_PREFIX=…` 与
-  `ACCESS_TOKEN=<40 位 hex>`；取值 `grep ACCESS_TOKEN ~/.secrets/gitea | cut -d= -f2-`，
-  用作 `Authorization: token <值>` 请求头。**token 值不入仓、不回显、不写进产出物**
-  （baseline 硬边界 2：只记录来源、字段与可用状态）。
+- **API token**：`~/.secrets/gitea` HOME/IMC 段各有 `- URL_PREFIX=`/`- ACCESS_TOKEN=<40 hex>`，本仓取 HOME 段(F2)：
+  `awk '/^HOME:/{h=1;next} /^IMC:/{h=0} h&&sub(/^- ACCESS_TOKEN=/,""){print;exit}' ~/.secrets/gitea`，作 `Authorization: token <值>` 头。**token 值不入仓、不回显、不进产出物**(baseline 硬边界 2)。
 - **SSH key**：`~/.ssh/home_gitea`（ED25519，公钥指纹
  `SHA256:GYXaOXIMeloQ1fKSfdJ/ndgPQzs/Ix7Hk62ekfkfQ20`，Gitea 侧密钥名 GMKtec）；
  `~/.ssh/config` 已有 `git-ssh.u1680832.nyat.app` 主机条目，克隆/推送直接可用。
@@ -35,6 +35,8 @@ DEFAULT_MAX_AGE = 14400
 PAGE_LIMIT = 50
 MAX_PAGES = 100
 SECRETS_FILE = "~/.secrets/gitea"
+SECTION_RE = re.compile(r"^([A-Za-z][A-Za-z0-9_-]*):$")
+TOKEN_RE = re.compile(r"[A-Za-z0-9._~+/=-]+")
 CARD_RE = re.compile(r"^\.task/[^/]+\.md$")
 CANDIDATE_KEYS = ("branch", "head_sha", "card_path", "card_sha256", "pr_number", "pr_state", "pr_base", "pr_head_sha")
 RECEIPT_KEYS = ("repo", "root_target", "root_target_sha", "requirement", "discovered_at",
@@ -117,21 +119,43 @@ def api_base():
    return (os.environ.get("GITEA_API_URL", "").strip() or DEFAULT_API).rstrip("/")


-def read_token():
-    token = os.environ.get("GITEA_TOKEN", "").strip()
-    if token:
-        return token
+def secrets_file_token():
+    """HOME/IMC 分段时只取 HOME 段的 ACCESS_TOKEN（同 policy/facts.md F2）；无分段时取第一条。"""
    try:
        with open(os.path.expanduser(SECRETS_FILE), encoding="utf-8") as fh:
-            for raw in fh:
-                line = raw.strip()
-                if line.startswith("- "):
-                    line = line[2:].strip()
-                if line.startswith("ACCESS_TOKEN="):
-                    return line.split("=", 1)[1].strip() or None
+            lines = fh.read().splitlines()
    except (OSError, UnicodeDecodeError):
        return None
-    return None
+    first = home = None
+    sectioned = False
+    section = None
+    for raw in lines:
+        line = raw.strip()
+        header = SECTION_RE.match(line)
+        if header:
+            sectioned = True
+            section = header.group(1)
+            continue
+        if line.startswith("- "):
+            line = line[2:].strip()
+        if not line.startswith("ACCESS_TOKEN="):
+            continue
+        value = line.split("=", 1)[1].strip()
+        if first is None:
+            first = value
+        if section == "HOME" and home is None:
+            home = value
+    return (home if sectioned else first) or None
+
+
+def read_token():
+    token = os.environ.get("GITEA_TOKEN", "").strip() or secrets_file_token()
+    if not token:
+        return None
+    if not TOKEN_RE.fullmatch(token):
+        raise Stop(EXIT_REMOTE, "Gitea token is malformed (expected one value without whitespace); "
+                                "check GITEA_TOKEN or ~/.secrets/gitea")
+    return token


 def remote_heads():
@@ -147,15 +171,20 @@ def remote_heads():


 def http_get_json(url, token):
-    request = urllib.request.Request(
-        url, headers={"Accept": "application/json", "Authorization": f"token {token}"})
    try:
+        request = urllib.request.Request(
+            url, headers={"Accept": "application/json", "Authorization": f"token {token}"})
        with urllib.request.urlopen(request, timeout=20) as response:
            body = response.read()
    except urllib.error.HTTPError as exc:
-        raise Stop(EXIT_REMOTE, f"Gitea API returned HTTP {exc.code}", status=exc.code)
+        raise Stop(EXIT_REMOTE, f"Gitea API returned HTTP {exc.code}", status=exc.code) from None
    except (urllib.error.URLError, http.client.HTTPException, OSError) as exc:
-        raise Stop(EXIT_REMOTE, f"Gitea API unreachable ({type(exc).__name__})")
+        raise Stop(EXIT_REMOTE, f"Gitea API unreachable ({type(exc).__name__})") from None
+    except Stop:
+        raise
+    except Exception as exc:
+        # 异常文本可能含 Authorization 头（如 http.client 的 Invalid header value），只报类型名。
+        raise Stop(EXIT_REMOTE, f"Gitea API request failed ({type(exc).__name__})") from None
    try:
        return json.loads(body.decode("utf-8"))
    except ValueError:
@@ -11,6 +11,7 @@ import sys
 import tempfile
 import unittest
 from pathlib import Path
+from unittest import mock

 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from test_task import FakeGitea  # noqa: E402
@@ -397,5 +398,60 @@ class RedactionTests(unittest.TestCase):
        self.assertEqual(task_triage.redact_url("/tmp/plain/remote.git"), "/tmp/plain/remote.git")


+class TokenSafetyTests(unittest.TestCase):
+    FAKE_A = "a" * 40
+    FAKE_B = "b" * 40
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory(prefix="triage-token-")
+        self.addCleanup(tmp.cleanup)
+        self.home = Path(tmp.name)
+
+    def env(self, token=""):
+        return mock.patch.dict(os.environ, {"HOME": str(self.home), "GITEA_TOKEN": token})
+
+    def write_secrets(self, text):
+        (self.home / ".secrets").mkdir(exist_ok=True)
+        (self.home / ".secrets" / "gitea").write_text(text, encoding="utf-8")
+
+    def test_multiline_env_token_is_rejected_without_echoing_values(self):
+        with self.env(f"{self.FAKE_A}\n{self.FAKE_B}"):
+            with self.assertRaises(task_triage.Stop) as caught:
+                task_triage.read_token()
+        self.assertEqual(caught.exception.code, task_triage.EXIT_REMOTE)
+        self.assertIn("malformed", caught.exception.message)
+        for value in (self.FAKE_A, self.FAKE_B):
+            self.assertNotIn(value, caught.exception.message)
+            self.assertNotIn(value, str(caught.exception))
+
+    def test_secrets_file_with_sections_yields_home_value(self):
+        self.write_secrets("IMC:\n- URL_PREFIX=https://imc.invalid\n- ACCESS_TOKEN=" + self.FAKE_B + "\n"
+                           "HOME:\n- URL_PREFIX=https://home.invalid\n- ACCESS_TOKEN=" + self.FAKE_A + "\n")
+        with self.env():
+            self.assertEqual(task_triage.read_token(), self.FAKE_A)
+
+    def test_secrets_file_without_home_access_token_returns_none(self):
+        self.write_secrets("HOME:\n- URL_PREFIX=https://home.invalid\nIMC:\n- ACCESS_TOKEN=" + self.FAKE_B + "\n")
+        with self.env():
+            self.assertIsNone(task_triage.read_token())
+
+    def test_missing_configuration_returns_none(self):
+        with self.env():
+            self.assertIsNone(task_triage.read_token())
+
+    def test_request_value_error_does_not_leak_token(self):
+        token = self.FAKE_A
+        leak = ValueError(f"Invalid header value b'token {token}\\n{self.FAKE_B}'")
+        with mock.patch.object(task_triage.urllib.request, "urlopen", side_effect=leak):
+            with self.assertRaises(task_triage.Stop) as caught:
+                task_triage.http_get_json("https://gitea.invalid/api/v1/x", token)
+        self.assertEqual(caught.exception.code, task_triage.EXIT_REMOTE)
+        self.assertEqual(caught.exception.message, "Gitea API request failed (ValueError)")
+        self.assertIsNone(caught.exception.__cause__)
+        self.assertTrue(caught.exception.__suppress_context__)
+        for value in (token, self.FAKE_B):
+            self.assertNotIn(value, caught.exception.message)
+
+
 if __name__ == "__main__":
    unittest.main()