445 lines
28 KiB
Python
445 lines
28 KiB
Python
|
#!/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'<div class="metric"><div>{k.upper()}</div><strong>{data["summary"][k]}</strong></div>' 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'<div class="item" data-status="{e(c["status"])}"><span class="tag {e(c["status"])}">{e(c["status"].upper())}</span><div><div class="description">{e(c["description"])}</div><div class="detail">{e(c["detail"])}</div>' + (f'<div class="detail">建议：{e(c["remediation"])}</div>' if c["remediation"] else "") + f'</div><span class="secondary">{e(c["code"])}</span></div>')
|
|||
|
|
return ("<!doctype html><html lang=\"zh-CN\"><head><meta charset=\"utf-8\"><meta name=\"viewport\" content=\"width=device-width,initial-scale=1\"><title>SiteForge · 验收中心</title><style>" + STYLE + "</style></head><body>"
|
|||
|
|
'<header><h1>SiteForge · 生产验收中心</h1><div>V8.18 Read-only Acceptance Kit · 生成时间：' + html.escape(data["generated_at"]) + '</div></header>'
|
|||
|
|
'<main><section class="cards">' + cards + '</section><section class="panel"><h2>检查结论</h2><p>本报告不会将跳过项、模拟测试或人工登记直接视为生产通过。正式发布必须有真实数据库、Docker、浏览器和回滚证据。</p><div class="tools"><button aria-pressed="true" data-filter="all">全部</button><button aria-pressed="false" data-filter="fail">失败</button><button aria-pressed="false" data-filter="warn">警告</button><button aria-pressed="false" data-filter="skip">待验收</button><button aria-pressed="false" data-filter="pass">通过</button></div><div id="checks">' + "".join(rows) + '</div></section></main><footer>该页面离线可查看，不上传任何诊断数据。</footer><script>document.querySelectorAll("button[data-filter]").forEach(b=>b.addEventListener("click",()=>{document.querySelectorAll("button[data-filter]").forEach(x=>x.setAttribute("aria-pressed",String(x===b)));document.querySelectorAll(".item[data-status]").forEach(x=>x.hidden=b.dataset.filter!=="all"&&x.dataset.status!==b.dataset.filter)}));</script></body></html>')
|
|||
|
|
|
|||
|
|
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())
|