Reject non-plaintext credential tokens before pool entry

app/auth_oauth.py

@@ -70,6 +70,18 @@ def loads_strict(text):
     return json.loads(text, parse_constant=_reject_constant)
 
 
+class AuthTokenTypeError(ValueError):
+    """A credential token field is not a plaintext string (e.g. an encrypted envelope)."""
+
+
+def ensure_plaintext_tokens(auth: dict) -> None:
+    """Reject non-string token fields so encrypted envelope objects never reach upstream headers."""
+    for field in ("accessToken", "refreshToken"):
+        value = auth.get(field)
+        if value is not None and not isinstance(value, str):
+            raise AuthTokenTypeError(
+                f"auth.{field} 不是明文字符串（可能为加密信封格式），请改用 OAuth 登录获取明文凭据")
+
 def validate_cred_data(data) -> tuple[str | None, str | None]:
     """Validate credentials and return either the account UID or a safe failure reason."""
     if not isinstance(data, dict):
@@ -85,6 +97,11 @@ def validate_cred_data(data) -> tuple[str | None, str | None]:
     token = auth.get("accessToken") or auth.get("access_token") or auth.get("token")
     if not isinstance(token, str) or not token:
         return None, "缺少有效的 accessToken"
+    # Envelope objects must be rejected before persistence, including token aliases.
+    for field in ("accessToken", "access_token", "token", "refreshToken", "refresh_token"):
+        value = auth.get(field)
+        if value is not None and not isinstance(value, str):
+            return None, f"{field} 不是明文字符串（可能为加密信封格式）"
     for field in ("expiresAt", "lastRefreshTime"):
         value = auth.get(field)
         if value is None:

converter.py

@@ -268,7 +268,12 @@ def _session(self) -> dict:
         self._load_if_stale()
         if self._cached is None:
             raise RuntimeError(f"无法读取 auth 文件：{self.path}")
-        return self._cached
+        s = self._cached
+        auth = s.get("auth")
+        if isinstance(auth, dict):
+            # Encrypted envelope tokens must fail fast instead of leaking into upstream headers.
+            auth_oauth.ensure_plaintext_tokens(auth)
+        return s
 
     def _is_expired(self) -> bool:
         s = self._session()
@@ -517,7 +522,10 @@ def __init__(self, paths: list[Path] | None = None, scan: bool = False,
         self._ledger = None              # Prefer credits expiring sooner.
         self._capacity = AccountCapacity()
         self._scan = scan                # Rescan credentials before selection.
+        # Explicit paths must survive rejection or eviction so a repaired file re-enters.
+        self._configured: list[str] = []
         self._ignored_duplicates: set[str] = set()
+        self._ignored_invalid: dict[str, str | None] = {}
         self._sync_pending: set[str] = set()
         self._syncing: set[str] = set()
         self._sync_event = threading.Event()
@@ -535,6 +543,8 @@ def reload(self, paths: list[Path], *, reset: bool = True):
                          for entry in self._entries if entry.get("uid")}
             for path in paths:
                 cid = str(Path(path).resolve())
+                if cid not in self._configured:
+                    self._configured.append(cid)
                 if not os.path.exists(cid):
                     continue
                 entry = by_id.get(cid)
@@ -543,6 +553,13 @@ def reload(self, paths: list[Path], *, reset: bool = True):
                         entry["cm"].invalidate()
                     try:
                         summary = entry["cm"].summary()
+                    except auth_oauth.AuthTokenTypeError:
+                        # An envelope replacement must not keep serving the stale identity.
+                        by_id.pop(cid, None)
+                        if entry.get("uid"):
+                            have_uids.pop(entry.get("account_key"), None)
+                        self._drop_entry(entry)
+                        continue
                     except Exception:
                         continue  # A damaged file must not block other credentials.
                     generation = entry["cm"]._generation
@@ -576,6 +593,18 @@ def reload(self, paths: list[Path], *, reset: bool = True):
                 manager = CredentialManager(Path(cid))
                 try:
                     summary = manager.summary()
+                except auth_oauth.AuthTokenTypeError as error:
+                    # Envelope tokens cannot be refreshed or used; keep such files out of the pool.
+                    try:
+                        digest = hashlib.sha256(Path(cid).read_bytes()).hexdigest()
+                    except OSError:
+                        digest = None
+                    if self._ignored_invalid.get(cid) != digest:
+                        # Suppression is content-keyed so a different invalid credential
+                        # reusing the path warns again even without a scan while missing.
+                        _log(f"[cred] 拒绝入池（入库校验失败：{error}）: {Path(cid).name}")
+                        self._ignored_invalid[cid] = digest
+                    continue
                 except Exception:
                     summary = {}
                 uid = summary.get("uid")
@@ -594,6 +623,7 @@ def reload(self, paths: list[Path], *, reset: bool = True):
                 self._entries.append(entry)
                 by_id[cid] = entry
                 self._ignored_duplicates.discard(cid)
+                self._ignored_invalid.pop(cid, None)
                 if uid:
                     have_uids[identity_key] = cid
                 self._queue_sync(cid)
@@ -664,10 +694,27 @@ def apply_if_current(self, cm, generation, update):
             update()
             return True
 
+
+    def _drop_entry(self, entry):
+        """Purge one entry whose file became unusable, mirroring prune()'s cleanup."""
+        cid = entry["id"]
+        if self._ledger is not None:
+            self._ledger.remove(cid)
+        self.forget_credential_state(entry)
+        self._entries = [e for e in self._entries if e is not entry]
+        self._model_fail = {key: until for key, until in self._model_fail.items() if key[0] != cid}
+        self._sticky = OrderedDict((key, value) for key, value in self._sticky.items() if value[0] != cid)
+        self._sync_pending.discard(cid)
+        self._syncing.discard(cid)
+        self._sync_retry.pop(cid, None)
+        self._sync_attempts.pop(cid, None)
+        invalidate_model_table()
     def prune(self):
         """Remove missing credential files and their session bindings."""
         with self._lock:
             self._ignored_duplicates = {p for p in self._ignored_duplicates if os.path.exists(p)}
+            # A path whose invalid file vanished must warn again when a new file reuses it.
+            self._ignored_invalid = {p: v for p, v in self._ignored_invalid.items() if os.path.exists(p)}
             before = len(self._entries)
             removed = [e for e in self._entries if not os.path.exists(e["id"])]
             for entry in removed:
@@ -861,7 +908,8 @@ def _expiry_rank(self, e: dict) -> tuple:
         return (exp is None, exp or 0.0)
     def _rescan(self):
         self.prune()
-        paths = find_auth_files() if self._scan else [Path(entry["id"]) for entry in self.entries()]
+        # Explicit mode retries configured paths, including rejected or evicted ones.
+        paths = find_auth_files() if self._scan else [Path(cid) for cid in self._configured]
         self.reload(paths, reset=False)
 
     def _healthy(self, e: dict) -> bool:

tests/test_auth_oauth.py

@@ -77,6 +77,35 @@ def test_validate_cred_data():
     print("✅ test_validate_cred_data")
 
 
+def test_plaintext_token_guard():
+    from app.auth_oauth import AuthTokenTypeError, ensure_plaintext_tokens
+    ensure_plaintext_tokens({})
+    ensure_plaintext_tokens({"accessToken": "t", "refreshToken": "r"})
+    for field in ("accessToken", "refreshToken"):
+        for bad in ({"$wbEncrypted": 1, "envelope": "x"}, 5, ["x"], True):
+            try:
+                ensure_plaintext_tokens({field: bad})
+                raise AssertionError((field, bad))
+            except AuthTokenTypeError as e:
+                assert field in str(e)
+    # Import validation keeps rejecting envelope files before persistence,
+    # including a plaintext accessToken paired with an envelope refreshToken or alias.
+    envelope = _cred(domain="www.workbuddy.ai")
+    envelope["auth"]["accessToken"] = {"$wbEncrypted": 1, "envelope": "x"}
+    uid, err = validate_cred_data(envelope)
+    assert uid is None and "accessToken" in err
+    for field in ("refreshToken", "refresh_token", "access_token", "token"):
+        c = _cred(domain="www.workbuddy.ai")
+        c["auth"][field] = {"$wbEncrypted": 1, "envelope": "x"}
+        uid, err = validate_cred_data(c)
+        assert uid is None and field in err, field
+    alias = _cred(domain="www.workbuddy.ai")
+    alias["auth"].pop("accessToken")
+    alias["auth"]["access_token"] = "plain-token"
+    assert validate_cred_data(alias) == ("u1", None)
+    print("✅ test_plaintext_token_guard")
+
+
 def test_helpers():
     assert _normalize_origin("www.codebuddy.cn") == "https://www.codebuddy.cn"
     assert _normalize_origin("https://WWW.workbuddy.ai/x") == "https://www.workbuddy.ai"
@@ -361,6 +390,7 @@ def test_oauth_endpoint_import(tmp_path=None):
 
 if __name__ == "__main__":
     test_validate_cred_data()
+    test_plaintext_token_guard()
     test_helpers()
     test_build_auth_file()
     test_merge_existing_accounts()

tests/test_credential_runtime.py

@@ -264,6 +264,116 @@ def test_directory_changes_still_add_and_remove_accounts(self):
         pool._rescan()
         self.assertEqual([e["id"] for e in pool.entries()], [str(second)])
 
+    def test_encrypted_envelope_tokens_fail_fast_and_stay_out_of_pool(self):
+        valid = self.credential("valid.info", uid="valid", age=0)
+        envelope = self.root / "envelope.info"
+        envelope.write_text(json.dumps({"account": {"uid": "envelope"}, "auth": {
+            "accessToken": {"$wbEncrypted": 1, "envelope": "c3lu"},
+            "refreshToken": {"$wbEncrypted": 1, "envelope": "c3lu"},
+            "domain": "www.workbuddy.ai", "expiresAt": (time.time() + 86400) * 1000}}),
+            encoding="utf-8")
+        pool = converter.CredentialPool([valid, envelope])
+        self.assertEqual([Path(e["id"]).name for e in pool.entries()], ["valid.info"])
+        rejected = [c for c in self.logs.call_args_list if "拒绝入池" in c.args[0]]
+        self.assertEqual(len(rejected), 1)
+        self.assertIn("accessToken", rejected[0].args[0])
+        pool.reload([valid, envelope], reset=False)  # Repeat reloads neither re-log nor admit.
+        self.assertEqual([Path(e["id"]).name for e in pool.entries()], ["valid.info"])
+        self.assertEqual(len([c for c in self.logs.call_args_list if "拒绝入池" in c.args[0]]), 1)
+        with self.assertRaises(converter.auth_oauth.AuthTokenTypeError):  # Direct use fails fast.
+            converter.CredentialManager(envelope).get_headers()
+
+    def test_envelope_refresh_token_alone_is_rejected(self):
+        path = self.credential("rt.info", uid="rt", age=0)
+        data = json.loads(path.read_text(encoding="utf-8"))
+        data["auth"]["refreshToken"] = {"$wbEncrypted": 1, "envelope": "c3lu"}
+        path.write_text(json.dumps(data), encoding="utf-8")
+        pool = converter.CredentialPool([path])
+        self.assertEqual(pool.entries(), [])
+        rejected = [c for c in self.logs.call_args_list if "拒绝入池" in c.args[0]]
+        self.assertEqual(len(rejected), 1)
+        self.assertIn("refreshToken", rejected[0].args[0])
+
+
+    def test_replaced_envelope_file_at_same_path_logs_again(self):
+        envelope = self.root / "churn.info"
+        def write_envelope(uid):
+            envelope.write_text(json.dumps({"account": {"uid": uid}, "auth": {
+                "accessToken": {"$wbEncrypted": 1, "envelope": "c3lu"},
+                "domain": "www.workbuddy.ai", "expiresAt": (time.time() + 86400) * 1000}}),
+                encoding="utf-8")
+        def rejections():
+            return len([c for c in self.logs.call_args_list if "拒绝入池" in c.args[0]])
+        write_envelope("churn")
+        pool = converter.CredentialPool([envelope], scan=True)
+        self.assertEqual(rejections(), 1)
+        envelope.unlink()
+        write_envelope("other-account")  # A different credential, no scan ran while missing.
+        pool._rescan()
+        self.assertEqual(pool.entries(), [])
+        self.assertEqual(rejections(), 2)
+        pool._rescan()  # Identical content on the next scan stays silent.
+        self.assertEqual(rejections(), 2)
+    def test_recreated_envelope_file_at_same_path_logs_again(self):
+        envelope = self.root / "reuse.info"
+        def write_envelope():
+            envelope.write_text(json.dumps({"account": {"uid": "reuse"}, "auth": {
+                "accessToken": {"$wbEncrypted": 1, "envelope": "c3lu"},
+                "domain": "www.workbuddy.ai", "expiresAt": (time.time() + 86400) * 1000}}),
+                encoding="utf-8")
+        def rejections():
+            return len([c for c in self.logs.call_args_list if "拒绝入池" in c.args[0]])
+        write_envelope()
+        pool = converter.CredentialPool([envelope], scan=True)
+        self.assertEqual(pool.entries(), [])
+        self.assertEqual(rejections(), 1)
+        envelope.unlink()
+        pool._rescan()  # Prune drops the suppressed path once the file is gone.
+        write_envelope()
+        pool._rescan()  # A new invalid file at the same path must warn again.
+        self.assertEqual(pool.entries(), [])
+        self.assertEqual(rejections(), 2)
+
+    def test_swapped_envelope_file_fails_fast_on_next_use(self):
+        path = self.credential(age=0)
+        pool = converter.CredentialPool([path])
+        self.assertEqual(len(pool.entries()), 1)
+        data = json.loads(path.read_text(encoding="utf-8"))
+        data["auth"]["accessToken"] = {"$wbEncrypted": 1, "envelope": "c3lu"}
+        path.write_text(json.dumps(data), encoding="utf-8")
+        pool.reload([path], reset=True)  # A damaged file must not block other credentials.
+        self.assertEqual(pool.entries(), [])  # The stale identity is evicted, not retained.
+        with self.assertRaises(converter.auth_oauth.AuthTokenTypeError):  # Direct use fails fast.
+            converter.CredentialManager(path).get_headers()
+        pool.reload([path], reset=False)  # The next scan rejects it once like any new file.
+        self.assertEqual(pool.entries(), [])
+        rejected = [c for c in self.logs.call_args_list if "拒绝入池" in c.args[0]]
+        self.assertEqual(len(rejected), 1)
+
+    def test_rejected_explicit_path_recovers_when_repaired(self):
+        path = self.credential("explicit.info", uid="explicit", age=0)
+        data = json.loads(path.read_text(encoding="utf-8"))
+        data["auth"]["accessToken"] = {"$wbEncrypted": 1, "envelope": "c3lu"}
+        path.write_text(json.dumps(data), encoding="utf-8")
+        pool = converter.CredentialPool([path])  # Explicit mode (scan=False).
+        self.assertEqual(pool.entries(), [])
+        self.credential("explicit.info", uid="explicit", age=0)  # Repair the same file.
+        pool._rescan()
+        self.assertEqual([Path(e["id"]).name for e in pool.entries()], ["explicit.info"])
+
+    def test_evicted_explicit_path_recovers_when_repaired(self):
+        path = self.credential("evicted.info", uid="evicted", age=0)
+        pool = converter.CredentialPool([path])
+        self.assertEqual(len(pool.entries()), 1)
+        data = json.loads(path.read_text(encoding="utf-8"))
+        data["auth"]["refreshToken"] = {"$wbEncrypted": 1, "envelope": "c3lu"}
+        path.write_text(json.dumps(data), encoding="utf-8")
+        pool.reload([path], reset=True)  # The envelope replacement evicts the entry.
+        self.assertEqual(pool.entries(), [])
+        self.credential("evicted.info", uid="evicted", age=0)  # Repair the same file.
+        pool._rescan()
+        self.assertEqual([Path(e["id"]).name for e in pool.entries()], ["evicted.info"])
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)