#!/usr/bin/env python3 """SiteForge V8.18 read-only enterprise acceptance audit. Runs only non-destructive inspections. No migrations, deploys, account operations, secret dumps, or changes to a SiteForge instance are performed. """ from __future__ import annotations import argparse import datetime as dt import html import ipaddress import json import os import re import shutil import subprocess import sys import urllib.error import urllib.parse import urllib.request from collections import Counter from dataclasses import asdict, dataclass from pathlib import Path from typing import Any from html.parser import HTMLParser CATEGORIES = {"repository": "源码结构", "http": "HTTP 健康探针", "docker": "Docker / Worker", "postgres": "PostgreSQL", "redis": "Redis", "build": "静态构建 / SEO", "release": "发布门禁"} @dataclass class Check: code: str category: str status: str # pass | fail | warn | skip description: str detail: str = "" remediation: str = "" class Report: def __init__(self, args): self.args = args self.checks: list[Check] = [] def add(self, code, category, status, description, detail="", remediation=""): assert status in {"pass", "fail", "warn", "skip"} # Never include raw environment values or exception objects containing DSNs. self.checks.append(Check(code, category, status, description, str(detail), remediation)) def data(self): counts = Counter(c.status for c in self.checks) return { "tool_version": "8.21.1-dev.1", "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(), "mode": "read-only", "live_production_verified": False, "note": "本报告只证明实际执行并标记 PASS 的只读检查；SKIP/WARN 不等同通过，未进行真实发布。", "summary": {x: counts[x] for x in ("pass", "fail", "warn", "skip")}, "selected_checks_clear": bool(self.checks) and counts["fail"] == 0 and counts["warn"] == 0 and counts["skip"] == 0, "production_ready": False, # human approval + end-to-end validation cannot be automated by this kit "checks": [asdict(c) for c in self.checks] } def audit_repo(r: Report, repo: Path): p = repo.resolve() items = { "app": "应用代码 app/", "app/builder.py": "静态构建器 app/builder.py", "app/multilingual.py": "多语言模块 app/multilingual.py", "app/templates/workspace.html": "工作区模板", "app/static/workspace.js": "后台交互脚本", "scripts/check.sh": "Linux 检查脚本", "scripts/http_end_to_end.py": "HTTP 端到端测试", "scripts/verify_pg_bootstrap.py": "PG 隔离验证", "scripts/browser_smoke.py": "浏览器烟测", "compose.enterprise.yaml": "企业 Compose 配置" } if not p.is_dir(): r.add("repo.root", "repository", "fail", "仓库路径存在", "指定路径不存在", "设置 --repo 为 SiteForge V8.16 源码根目录") return r.add("repo.root", "repository", "pass", "仓库根目录可访问", "仅进行只读检查") for name, label in items.items(): ok = (p / name).exists() r.add("repo." + name.replace("/", "."), "repository", "pass" if ok else "warn", label, "存在" if ok else "在当前源码中未找到", "核对 V8.16 工程是否解压完整" if not ok else "") ps = list((p / "app").rglob("*.py")) if (p / "app").is_dir() else [] if ps: import ast errors = [] for f in ps: try: ast.parse(f.read_text(encoding="utf-8-sig"), filename=str(f.relative_to(p))) except (SyntaxError, UnicodeError, OSError) as exc: errors.append(f"{f.relative_to(p)}:{getattr(exc, 'lineno', '?')}") if len(errors) >= 5: break r.add("repo.python_syntax", "repository", "fail" if errors else "pass", "Python 源码语法", ", ".join(errors) if errors else f"解析通过 {len(ps)} 个文件") else: r.add("repo.python_syntax", "repository", "skip", "Python 源码语法", "未找到 app/*.py") if (p / ".env").exists(): try: text = (p / ".env").read_text(encoding="utf-8", errors="replace") bad = [] for key in ("SITEFORGE_COOKIE_SECURE",): m = re.search(rf"(?m)^\s*{key}\s*=\s*(.+?)\s*$", text) if m and m.group(1).strip().strip('"\'').lower() in ("false", "0", "no", "off"): bad.append(key) r.add("repo.env_security", "repository", "warn" if bad else "pass", ".env 公开部署安全配置", "存在应检查的非安全设置（仅字段名）：" + ", ".join(bad) if bad else "未发现显式关闭安全 Cookie 的配置；这并非完整安全审计", "公网环境启用 SITEFORGE_COOKIE_SECURE=true 并确认 HTTPS" if bad else "") except OSError: r.add("repo.env_security", "repository", "warn", ".env 配置读取", "读取失败") else: r.add("repo.env_security", "repository", "skip", ".env 安全配置", "缺少 .env；不打印或记录凭据") def _local_host(url: str) -> bool: parsed = urllib.parse.urlsplit(url) if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password: return False if parsed.hostname == "localhost": return True try: return ipaddress.ip_address(parsed.hostname).is_loopback except ValueError: return False class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def _probe(url: str, limit: int = 256 * 1024) -> tuple[int, str, str]: req = urllib.request.Request(url, headers={"User-Agent": "SiteForgeAcceptance/8.17", "Accept": "application/json,text/html"}) op = urllib.request.build_opener(NoRedirect) try: with op.open(req, timeout=4) as resp: body = resp.read(limit + 1) if len(body) > limit: return resp.status, "", "响应超过上限" return resp.status, body.decode("utf-8", errors="replace"), "" except urllib.error.HTTPError as exc: return exc.code, "", "返回非 2xx 或重定向" except (urllib.error.URLError, TimeoutError, OSError, ValueError): return 0, "", "无法连接或请求超时" def audit_http(r: Report, base: str, allow_remote: bool): p = urllib.parse.urlsplit(base) if not _local_host(base) and not allow_remote: r.add("http.scope", "http", "fail", "检查目标授权范围", "默认仅允许 localhost / 回环地址", "外部已授权的测试域名需要明确传入 --allow-remote") return if p.scheme not in ("http", "https") or not p.netloc or p.query or p.fragment or p.username or p.password: r.add("http.scope", "http", "fail", "检查目标 URL", "URL 无效或包含不允许的参数") return root = f"{p.scheme}://{p.netloc}" + p.path.rstrip("/") r.add("http.scope", "http", "pass", "检查目标授权范围", "只读取健康接口，不提交数据") for endpoint in ("/api/health", "/api/ready"): status, body, error = _probe(root + endpoint) label = "存活探针" if endpoint.endswith("health") else "就绪探针（数据库与存储）" key = "http.health" if endpoint.endswith("health") else "http.ready" result = "pass" if status == 200 and not error else "fail" detail = f"HTTP {status or '不可用'}" if body: try: j = json.loads(body) detail += "; response.status=" + str(j.get("status", "未提供"))[:36] if isinstance(j.get("status"), str) and j["status"].lower() in ("unavailable", "error", "failed"): result = "fail" except (json.JSONDecodeError, AttributeError): detail += "; 非 JSON 响应" result = "fail" if error: detail += "; " + error r.add(key, "http", result, label, detail, "核对 Web/DB/data 卷状态及日志；不得以 /api/health 代替 /api/ready" if result == "fail" else "") status, _, error = _probe(root + "/", limit=1024 * 1024) r.add("http.entry", "http", "pass" if status in (200, 301, 302, 303, 307, 308) else "warn", "后台入口可访问", f"HTTP {status or '不可用'}" + (f"；{error}" if error else ""), "检查反向代理、登录页和静态资源" if status not in (200, 301, 302, 303, 307, 308) else "") def audit_docker(r: Report, repo: Path, compose_file: str): if not shutil.which("docker"): r.add("docker.executable", "docker", "skip", "Docker CLI 可用", "本机未发现 docker 命令") return cmd = ["docker", "compose"] if compose_file: cmd += ["-f", compose_file] try: # No `config` output: resolved Compose may contain environment secrets. check = subprocess.run(cmd + ["config", "-q"], cwd=repo, capture_output=True, timeout=15) except (OSError, subprocess.TimeoutExpired): r.add("docker.compose_config", "docker", "fail", "Compose 配置可解析", "无法执行命令或超时") return r.add("docker.compose_config", "docker", "pass" if check.returncode == 0 else "fail", "Compose 配置可解析", "退出码 " + str(check.returncode) + "（敏感输出已隐藏）") if check.returncode: return try: ps = subprocess.run(cmd + ["ps", "--format", "json"], cwd=repo, capture_output=True, timeout=15) except (OSError, subprocess.TimeoutExpired): r.add("docker.compose_ps", "docker", "fail", "容器状态检查", "无法执行命令或超时") return if ps.returncode: r.add("docker.compose_ps", "docker", "fail", "容器状态检查", "Compose ps 失败（原始错误已隐藏）") return raw = ps.stdout.decode("utf-8", errors="replace").strip() records = [] try: payload = json.loads(raw or "[]") records = payload if isinstance(payload, list) else [payload] except json.JSONDecodeError: try: records = [json.loads(x) for x in raw.splitlines() if x.strip()] except json.JSONDecodeError: r.add("docker.compose_ps", "docker", "warn", "容器状态检查", "无法解析 Compose JSON 版本格式") return if not records: r.add("docker.compose_ps", "docker", "fail", "容器已启动", "未发现运行服务") return for rec in records: service = str(rec.get("Service") or rec.get("Name") or "unknown")[:64] state = str(rec.get("State", "unknown")).lower() health = str(rec.get("Health", "")).lower() ok = state == "running" and health in ("", "healthy", "none") r.add("docker.service." + re.sub(r"[^a-z0-9_-]", "_", service.lower()), "docker", "pass" if ok else "fail", f"服务 {service} 运行及健康", f"state={state}; health={health or '未报告'}") if not any("worker" in str(x.get("Service", "")).lower() for x in records): r.add("docker.worker", "docker", "warn", "独立 Worker 服务", "ps 中未找到名称含 worker 的服务", "核对企业 Compose 服务命名及独立进程") def audit_postgres(r: Report): dsn = os.getenv("SITEFORGE_DATABASE_URL", "") if not dsn.lower().startswith(("postgresql://", "postgres://")): r.add("postgres.config", "postgres", "skip", "PostgreSQL 真实只读检查", "未配置 PostgreSQL DSN 环境变量") return try: import psycopg except ImportError: r.add("postgres.driver", "postgres", "skip", "PostgreSQL 驱动", "未安装 psycopg（不进行写入测试）") return try: with psycopg.connect(dsn, connect_timeout=4, options="-c default_transaction_read_only=on") as con: with con.cursor() as cur: cur.execute("SELECT 1") good = cur.fetchone()[0] == 1 con.rollback() r.add("postgres.query", "postgres", "pass" if good else "fail", "PostgreSQL 只读连接 SELECT 1", "查询成功（凭据/地址均不写入报告）" if good else "返回结果异常") except Exception: r.add("postgres.query", "postgres", "fail", "PostgreSQL 只读连接 SELECT 1", "连接或查询失败（异常详情已隐藏，避免泄露 DSN）", "核对隔离环境、网络、TLS、只读账号与连接设置") def audit_redis(r: Report): url = os.getenv("SITEFORGE_REDIS_URL", "") if not url.lower().startswith(("redis://", "rediss://")): r.add("redis.config", "redis", "skip", "Redis 真实只读检查", "未配置 SITEFORGE_REDIS_URL") return try: import redis except ImportError: r.add("redis.driver", "redis", "skip", "Redis 驱动", "未安装 redis（不进行写入测试）") return try: client = redis.Redis.from_url(url, socket_connect_timeout=4, socket_timeout=4) good = bool(client.ping()) client.close() r.add("redis.ping", "redis", "pass" if good else "fail", "Redis PING 连接", "PING 成功" if good else "PING 响应异常") except Exception: r.add("redis.ping", "redis", "fail", "Redis PING 连接", "连接失败（异常详情已隐藏）", "核对 Redis 凭据、TLS、DNS、防火墙") class Analyzer(HTMLParser): def __init__(self): super().__init__(convert_charrefs=True) self.lang = "" self.canonicals = [] self.links = [] self.hreflangs = [] def handle_starttag(self, tag, attrs): a = dict(attrs) if tag == "html": self.lang = a.get("lang", "") if tag == "a" and a.get("href"): self.links.append(a["href"]) if tag == "link": rel = (a.get("rel") or "").lower().split() if "canonical" in rel: self.canonicals.append(a.get("href", "")) if "alternate" in rel and a.get("hreflang"): self.hreflangs.append((a["hreflang"], a.get("href", ""))) def normalized_route(raw: str) -> str | None: s = (raw or "").strip() if not s or s.startswith("#"): return None if not s.startswith("/"): return None p = urllib.parse.urlsplit(s) if p.query or p.fragment: return None un = urllib.parse.unquote(p.path) if "\\" in un or "\x00" in un or any(x in ("..", ".") for x in un.split("/")): return None if "//" in un: return None return un def path_for_route(build: Path, route: str) -> Path | None: if route == "/": return build / "index.html" rel = route.lstrip("/") p = build / rel / "index.html" if route.endswith("/") else build / rel if p.suffix == "": p = build / rel / "index.html" # resolve may traverse symlinks - do not open files outside build root root = build.resolve() actual = p.resolve() if not actual.is_relative_to(root): return None if p.is_dir(): p = p / "index.html" return p def audit_build(r: Report, build: Path, baseline: Path | None): if not build.is_dir(): r.add("build.root", "build", "fail", "静态构建目录", "构建目录不存在") return files = sorted(build.rglob("*.html")) if not files: r.add("build.html", "build", "fail", "已生成 HTML 页面", "未找到 HTML 文件") return r.add("build.html", "build", "pass", "已生成 HTML 页面", f"发现 {len(files)} 个 HTML 文件") missing_lang, bad_canon, dup_canon, missing_internal, bad_hreflang = [], [], [], [], [] scanned = 0 for f in files[:10000]: if f.is_symlink() or not f.resolve().is_relative_to(build.resolve()): continue try: a = Analyzer() a.feed(f.read_text(encoding="utf-8", errors="replace")) except (OSError, ValueError): continue scanned += 1 rel = str(f.relative_to(build)) if not a.lang.strip(): missing_lang.append(rel) if len(a.canonicals) > 1: dup_canon.append(rel) for c in a.canonicals: if not c.startswith("https://"): bad_canon.append(rel) for lang, href in a.hreflangs: if not re.fullmatch(r"[a-zA-Z]{2,8}(?:-[a-zA-Z0-9]{2,8})*|x-default", lang) or not href.startswith("https://"): bad_hreflang.append(rel) for link in a.links: # SiteForge static output may include absolute route links. Do not resolve # /preview/... against build output; those are environment-specific. route = normalized_route(link) if route is not None: target = path_for_route(build, route) if target is not None and not target.is_file(): missing_internal.append(f"{rel} -> {route}") if len(files) > 10000: r.add("build.scan_limit", "build", "warn", "扫描 HTML 数量", "超过 10000；仅扫描前 10000 个") for code, label, values, severity, solution in [ ("build.lang", "HTML lang 属性", missing_lang, "warn", "为多语言页面明确设置 HTML lang"), ("build.canonical_duplicate", "同页 canonical 唯一", dup_canon, "fail", "一页只能存在一条 canonical"), ("build.canonical_scheme", "绝对 HTTPS canonical", bad_canon, "warn", "生产域名配置完成后重新构建"), ("build.hreflang", "hreflang 语言标记与 HTTPS 地址", bad_hreflang, "warn", "只为已发布且审核过的翻译输出 alternate"), ("build.internal_links", "站内链接对应静态文件", missing_internal, "fail", "修复失效路径或添加正确重定向")]: r.add(code, "build", severity if values else "pass", label, f"问题 {len(values)} 项；样例：" + "; ".join(values[:6]) if values else f"扫描 {scanned} 个文件", solution if values else "") sitemap = build / "sitemap.xml" r.add("build.sitemap", "build", "pass" if sitemap.is_file() else "warn", "Sitemap 文件", "存在" if sitemap.is_file() else "不存在") robots = build / "robots.txt" r.add("build.robots", "build", "pass" if robots.is_file() else "warn", "robots.txt", "存在" if robots.is_file() else "不存在") if baseline is not None: if not baseline.is_file(): r.add("build.url_baseline", "build", "fail", "旧站 URL 清单", "基线文件不存在") else: routes, invalid = set(), [] for idx, s in enumerate(baseline.read_text(encoding="utf-8-sig").splitlines(), 1): if not s.strip() or s.lstrip().startswith("#"): continue route = normalized_route(s) if route is None: invalid.append(idx) else: routes.add(route) misses = sorted(path for path in routes if not (path_for_route(build, path) or Path("/__invalid__")).is_file()) r.add("build.url_baseline", "build", "fail" if invalid or misses else "pass", "旧 URL 1:1 路径覆盖", f"清单 {len(routes)} 条，缺失 {len(misses)} 条，非法输入行 {len(invalid)} 条；缺失样例 " + ", ".join(misses[:8]), "不要静默更改旧站 SEO 路径；缺失 URL 必须修复或提供已审核 301 映射" if misses or invalid else "") def release_gate(r: Report, evidence: Path | None): # The kit NEVER gives a pass for production readiness without signed-off evidence. required = ["isolated_pg", "redis_ping", "docker_health", "backup_restore", "browser_smoke", "wordpress_url_parity", "staging_deploy", "rollback", "production_approval", "art_design_pro"] if evidence is None: r.add("release.evidence", "release", "skip", "正式发布证据表", "未提供 --evidence；不会宣称生产验收完成") return if not evidence.is_file(): r.add("release.evidence", "release", "fail", "正式发布证据表", "找不到 evidence JSON") return try: obj = json.loads(evidence.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): r.add("release.evidence", "release", "fail", "正式发布证据表", "不是可解析 JSON") return for key in required: val = obj.get(key) # Validate dates/links/owners, not merely booleans. valid = isinstance(val, dict) and val.get("status") == "passed" and bool(val.get("operator")) and bool(val.get("date")) and bool(val.get("evidence")) r.add("release." + key, "release", "pass" if valid else "warn", f"签字验收：{key}", "证据字段完整（不验证证据真实性）" if valid else "尚缺 passed/operator/date/evidence 字段", "填入真实执行记录，未通过时禁止正式切流" if not valid else "") def markdown(data: dict) -> str: summ = data["summary"] result = ["# SiteForge V8.18 只读验收报告", "", f"检查日期：{data['generated_at']}", "", "**本工具不能证明完整生产上线。** 只执行只读探针和静态检查；人工证据需二次核实。", "", f"检查结果：PASS {summ['pass']} / FAIL {summ['fail']} / WARN {summ['warn']} / SKIP {summ['skip']}", "", "| 检查项 | 分类 | 结果 | 详情 |", "|---|---|---|---|"] def clean(s): return str(s).replace("|", "\\|").replace("\n", " ") for c in data["checks"]: result.append(f"| {clean(c['description'])} | {clean(CATEGORIES.get(c['category'], c['category']))} | {c['status'].upper()} | {clean(c['detail'])} |") result += ["", "## 修复清单", ""] issues = [c for c in data["checks"] if c["status"] in ("fail", "warn")] result += [f"- **{c['code']}**：{c['remediation'] or c['description']}" for c in issues] or ["- 没有发现 FAIL / WARN；仍需核实未执行的外部验收。"] return "\n".join(result) + "\n" STYLE = """body{font:14px/1.6 system-ui,-apple-system,'Microsoft YaHei',sans-serif;margin:0;background:#f4f6fa;color:#202b3d}header{padding:28px 4%;background:#14243d;color:white}header h1{margin:0;font-size:25px}main{max-width:1180px;margin:auto;padding:24px}.muted{color:#64748b}.panel{background:white;border:1px solid #e2e7f0;border-radius:14px;padding:22px;margin-bottom:20px}.cards{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:12px}.metric{background:white;border-radius:12px;border:1px solid #e2e7f0;padding:16px}.metric strong{display:block;font-size:29px}.tools{display:flex;gap:8px;flex-wrap:wrap;margin:16px 0}button{cursor:pointer;background:#fff;border:1px solid #ccd5e2;padding:9px 14px;border-radius:9px}button[aria-pressed=true]{background:#ff6a2b;color:white;border-color:#ff6a2b}.item{padding:13px 0;border-bottom:1px solid #edf0f4;display:flex;gap:12px;align-items:baseline}.tag{font-size:11px;border-radius:5px;padding:2px 7px;min-width:42px;text-align:center;display:inline-block;background:#e5e7eb}.pass{background:#dcfce7;color:#166534}.fail{background:#fee2e2;color:#991b1b}.warn{background:#fef3c7;color:#92400e}.skip{background:#e2e8f0;color:#475569}.description{font-weight:650}.detail{color:#64748b;font-size:13px}.secondary{margin-left:auto;color:#9ca3af}footer{padding:20px;text-align:center;color:#64748b}@media(max-width:660px){.cards{grid-template-columns:repeat(2,minmax(0,1fr))}.item{display:block}.secondary{display:none}}""" def html_report(data: dict) -> str: cards = "".join(f'
{k.upper()}
{data["summary"][k]}
' for k in ("pass", "fail", "warn", "skip")) rows = [] for c in data["checks"]: e = lambda s: html.escape(str(s), quote=True) rows.append(f'
{e(c["status"].upper())}
{e(c["description"])}
{e(c["detail"])}
' + (f'
建议：{e(c["remediation"])}
' if c["remediation"] else "") + f'
{e(c["code"])}
') return ("SiteForge · 验收中心" '

SiteForge · 生产验收中心

V8.18 Read-only Acceptance Kit · 生成时间：' + html.escape(data["generated_at"]) + '
' '
' + cards + '

检查结论

本报告不会将跳过项、模拟测试或人工登记直接视为生产通过。正式发布必须有真实数据库、Docker、浏览器和回滚证据。

' + "".join(rows) + '
') def main(argv=None): ap = argparse.ArgumentParser(description="SiteForge V8.18 non-destructive acceptance toolkit") ap.add_argument("--repo", type=Path, help="SiteForge 源码根目录") ap.add_argument("--url", help="只读 HTTP 健康检查 URL；默认仅支持 loopback") ap.add_argument("--allow-remote", action="store_true", help="明确允许访问已授权远端测试目标") ap.add_argument("--docker", action="store_true", help="运行只读 docker compose 命令") ap.add_argument("--compose-file", default="compose.enterprise.yaml", help="默认企业 Compose 文件") ap.add_argument("--postgres", action="store_true", help="只执行 PostgreSQL SELECT 1") ap.add_argument("--redis", action="store_true", help="只执行 Redis PING") ap.add_argument("--build-dir", type=Path, help="检查已生成的静态站目录") ap.add_argument("--url-baseline", type=Path, help="旧站 URL 路径清单 TXT") ap.add_argument("--evidence", type=Path, help="人工签字验收证据 JSON") ap.add_argument("--output", type=Path, default=Path("siteforge-audit"), help="报告文件前缀") args = ap.parse_args(argv) r = Report(args) if not any((args.repo, args.url, args.docker, args.postgres, args.redis, args.build_dir, args.evidence)): ap.error("至少选择一个检查目标，例如 --repo . --docker 或 --url http://127.0.0.1:8787") if args.repo: audit_repo(r, args.repo) if args.url: audit_http(r, args.url, args.allow_remote) if args.docker: audit_docker(r, args.repo or Path.cwd(), args.compose_file) if args.postgres: audit_postgres(r) if args.redis: audit_redis(r) if args.build_dir: audit_build(r, args.build_dir, args.url_baseline) if args.url_baseline and not args.build_dir: r.add("build.baseline_without_build", "build", "warn", "旧站 URL 路径基线", "必须同时传入 --build-dir") if args.evidence: release_gate(r, args.evidence) else: release_gate(r, None) data = r.data() args.output.parent.mkdir(parents=True, exist_ok=True) args.output.with_suffix(".json").write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") args.output.with_suffix(".md").write_text(markdown(data), encoding="utf-8") args.output.with_suffix(".html").write_text(html_report(data), encoding="utf-8") print(f"PASS={data['summary']['pass']} FAIL={data['summary']['fail']} WARN={data['summary']['warn']} SKIP={data['summary']['skip']}") print("报告已写入：", args.output.with_suffix(".json"), args.output.with_suffix(".md"), args.output.with_suffix(".html")) return 2 if data["summary"]["fail"] else 0 if __name__ == "__main__": sys.exit(main())