audit: resident website loop architecture and code quality to 10/10

Fix F1 name race with partial unique index, F2 pause/end mid-poll race with guarded updates, F3 redirect/numeric-IP SSRF, F4 quota prefix check, F5 duplicate row_get, F6 audit ternary; S1 unused import, S2 backoff doc, S3 defensive tearDown. Directed 20 pass, full 1700 pass.

cloud/resident.py

@@ -56,21 +56,26 @@ process_due_resident_watches（见 worker 里的调用点），或
 process_due_resident_watches；两者都走 claim 门，两个驱动
 同时跑也不会在同一间隔内重复轮询。
 
-本模块只依赖标准库 + cloud.automations 的两条复用函数
-（_audit_row、_automation_quota_exceeded），不 import
-cloud 其它模块，避免循环 import。
+本模块只依赖标准库 + cloud.automations 的复用函数
+（row_get、_audit_row、_automation_quota_exceeded）与
+cloud.db 的完整性错误类型（仅用于把唯一索引冲突转成
+ValueError，不 import 其它 cloud 模块），避免循环 import。
 """
 
 import hashlib
 import ipaddress
 import re
+import sqlite3
 import time
 import urllib.request
 import uuid
+from urllib.parse import urljoin
 from urllib.parse import urlsplit
 
 from cloud.automations import _audit_row as _auto_audit_row
 from cloud.automations import _automation_quota_exceeded
+from cloud.automations import row_get
+from cloud.db import PostgresIntegrityError as _PgIntegrityError
 
 MIN_INTERVAL_SECONDS = 60
 MAX_INTERVAL_SECONDS = 30 * 86400
@@ -95,14 +100,6 @@ ENDED = "ended"
 _WS_RE = re.compile(r"\s+")
 
 
-def row_get(row, name):
-    """行取值容错：与 automations.row_get 同款语义。"""
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return None
-
-
 WATCH_COLUMNS = (
     "id", "project_id", "name", "target_url", "prompt",
     "interval_seconds", "fetch_timeout_seconds", "status",
@@ -191,13 +188,28 @@ def ensure_resident_tables(conn):
     conn.execute(
         "CREATE INDEX IF NOT EXISTS idx_resident_watches_project"
         " ON resident_watches(project_id, status)")
+    # 同项目未结束名唯一：先查后插挡不住并发，唯一索引才是
+    # 真正的关口（server environments 的同款部分唯一索引，
+    # WHERE status != 'archived' 的 resident 版）；已结束名
+    # 不在索引内，可重用。已存在重复名的旧库建索引会失败，
+    # 此时不炸掉整次建表调用，SELECT 查重仍挡正常路径。
+    try:
+        conn.execute(
+            "CREATE UNIQUE INDEX IF NOT EXISTS"
+            " idx_resident_watches_name"
+            " ON resident_watches(project_id, name)"
+            " WHERE status != 'ended'")
+    except Exception:
+        pass
     conn.execute(
         "CREATE INDEX IF NOT EXISTS idx_resident_polls_watch"
         " ON resident_polls(watch_id, created_at)")
 
 
 def _backoff_seconds(failures):
-    """失败退避：30、60、120……秒，上限 3600 秒。"""
+    """失败退避增量：30、60、120……秒，上限 3600 秒。
+    调用方把它加在间隔之上（下次 = now + interval + backoff），
+    失败后绝不比正常间隔更频繁地重试，行为见失败分支测试。"""
     if failures <= 0:
         return 0
     return min(3600, 30 * (2 ** (failures - 1)))
@@ -236,11 +248,42 @@ def watch_to_dict(row, now=None):
     }
 
 
+def _is_noncanonical_numeric_host(host):
+    """非规范数字 IP 写法：单整数（2130706433）、十六进制
+    （0x7f000001）、前导零八进制（0177.0.0.1）、含十六进制段
+    的点分写法——urllib/socket 的 inet_aton 会把它们解析成
+    数字 IP，ipaddress 却认不出来。规范点分十进制（4 段、
+    每段 0-255、无前导零，0 本身除外）不在此列；普通域名
+    （含字母的标签）也不在此列。"""
+    if re.fullmatch(r"0[xX][0-9a-fA-F]+", host):
+        return True
+    if re.fullmatch(r"[0-9]+", host):
+        return True
+    if "." not in host:
+        return False
+    labels = host.split(".")
+    if not all(re.fullmatch(r"0[xX][0-9a-fA-F]+|[0-9]+", lab)
+               for lab in labels):
+        return False
+    if len(labels) != 4:
+        return True
+    for lab in labels:
+        if lab.startswith("0x") or lab.startswith("0X"):
+            return True
+        if len(lab) > 1 and lab.startswith("0"):
+            return True
+        if int(lab) > 255:
+            return True
+    return False
+
+
 def validate_target_url(url):
     """目标 URL 校验：仅 http/https、无 userinfo、主机非空、
-    字面 localhost / 私有 / 回环 / 链路本地 IP 拒绝（与
+    字面 localhost（含尾点、*.localhost）/ 私有 / 回环 /
+    链路本地 IP / 非规范数字 IP 写法拒绝（与
     websearch._is_blocked_fetch_url 的字面 IP 口径同款，
-    DNS 解析后的私有地址不在 v1 覆盖内，见摘要已知偏差）。"""
+    DNS 解析后的私有地址不在 v1 覆盖内，见摘要已知偏差；
+    重定向目标由 _SafeRedirectHandler 逐跳复用本校验）。"""
     if not url or not isinstance(url, str) or not url.strip():
         raise ValueError("target_url required")
     url = url.strip()
@@ -251,11 +294,13 @@ def validate_target_url(url):
         raise ValueError("target_url must be http or https")
     if parts.username or parts.password:
         raise ValueError("target_url must not contain credentials")
-    host = (parts.hostname or "").strip()
+    host = (parts.hostname or "").strip().lower().rstrip(".")
     if not host:
         raise ValueError("target_url host required")
-    if host == "localhost":
+    if host == "localhost" or host.endswith(".localhost"):
         raise ValueError("localhost URLs are blocked")
+    if _is_noncanonical_numeric_host(host):
+        raise ValueError("non-canonical numeric IP URLs are blocked")
     try:
         addr = ipaddress.ip_address(host)
     except ValueError:
@@ -322,18 +367,29 @@ def create_watch(conn, project_id, name, target_url, prompt, created_by,
         raise ValueError(
             "resident watch name already exists in this project")
     watch_id = uuid.uuid4().hex
-    conn.execute(
-        "INSERT INTO resident_watches(id, project_id, name, target_url,"
-        " prompt, interval_seconds, fetch_timeout_seconds, status,"
-        " last_poll_at, last_claimed_at, next_poll_at,"
-        " last_content_hash, last_change_at, last_change_text,"
-        " consecutive_failures, paused_reason, ended_at,"
-        " created_by, created_at, updated_at)"
-        " VALUES(?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?,"
-        " NULL, NULL, NULL, 0, NULL, NULL, ?, ?, ?)",
-        (watch_id, project_id, name, target_url, prompt,
-         interval_seconds, fetch_timeout_seconds, RUNNING, now,
-         created_by, now, now))
+    try:
+        conn.execute(
+            "INSERT INTO resident_watches(id, project_id, name, target_url,"
+            " prompt, interval_seconds, fetch_timeout_seconds, status,"
+            " last_poll_at, last_claimed_at, next_poll_at,"
+            " last_content_hash, last_change_at, last_change_text,"
+            " consecutive_failures, paused_reason, ended_at,"
+            " created_by, created_at, updated_at)"
+            " VALUES(?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?,"
+            " NULL, NULL, NULL, 0, NULL, NULL, ?, ?, ?)",
+            (watch_id, project_id, name, target_url, prompt,
+             interval_seconds, fetch_timeout_seconds, RUNNING, now,
+             created_by, now, now))
+    except (sqlite3.IntegrityError, _PgIntegrityError) as exc:
+        # 并发下 SELECT 查重与 INSERT 之间可能插进同名行，
+        # 唯一索引是真正的关口：冲突转回与查重同款 ValueError，
+        # 调用方（server）据此返回 400，不把数据库错误外泄。
+        try:
+            conn.rollback()
+        except Exception:
+            pass
+        raise ValueError(
+            "resident watch name already exists in this project") from exc
     conn.commit()
     row = conn.execute(
         "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
@@ -484,11 +540,19 @@ def update_watch(conn, watch_id, fields, now=None):
         if clash is not None:
             raise ValueError(
                 "resident watch name already exists in this project")
-    conn.execute(
-        "UPDATE resident_watches SET name = ?, prompt = ?,"
-        " target_url = ?, interval_seconds = ?,"
-        " fetch_timeout_seconds = ?, updated_at = ? WHERE id = ?",
-        (name, prompt, target_url, interval, timeout, now, watch_id))
+    try:
+        conn.execute(
+            "UPDATE resident_watches SET name = ?, prompt = ?,"
+            " target_url = ?, interval_seconds = ?,"
+            " fetch_timeout_seconds = ?, updated_at = ? WHERE id = ?",
+            (name, prompt, target_url, interval, timeout, now, watch_id))
+    except (sqlite3.IntegrityError, _PgIntegrityError) as exc:
+        try:
+            conn.rollback()
+        except Exception:
+            pass
+        raise ValueError(
+            "resident watch name already exists in this project") from exc
     conn.commit()
     row = conn.execute(
         "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
@@ -540,13 +604,27 @@ def _content_hash(text):
     return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
 
 
+class _SafeRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """重定向逐跳校验：创建时只校验了初始 URL，公开 URL
+    重定向到内网地址会绕过那道校验；相对跳转先 urljoin 成
+    绝对 URL 再校验，不合格直接抛 ValueError，不跟随。"""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        validate_target_url(urljoin(req.full_url, newurl))
+        return super().redirect_request(req, fp, code, msg, headers,
+                                         newurl)
+
+
 def default_fetch(url, timeout):
     """默认廉价抓取：urllib，超时可配，读满 1MB 即断，
     解码失败按 replace 处理。调用方可注入 fetcher 替换它
-    （测试与未来选择器提取都走注入点，不改本函数签名）。"""
+    （测试与未来选择器提取都走注入点，不改本函数签名）。
+    build_opener 带默认处理器（含 ProxyHandler，no_proxy
+    行为与 urlopen 同款），只把重定向处理器换成逐跳校验版。"""
     req = urllib.request.Request(url, headers={
         "User-Agent": "codex-app-server-resident-watch/1.0"})
-    with urllib.request.urlopen(req, timeout=timeout) as resp:
+    opener = urllib.request.build_opener(_SafeRedirectHandler())
+    with opener.open(req, timeout=timeout) as resp:
         raw = resp.read(MAX_FETCH_BYTES + 1)[:MAX_FETCH_BYTES]
     return raw.decode("utf-8", errors="replace")
 
@@ -606,6 +684,18 @@ def _create_change_task(conn, watch, excerpt, now):
     return task_id
 
 
+def _guarded_watch_update(conn, watch_id, sets, vals):
+    """终局 UPDATE 带 status='running' 条件：抓取中途被用户
+    暂停/结束时，条件不满足、rowcount 为 0，调用方据此回滚
+    并放弃本轮（不建任务、不记轮询），而不是把 paused/ended
+    行的 next_poll_at 改回未来值。返回是否成功。"""
+    cur = conn.execute(
+        "UPDATE resident_watches SET %s WHERE id = ? AND status = ?"
+        % ", ".join(sets),
+        tuple(vals) + (watch_id, RUNNING))
+    return cur.rowcount == 1
+
+
 def run_due_resident_watch(conn, watch_id, fetcher=None, now=None):
     """单条常驻监听的驱动唯一入口：抢单 -> 廉价抓取 -> hash
     比对 -> 记轮询 -> 仅变化时建任务。
@@ -624,7 +714,13 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None):
         return None
     row = conn.execute(
         "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
+    if row is None:
+        return None
     watch = watch_to_dict(row, now)
+    if watch["status"] != RUNNING:
+        # 认领后、抓取前被用户暂停/结束：不抓取、不建任务、
+        # 不改行（暂停/结束已把 next_poll_at 清掉，保持原样）。
+        return None
     interval = watch["interval_seconds"]
     timeout = watch["fetch_timeout_seconds"]
     fetch = fetcher or default_fetch
@@ -633,10 +729,13 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None):
     except Exception as exc:
         failures = watch["consecutive_failures"] + 1
         nxt = now + interval + _backoff_seconds(failures)
-        conn.execute(
-            "UPDATE resident_watches SET consecutive_failures = ?,"
-            " next_poll_at = ?, updated_at = ? WHERE id = ?",
-            (failures, nxt, now, watch_id))
+        if not _guarded_watch_update(
+                conn, watch_id,
+                ("consecutive_failures = ?", "next_poll_at = ?",
+                 "updated_at = ?"),
+                (failures, nxt, now)):
+            conn.rollback()
+            return None
         poll = _record_poll(conn, watch_id, watch["project_id"],
                             "failed", None, str(exc)[:MAX_POLL_DETAIL_LEN],
                             None, now)
@@ -651,32 +750,42 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None):
     excerpt = text[:MAX_CHANGE_TEXT_LEN]
     old_hash = watch["last_content_hash"]
     if old_hash is None:
-        conn.execute(
-            "UPDATE resident_watches SET last_content_hash = ?,"
-            " consecutive_failures = 0, next_poll_at = ?,"
-            " updated_at = ? WHERE id = ?",
-            (observed, now + interval, now, watch_id))
+        if not _guarded_watch_update(
+                conn, watch_id,
+                ("last_content_hash = ?", "consecutive_failures = 0",
+                 "next_poll_at = ?", "updated_at = ?"),
+                (observed, now + interval, now)):
+            conn.rollback()
+            return None
         poll = _record_poll(conn, watch_id, watch["project_id"],
                             "baseline", observed, "first poll", None, now)
         conn.commit()
         return poll
     if observed == old_hash:
-        conn.execute(
-            "UPDATE resident_watches SET consecutive_failures = 0,"
-            " next_poll_at = ?, updated_at = ? WHERE id = ?",
-            (now + interval, now, watch_id))
+        if not _guarded_watch_update(
+                conn, watch_id,
+                ("consecutive_failures = 0", "next_poll_at = ?",
+                 "updated_at = ?"),
+                (now + interval, now)):
+            conn.rollback()
+            return None
         poll = _record_poll(conn, watch_id, watch["project_id"],
                             "unchanged", observed, "", None, now)
         conn.commit()
         return poll
-    # 真实变化：配额门只在这里过。
+    # 真实变化：配额门只在这里过。配额门契约是前缀
+    # （quota_check_failed: / quota_exceeded: / None），不用
+    # 子串判定，以免超限文本含该子串时误判。
     gate = _automation_quota_exceeded(conn, watch["created_by"])
-    if gate is not None and "quota_check_failed" not in gate:
-        conn.execute(
+    if gate is not None and not gate.startswith("quota_check_failed"):
+        cur = conn.execute(
             "UPDATE resident_watches SET status = ?,"
             " paused_reason = ?, next_poll_at = NULL, updated_at = ?"
-            " WHERE id = ?",
-            (PAUSED, "quota_exceeded", now, watch_id))
+            " WHERE id = ? AND status = ?",
+            (PAUSED, "quota_exceeded", now, watch_id, RUNNING))
+        if cur.rowcount != 1:
+            conn.rollback()
+            return None
         poll = _record_poll(conn, watch_id, watch["project_id"],
                             "changed_quota_paused", observed, gate,
                             None, now)
@@ -688,22 +797,29 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None):
     if gate is not None:
         failures = watch["consecutive_failures"] + 1
         nxt = now + interval + _backoff_seconds(failures)
-        conn.execute(
-            "UPDATE resident_watches SET consecutive_failures = ?,"
-            " next_poll_at = ?, updated_at = ? WHERE id = ?",
-            (failures, nxt, now, watch_id))
+        if not _guarded_watch_update(
+                conn, watch_id,
+                ("consecutive_failures = ?", "next_poll_at = ?",
+                 "updated_at = ?"),
+                (failures, nxt, now)):
+            conn.rollback()
+            return None
         poll = _record_poll(conn, watch_id, watch["project_id"],
                             "changed_quota_check_failed", observed, gate,
                             None, now)
         conn.commit()
         return poll
+    # 先受保护地推进游标，再建任务：抓取中途被暂停/结束时，
+    # 条件不满足即回滚，本轮不建任务、不记轮询。
+    if not _guarded_watch_update(
+            conn, watch_id,
+            ("last_content_hash = ?", "last_change_at = ?",
+             "last_change_text = ?", "consecutive_failures = 0",
+             "next_poll_at = ?", "updated_at = ?"),
+            (observed, now, excerpt, now + interval, now)):
+        conn.rollback()
+        return None
     task_id = _create_change_task(conn, watch, excerpt, now)
-    conn.execute(
-        "UPDATE resident_watches SET last_content_hash = ?,"
-        " last_change_at = ?, last_change_text = ?,"
-        " consecutive_failures = 0, next_poll_at = ?,"
-        " updated_at = ? WHERE id = ?",
-        (observed, now, excerpt, now + interval, now, watch_id))
     poll = _record_poll(conn, watch_id, watch["project_id"],
                         "changed", observed, excerpt[:MAX_POLL_DETAIL_LEN],
                         task_id, now)

cloud/server.py

@@ -10245,13 +10245,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     "error": "resident watch in status %s cannot %s"
                              % (watch["status"], action)})
             return
+        audit_action = {"pause": "resident_watch_paused",
+                        "resume": "resident_watch_resumed",
+                        "end": "resident_watch_ended"}[action]
         with _auth_store(self.server.db_path) as store:
-            store.audit(user_id, project_id,
-                        "resident_watch_%s" % ("ended" if action == "end"
-                                               else action + "d"
-                                               if action == "pause"
-                                               else "resumed"),
-                        watch_id)
+            store.audit(user_id, project_id, audit_action, watch_id)
         self._send_json(200, watch)
 
     def _triage_inbox(self, project_id):

cloud/tests/test_resident.py

@@ -2,19 +2,23 @@
 抓取一律注入假 fetcher。覆盖创建校验、暂停保游标、恢复续跑、
 结束终态、防重复轮询、仅变化建任务、失败退避、配额耗尽转暂停、
 worker 驱动批量入口。"""
+import http.server
 import os
+import socket
 import sqlite3
 import sys
 import tempfile
-import time
+import threading
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
 
 try:
+    import cloud.resident as resident_module
     from cloud.resident import (
         claim_due_watch,
         create_watch,
+        default_fetch,
         end_watch,
         ensure_resident_tables,
         get_watch,
@@ -24,14 +28,17 @@ try:
         resume_watch,
         run_due_resident_watch,
         update_watch,
+        validate_target_url,
     )
     from cloud.server import init_db
     from cloud.worker import Worker
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    import cloud.resident as resident_module
     from cloud.resident import (
         claim_due_watch,
         create_watch,
+        default_fetch,
         end_watch,
         ensure_resident_tables,
         get_watch,
@@ -41,6 +48,7 @@ except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.p
         resume_watch,
         run_due_resident_watch,
         update_watch,
+        validate_target_url,
     )
     from cloud.server import init_db
     from cloud.worker import Worker
@@ -314,6 +322,172 @@ class ResidentTestCase(unittest.TestCase):
         self.assertTrue(all(p["status"] == "changed" for p in polls))
         self.assertEqual(len(self._tasks()), 2)
 
+    def test_url_validation_blocks_localhost_and_numeric_forms(self):
+        for bad in ("http://localhost./x", "http://foo.localhost/x",
+                    "http://2130706433/x", "http://0x7f000001/x",
+                    "http://0177.0.0.1/x", "http://0x7f.0.0.1/x"):
+            with self.assertRaises(ValueError, msg=bad):
+                validate_target_url(bad)
+        # 规范点分十进制公网 IP 与带尾点的普通域名仍放行。
+        self.assertEqual(validate_target_url("https://8.8.8.8/x"),
+                         "https://8.8.8.8/x")
+        self.assertEqual(validate_target_url("https://example.com./x"),
+                         "https://example.com./x")
+
+    def test_unique_index_guards_concurrent_name_clash(self):
+        idx = self.conn.execute(
+            "SELECT name FROM sqlite_master WHERE type = 'index'"
+            " AND name = 'idx_resident_watches_name'").fetchone()
+        self.assertIsNotNone(idx)
+        self._watch()
+        # 正常路径仍先报 ValueError。
+        with self.assertRaises(ValueError):
+            self._watch()
+        # 绕过 SELECT 查重直接插：唯一索引本身必须拦住，
+        # 这才是并发下真正的关口。
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.conn.execute(
+                "INSERT INTO resident_watches(id, project_id, name,"
+                " target_url, prompt, interval_seconds, status,"
+                " created_by, created_at, updated_at)"
+                " VALUES('dup', 'proj', 'watch1', ?, 'p', 60,"
+                " 'running', 'owner1', ?, ?)",
+                (URL, self.now, self.now))
+        other = self._watch(name="watch2")
+        with self.assertRaises(ValueError):
+            update_watch(self.conn, other["id"], {"name": "watch1"},
+                         now=self.now)
+
+    def test_quota_check_failed_is_transient_not_paused(self):
+        fetcher = _Fetcher("v1")
+        watch = self._watch()
+        run_due_resident_watch(self.conn, watch["id"],
+                               fetcher=fetcher, now=self.now)
+        cursor_hash = get_watch(self.conn, watch["id"],
+                                now=self.now)["last_content_hash"]
+        original = resident_module._automation_quota_exceeded
+        resident_module._automation_quota_exceeded = lambda conn, owner: (
+            "quota_check_failed: boom")
+        try:
+            fetcher.value = "v2"
+            self.now += 61
+            poll = run_due_resident_watch(
+                self.conn, watch["id"], fetcher=fetcher, now=self.now)
+        finally:
+            resident_module._automation_quota_exceeded = original
+        self.assertEqual(poll["status"], "changed_quota_check_failed")
+        after = get_watch(self.conn, watch["id"], now=self.now)
+        self.assertEqual(after["status"], "running")
+        self.assertEqual(after["last_content_hash"], cursor_hash)
+        self.assertEqual(after["consecutive_failures"], 1)
+        self.assertEqual(after["next_poll_at"], self.now + 60 + 30)
+        self.assertEqual(len(self._tasks()), 0)
+
+    def test_pause_during_fetch_aborts_without_task(self):
+        fetcher = _Fetcher("v1")
+        watch = self._watch()
+        run_due_resident_watch(self.conn, watch["id"],
+                               fetcher=fetcher, now=self.now)
+        cursor_hash = get_watch(self.conn, watch["id"],
+                                now=self.now)["last_content_hash"]
+        db_path = self.db_path
+
+        def pausing_fetcher(url, timeout):
+            conn2 = sqlite3.connect(db_path)
+            try:
+                conn2.row_factory = sqlite3.Row
+                pause_watch(conn2, watch["id"], now=self.now + 61)
+            finally:
+                conn2.close()
+            return "v2"
+
+        self.now += 61
+        poll = run_due_resident_watch(
+            self.conn, watch["id"], fetcher=pausing_fetcher, now=self.now)
+        self.assertIsNone(poll)
+        after = get_watch(self.conn, watch["id"], now=self.now)
+        self.assertEqual(after["status"], "paused")
+        self.assertIsNone(after["next_poll_at"])
+        self.assertEqual(after["last_content_hash"], cursor_hash)
+        self.assertEqual(len(self._tasks()), 0)
+        self.assertEqual(
+            [p["status"] for p in list_polls(self.conn, watch["id"])],
+            ["baseline"])
+
+
+class RedirectFetchTestCase(unittest.TestCase):
+    """重定向校验：只把 redirect-test.test 指到本地，其它主机
+    走原始解析；no_proxy 同款处理，tearDown 全部还原。"""
+
+    HOST = "redirect-test.test"
+
+    def setUp(self):
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def do_GET(self):
+                if self.path == "/go":
+                    self.send_response(302)
+                    self.send_header("Location", "/final")
+                elif self.path == "/bad":
+                    self.send_response(302)
+                    self.send_header(
+                        "Location", "http://127.0.0.1:1/secret")
+                else:
+                    body = b"final-body"
+                    self.send_response(200)
+                    self.send_header("Content-Length", str(len(body)))
+                    self.end_headers()
+                    self.wfile.write(body)
+                    return
+                self.end_headers()
+
+            def log_message(self, *args):
+                pass
+
+        self.server = http.server.ThreadingHTTPServer(
+            ("127.0.0.1", 0), Handler)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self._orig_getaddrinfo = socket.getaddrinfo
+
+        def patched(host, port, family=0, type=0, proto=0, flags=0):
+            if host == self.HOST:
+                return [(socket.AF_INET, socket.SOCK_STREAM, 6, "",
+                         ("127.0.0.1", port))]
+            return self._orig_getaddrinfo(
+                host, port, family, type, proto, flags)
+
+        socket.getaddrinfo = patched
+        self._old_no_proxy = os.environ.get("no_proxy")
+        self._old_NO_PROXY = os.environ.get("NO_PROXY")
+        os.environ["no_proxy"] = (
+            (self._old_no_proxy + ",") if self._old_no_proxy
+            else "") + self.HOST
+        os.environ["NO_PROXY"] = (
+            (self._old_NO_PROXY + ",") if self._old_NO_PROXY
+            else "") + self.HOST
+
+    def tearDown(self):
+        socket.getaddrinfo = self._orig_getaddrinfo
+        if self._old_no_proxy is None:
+            os.environ.pop("no_proxy", None)
+        else:
+            os.environ["no_proxy"] = self._old_no_proxy
+        if self._old_NO_PROXY is None:
+            os.environ.pop("NO_PROXY", None)
+        else:
+            os.environ["NO_PROXY"] = self._old_NO_PROXY
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+
+    def test_relative_redirect_followed_and_private_blocked(self):
+        base = "http://%s:%d" % (self.HOST, self.port)
+        self.assertEqual(default_fetch(base + "/go", 10), "final-body")
+        with self.assertRaises(ValueError):
+            default_fetch(base + "/bad", 10)
+
 
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_resident_website_loop.py

@@ -114,13 +114,18 @@ class WebsiteLoopTestCase(unittest.TestCase):
                 returncode=0, stdout="{}", stderr=""))
 
     def tearDown(self):
-        socket.getaddrinfo = self._orig_getaddrinfo
-        if self._old_no_proxy is None:
-            os.environ.pop("no_proxy", None)
+        # setUp 中途失败时属性可能未建：getattr 防御，避免
+        # tearDown 的 AttributeError 掩盖原始失败。
+        if getattr(self, "_orig_getaddrinfo", None) is not None:
+            socket.getaddrinfo = self._orig_getaddrinfo
+        if getattr(self, "_old_no_proxy", None) is None:
+            if "_old_no_proxy" in self.__dict__:
+                os.environ.pop("no_proxy", None)
         else:
             os.environ["no_proxy"] = self._old_no_proxy
-        if self._old_NO_PROXY is None:
-            os.environ.pop("NO_PROXY", None)
+        if getattr(self, "_old_NO_PROXY", None) is None:
+            if "_old_NO_PROXY" in self.__dict__:
+                os.environ.pop("NO_PROXY", None)
         else:
             os.environ["NO_PROXY"] = self._old_NO_PROXY
         try:
@@ -128,15 +133,27 @@ class WebsiteLoopTestCase(unittest.TestCase):
             self.content_server.server_close()
         except Exception:
             pass
-        self.content_thread.join(timeout=5)
-        self.server.shutdown()
-        self.server.server_close()
-        self.thread.join(timeout=5)
+        try:
+            self.content_thread.join(timeout=5)
+        except Exception:
+            pass
+        try:
+            self.server.shutdown()
+            self.server.server_close()
+        except Exception:
+            pass
+        try:
+            self.thread.join(timeout=5)
+        except Exception:
+            pass
         try:
             self.auth.conn.close()
         except Exception:
             pass
-        self.tmp.cleanup()
+        try:
+            self.tmp.cleanup()
+        except Exception:
+            pass
 
     def _req(self, method, path, body=None, key=None):
         conn = http.client.HTTPConnection("127.0.0.1", self.port,

verification/resident-audit-summary.md

@@ -0,0 +1,201 @@
+# 常驻监听网站闭环：独立架构与代码质量审核报告
+
+审核员非实现者。审核对象：网站闭环 worktree
+`~/workspace/codex-app-server-resident-website-loop`，分支
+`feat-resident-website-loop`，头 `23725f9`（父 `561b389`，基底
+网站版 `0722ca3`，`origin/main 622c51e`）。本审核在独立 worktree
+`~/workspace/codex-app-server-resident-audit`（新分支
+`feat-resident-website-loop-audit`，从 `23725f9` 起）上做，闭环
+worktree、resident-task worktree、泛化 worktree、主 checkout
+（feat-repo-picker）一律未动。
+
+## 0. 评分标准（先定标准，再打分）
+
+只评架构与代码质量两项，满分各 10 分。控制台 UI、真实外网、
+生产 PG、双进程抢单实测、其它探针是其它维度/后续项，本轮不做，
+也不用它们冒充或无故扣这两项的分；但若某项确实卡住架构或代码
+质量本身，列为 blocker，不擅自定。
+
+### 架构 10 分标准
+
+- A1 分层干净：resident 模块自有状态机，与任务状态机、自动化
+  定义层互不干扰；server 只做路由/权限/审计，worker 只做驱动
+  边界调用；无循环依赖。
+- A2 状态机可推理：running/paused/ended、暂停保现场、恢复续跑、
+  结束终态 409、失败退避、配额暂停不结束的语义在所有路径一致，
+  且并发下不被竞态破坏（认领、改名、暂停/结束中途轮询）。
+- A3 驱动与认领正确：服务端不起定时器是明说的，实际驱动在
+  worker 轮次边界；claim 条件 UPDATE 是防重复的唯一关口；崩后
+  不锁死。
+- A4 快照/变化语义干净：轮询廉价、仅变化建任务、首次只建基线、
+  每次轮询可查；变化摘录与游标推进在同一事务边界内。
+- A5 安全边界在正确层：URL 校验、超时、1MB 截断、SSRF 防护不
+  只在创建时校验一次，抓取路径（包括重定向）也要守住；v1 的
+  DNS 解析后校验缺口必须明说，不得假装已解决。
+- A6 可扩展结构：网站是本轮唯一用例，但快照/抓取与状态机/认领
+  不应缠成一团；未来加探针时状态机与驱动骨架不应重写（本轮
+  不实现其它探针，只审结构）。
+
+### 代码质量 10 分标准
+
+- Q1 命名、函数长度与复杂度可读；长函数（run_due）分支清晰，
+  不靠注释掩盖逻辑。
+- Q2 无重复、无死代码、无测试值渗入生产代码；复用已有口径
+  （自动化 claim/配额/审计、环境名唯一索引先例）而不是另起一套。
+- Q3 错误处理诚实：失败、配额耗尽、配额检查失败三条路不混淆；
+  审计与轮询记录不静默丢失；异常不炸驱动循环。
+- Q4 测试可读、可隔离、可复现：本地 HTTP/DNS/no_proxy 等测试
+  手段干净还原、无环境泄漏；断言指向行为而非实现细节摆设。
+- Q5 静态检查：居民差分内 ruff E、F 干净（仓库惯用 `%` 格式、
+  宽泛 except 等不在仓库门禁内的规则不计入；基底已有的 E、F
+  不计入本轮）。
+
+## 1. 改动前分数
+
+- 架构：9.0 / 10
+- 代码质量：9.0 / 10
+
+基底是好的：分层、状态机、claim、配额复用、轮询可查、测试
+隔离（DNS/no_proxy 在 loop 测试里干净还原）都属实，定向 15 项
+基线全过（见第 4 节）。扣分只来自下面逐项 findings，不是整体
+推倒重来。
+
+## 2. Findings 清单（改动前）
+
+### 严重
+
+- F1 名称唯一性是先查后插、无数据库约束：create_watch /
+  update_watch 先 SELECT 查重再 INSERT/UPDATE，两个并发请求
+  可同时通过检查、插出同项目同名 running 监听；server 的
+  environments 已有先例：`CREATE UNIQUE INDEX … WHERE status
+  != 'archived'`。本轮用同款部分唯一索引
+  `WHERE status != 'ended'`（已结束名可重用）+ 完整性错误转
+  ValueError 修复。
+- F2 暂停/结束与进行中轮询竞态：run_due 认领后只读一次状态，
+  之后若用户暂停/结束，失败/基线/未变化/配额/变化分支的 UPDATE
+  均不带状态条件，会把 paused/ended 行的 next_poll_at 改回
+  未来值、并可能在用户已结束之后仍建出变化任务。修复：认领后
+  复查状态；全部终局 UPDATE 带 `status='running'` 条件，条件
+  不满足时回滚、不建任务、不记轮询。
+
+### 一般
+
+- F3 抓取路径 SSRF 只守了创建时：default_fetch 用默认 opener，
+  重定向目标不校验，公开 URL 重定向到内网地址可绕过创建校验；
+  另 localhost.、单整数/十六进制/前导零等非规范 IP 写法可绕过
+  字面 IP 检查。修复：重定向处理器逐跳校验（相对跳转先 urljoin
+  再校验）；校验时归一化主机（去尾点、*.localhost 挡掉）并拒绝
+  非规范数字 IP 写法。DNS 解析后校验仍属 v1 缺口，继续列为已知
+  偏差，不假装已解决。
+- F4 配额门判定靠子串：`\"quota_check_failed\" not in gate` 把
+  配额门返回文本当协议用，超限文本一旦含该子串即误判。配额门
+  的真实契约是前缀（`quota_check_failed:` / `quota_exceeded:` /
+  None），修复为 `startswith` 判定，并补 changed_quota_check_failed
+  的定向测试（该分支此前无测试）。
+- F5 row_get 重复造轮：resident 自带与 automations.row_get 逐字
+  同语义的 row_get，却已从 automations 导入另外两个函数。修复：
+  直接导入 automations.row_get，删本地定义。
+- F6 server 审计动作名嵌套三元：`_resident_transition` 里
+  `\"ended\" if … else action + \"d\" if … else \"resumed\"` 可读性差，
+  且改动时易错。修复为显式动作映射字典，审计字符串不变。
+
+### 建议
+
+- S1 test_resident.py 导入了未用的 `time`（ruff F401，居民差分
+  内唯一 E、F 问题）。删除。
+- S2 退避文档与代码不一致：`_backoff_seconds` 文档写“30、60、
+  120……秒”，代码实际下次轮询是 `now + interval + backoff`。
+  行为本身合理（不比间隔更频繁），文档误导。修复文档，行为不动。
+- S3 loop 测试 tearDown 不够防御：setUp 中途失败时 tearDown
+  访问未建属性会掩盖原始失败。修复为 getattr 防御。
+
+### 已核查、不是问题（不扣分）
+
+- 审计行“写在 commit 之后会丢”：核查 `_auto_audit_row` 本身
+  执行 INSERT 后自己 commit，失败/配额暂停/变化三处审计行在
+  模拟关闭连接后仍可查，不是缺陷。
+- server/worker 全量 ruff E、F 的其余项（如 worker.py:2781/2787
+  的 F821、server.py:10293 的 E501）经核对均为基底已有或由居民
+  插入导致的行号位移，非居民新增代码，不计入本轮。
+- 仓库默认 ruff 全规则的 39 项（如 UP031 `%` 格式）与仓库惯用
+  写法冲突，不在仓库门禁内，不计入。
+
+## 3. 修复（全部关闭，每项有对应测试或定向复核）
+
+- F1 名称唯一性：`ensure_resident_tables` 新增部分唯一索引
+  `idx_resident_watches_name ON resident_watches(project_id, name)
+  WHERE status != 'ended'`（server environments 同款口径）；
+  create/update 的 INSERT/UPDATE 包完整性错误转 ValueError
+  （sqlite3.IntegrityError / PostgresIntegrityError，调用方仍
+  返回 400）。测试：索引存在断言、绕过查重直接插报
+  IntegrityError、改名撞名报 ValueError。
+- F2 暂停/结束中途轮询：认领后复查状态（非 running 直接返回
+  None，不改行）；新增 `_guarded_watch_update`，失败/基线/
+  未变化/配额检查失败/变化推进游标的终局 UPDATE 全部带
+  `status='running'` 条件，条件不满足时回滚、不建任务、不记
+  轮询；变化分支先受保护推进游标、再建任务。测试：抓取回调
+  里经第二连接暂停监听，run_due 返回 None、无任务、游标与
+  next_poll_at 保持暂停态、轮询数不变。
+- F3 抓取路径 SSRF：新增 `_SafeRedirectHandler`，重定向逐跳
+  urljoin 后复用 `validate_target_url`，不合格抛 ValueError
+  不跟随；`default_fetch` 改走 `build_opener`（默认处理器含
+  ProxyHandler，no_proxy 行为与 urlopen 同款）；校验归一化
+  主机（去尾点、挡 *.localhost）并新增 `_is_noncanonical_
+  numeric_host` 拒绝单整数/十六进制/前导零八进制等非规范
+  数字 IP 写法。测试：localhost.、*.localhost、2130706433、
+  0x7f000001、0177.0.0.1、0x7f.0.0.1 均报 ValueError，规范
+  公网 IP 与带尾点域名仍放行；相对重定向跟随、私有绝对重
+  定向报 ValueError（DNS 映射测试主机，同 loop 测试口径）。
+- F4 配额门判定：`\"quota_check_failed\" not in gate` 改为
+  `gate.startswith(\"quota_check_failed\")`，与配额门前缀契约
+  对齐。测试：注入返回 `quota_check_failed: boom` 的配额门，
+  变化轮询记 changed_quota_check_failed、状态仍 running、
+  游标不动、失败数 1、下次 = now + 间隔 + 30、无任务。
+- F5 row_get 重复：删 resident 本地定义，直接导入
+  automations.row_get，内部调用点一字未动，模块依赖说明
+  同步更新。
+- F6 server 审计动作名：嵌套三元改为显式映射字典
+  {pause: paused, resume: resumed, end: ended}，审计字符串
+  不变，服务端测试的审计断言仍过。
+- S1 删 test_resident.py 未用 `time` 导入；S2 `_backoff_seconds`
+  文档改为“退避增量，下次 = now + 间隔 + backoff”，行为不动；
+  S3 loop 测试 tearDown 改为 getattr 防御，setUp 中途失败不
+  再掩盖原始失败。
+
+## 4. 测试数字（全部本轮实测，无删断言、无 skip、无改默认值）
+
+- 定向基线（改动前）：`cloud.tests.test_resident` +
+  `test_resident_server` + `test_resident_website_loop` = 15 项全过。
+- 定向改动后：同三文件 = 20 项全过（基线 15 + 新增 5：URL
+  变体、唯一索引、配额检查失败、中途暂停、重定向），4.450 秒；
+  ruff E、F 在居民差分 4 文件上全部通过。
+- 全量（CI 同款三条 discover，改动后、报告提交前跑出，之后
+  无任何代码或文案改动）：顶层 tests 105 项全过；cloud/tests
+  1551 项全过、9 跳过（384.252 秒）；deploy/tests 44 项全过
+  （单独重跑确认）；合计 1700 项、0 失败、9 跳过。对账：闭环
+  基线 1695（105 + 1546 + 44）+ 新增 5 = 1700，无测试丢失、
+  无新失败。
+- 覆盖率：本仓库 CI 无覆盖率门禁，不虚构覆盖率变化；新增
+  5 项定向测试即上述全量的一部分。
+
+## 5. 复评分数
+
+- 架构：10 / 10
+- 代码质量：10 / 10
+
+复评依据：F1/F2 两项严重 findings 已由数据库约束与状态条件
+UPDATE 从机制上关闭（非仅文档说明），并有定向测试锁定；
+F3 安全边界已从“创建时一次校验”补到“抓取重定向逐跳校验”
+与非规范 IP 写法；F4/F5/F6/S1/S2/S3 全部关闭且测试全过。
+剩余的已知偏差（控制台 UI、全文 hash 抖动、DNS 解析后 SSRF、
+真实外网/生产 PG/双进程抢单未验）属其它维度/后续项，已在
+第 6 节单独列出，不计入本轮两项分数，也不被用来冒充已完成。
+
+## 6. 剩余 blocker 与未做事项
+
+- 未做：未 push、未合并、未跑 deploy.py、未部署；不做其它探针、
+  不接控制台 UI、不做生产 PG 与双进程抢单实测。
+- 已知偏差（其它维度，不计入本轮两项分数）：控制台 UI 未接；
+  变化判定为全文 hash、无选择器提取，动态页面抖动可误报；
+  SSRF v1 不做 DNS 解析后校验；真实外网、生产 PG、双进程抢单
+  未验。