From 6525bcafd07f0091b02a8e757e2eacded45e984f Mon Sep 17 00:00:00 2001 From: host Date: Sun, 1 Nov 2026 19:58:03 +0000 Subject: [PATCH] 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 | 242 ++++++++++++++++------ cloud/server.py | 10 +- cloud/tests/test_resident.py | 176 +++++++++++++++- cloud/tests/test_resident_website_loop.py | 37 +++- verification/resident-audit-summary.md | 201 ++++++++++++++++++ 5 files changed, 586 insertions(+), 80 deletions(-) create mode 100644 verification/resident-audit-summary.md diff --git a/cloud/resident.py b/cloud/resident.py index 37684e9..4f76e27 100644 --- a/cloud/resident.py +++ b/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) diff --git a/cloud/server.py b/cloud/server.py index 21f3614..b0a00e4 100644 --- a/cloud/server.py +++ b/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): diff --git a/cloud/tests/test_resident.py b/cloud/tests/test_resident.py index f488731..5966f37 100644 --- a/cloud/tests/test_resident.py +++ b/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() diff --git a/cloud/tests/test_resident_website_loop.py b/cloud/tests/test_resident_website_loop.py index f9d25fe..3c6e999 100644 --- a/cloud/tests/test_resident_website_loop.py +++ b/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, diff --git a/verification/resident-audit-summary.md b/verification/resident-audit-summary.md new file mode 100644 index 0000000..870cca4 --- /dev/null +++ b/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、双进程抢单 + 未验。