diff --git a/AGENTS.md b/AGENTS.md index 5d1a05b6..6b7d522a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -86,10 +86,8 @@ API 前缀 `/api/v1`；偶发 broken pipe / 超时，重试即可）。 - **Git over SSH**：`ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/.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` 主机条目，克隆/推送直接可用。 diff --git a/scripts/task_triage.py b/scripts/task_triage.py index 32c03fd1..c77742c4 100644 --- a/scripts/task_triage.py +++ b/scripts/task_triage.py @@ -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: diff --git a/tests/test_task_triage.py b/tests/test_task_triage.py index 3c650243..80231ba5 100644 --- a/tests/test_task_triage.py +++ b/tests/test_task_triage.py @@ -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()