feat(governance): automate harness consistency checks (phase 3)
Harness governance / validate (push) Has been cancelled
Harness governance / validate (push) Has been cancelled
This commit is contained in:
chengma
@@ -0,0 +1,685 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Offline governance checks for a Harness Coding repository."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import unicodedata
|
||||
import urllib.parse
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any, Iterable
|
||||
|
||||
from validate_agent_context import validate_manifest
|
||||
|
||||
|
||||
TASK_ID = re.compile(r"^T-\d{3}[a-z]?$")
|
||||
SHA40 = re.compile(r"^[0-9a-fA-F]{40}$")
|
||||
MARKDOWN_LINK = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)")
|
||||
URI_SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:")
|
||||
AUTH_VALUE = re.compile(
|
||||
r"(?i)authorization\s*[:=]\s*['\"]?(?:basic|bearer|token)\s+[A-Za-z0-9._~+/=-]{8,}"
|
||||
)
|
||||
TOKEN_ASSIGNMENT = re.compile(
|
||||
r"(?i)^\s*\{?\s*(?:(?:export\s+)?(?:\$env:)?GITEA_TOKEN|['\"]GITEA_TOKEN['\"])"
|
||||
r"\s*[:=]\s*(.*?)\s*[,}]?\s*$"
|
||||
)
|
||||
URL_CREDENTIAL = re.compile(r"(?i)https?://[^/\s:@]+:[^/\s@]+@")
|
||||
GITEA_TOKEN_LITERAL = re.compile(r"\bgta_[A-Za-z0-9_-]{16,}\b")
|
||||
CMD_TOKEN_ASSIGNMENT = re.compile(
|
||||
r"(?ix)^\s*(?:"
|
||||
r"setx\s+(?:\"GITEA_TOKEN\"|GITEA_TOKEN)\s+(?:\"([^\"]*)\"|(.*?))"
|
||||
r"|set\s+(?:\"GITEA_TOKEN\s*=\s*([^\"]*)\"|GITEA_TOKEN\s*=\s*(.*?))"
|
||||
r")\s*$"
|
||||
)
|
||||
DOTNET_TOKEN_SETTER = re.compile(
|
||||
r"(?is)\[Environment\]::SetEnvironmentVariable\s*\(\s*['\"]GITEA_TOKEN['\"]"
|
||||
r"\s*,\s*(['\"])(.*?)\1"
|
||||
)
|
||||
SAFE_VARIABLE_REFERENCE = re.compile(
|
||||
r"(?i)(?:\$\{[A-Za-z_][A-Za-z0-9_]*\}|\$env:[A-Za-z_][A-Za-z0-9_]*|"
|
||||
r"\$[A-Za-z_][A-Za-z0-9_]*|%[A-Za-z_][A-Za-z0-9_]*%)"
|
||||
)
|
||||
TASK_REQUIRED_FIELDS = {
|
||||
"id",
|
||||
"title",
|
||||
"phase",
|
||||
"deps",
|
||||
"status",
|
||||
"created",
|
||||
"issue",
|
||||
"context_ref",
|
||||
"claim_branch",
|
||||
"work_branch",
|
||||
"write_paths",
|
||||
}
|
||||
TASK_REQUIRED_SECTIONS = {
|
||||
"问题 / 背景",
|
||||
"方案",
|
||||
"验收要点",
|
||||
"边界（不改什么）",
|
||||
"协作约束",
|
||||
"执行记录",
|
||||
}
|
||||
VALID_STATUS = {"TODO", "DOING", "DONE", "BLOCKED"}
|
||||
ACTIVE_STATUS = {"DOING", "BLOCKED"}
|
||||
KNOWN_TEXT_SUFFIXES = {
|
||||
".md",
|
||||
".py",
|
||||
".ps1",
|
||||
".sh",
|
||||
".json",
|
||||
".yaml",
|
||||
".yml",
|
||||
".toml",
|
||||
".txt",
|
||||
".env",
|
||||
".example",
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, order=True)
|
||||
class Finding:
|
||||
rule: str
|
||||
path: str
|
||||
line: int
|
||||
message: str
|
||||
|
||||
def render(self) -> str:
|
||||
location = self.path if self.line <= 0 else f"{self.path}:{self.line}"
|
||||
return f"ERROR [{self.rule}] {location}: {self.message}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Task:
|
||||
path: Path
|
||||
metadata: dict[str, Any]
|
||||
body: str
|
||||
|
||||
@property
|
||||
def task_id(self) -> str:
|
||||
value = self.metadata.get("id")
|
||||
return value if isinstance(value, str) else ""
|
||||
|
||||
@property
|
||||
def status(self) -> str:
|
||||
value = self.metadata.get("status")
|
||||
return value if isinstance(value, str) else ""
|
||||
|
||||
|
||||
def relative(path: Path, root: Path) -> str:
|
||||
return path.relative_to(root).as_posix()
|
||||
|
||||
|
||||
def read_text(path: Path) -> str | None:
|
||||
try:
|
||||
data = path.read_bytes()
|
||||
if data.startswith((b"\xff\xfe", b"\xfe\xff")):
|
||||
return data.decode("utf-16")
|
||||
return data.decode("utf-8-sig")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
def candidate_files(root: Path) -> list[Path]:
|
||||
command = [
|
||||
"git",
|
||||
"-C",
|
||||
str(root),
|
||||
"ls-files",
|
||||
"--cached",
|
||||
"--others",
|
||||
"--exclude-standard",
|
||||
"-z",
|
||||
]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command,
|
||||
check=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
names = [name for name in result.stdout.decode("utf-8").split("\0") if name]
|
||||
return sorted(root / PurePosixPath(name) for name in names if (root / name).is_file())
|
||||
except (OSError, subprocess.CalledProcessError, UnicodeDecodeError):
|
||||
return sorted(
|
||||
path for path in root.rglob("*") if path.is_file() and ".git" not in path.parts
|
||||
)
|
||||
|
||||
|
||||
def parse_scalar(value: str) -> Any:
|
||||
value = value.split(" #", 1)[0].strip()
|
||||
if not value or value.lower() in {"null", "~"}:
|
||||
return None
|
||||
if value == "[]":
|
||||
return []
|
||||
if value.startswith("[") and value.endswith("]"):
|
||||
inner = value[1:-1].strip()
|
||||
return [] if not inner else [parse_scalar(item) for item in inner.split(",")]
|
||||
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
||||
value = value[1:-1]
|
||||
if value.isdigit():
|
||||
return int(value)
|
||||
return value
|
||||
|
||||
|
||||
def parse_frontmatter(path: Path) -> tuple[dict[str, Any], str, list[str]]:
|
||||
text = read_text(path)
|
||||
if text is None:
|
||||
return {}, "", ["文件不是 UTF-8 文本。"]
|
||||
return parse_frontmatter_text(text)
|
||||
|
||||
|
||||
def parse_frontmatter_text(text: str) -> tuple[dict[str, Any], str, list[str]]:
|
||||
lines = text.splitlines()
|
||||
if not lines or lines[0].strip() != "---":
|
||||
return {}, text, ["缺少起始 frontmatter 分隔符。"]
|
||||
try:
|
||||
end = next(index for index in range(1, len(lines)) if lines[index].strip() == "---")
|
||||
except StopIteration:
|
||||
return {}, text, ["缺少结束 frontmatter 分隔符。"]
|
||||
|
||||
metadata: dict[str, Any] = {}
|
||||
current_list: str | None = None
|
||||
errors: list[str] = []
|
||||
for number, raw in enumerate(lines[1:end], start=2):
|
||||
if not raw.strip() or raw.lstrip().startswith("#"):
|
||||
continue
|
||||
item = re.match(r"^\s+-\s+(.+)$", raw)
|
||||
if item and current_list:
|
||||
metadata[current_list].append(parse_scalar(item.group(1)))
|
||||
continue
|
||||
field = re.match(r"^([A-Za-z_][A-Za-z0-9_-]*):(?:\s*(.*))?$", raw)
|
||||
if not field:
|
||||
errors.append(f"frontmatter 第 {number} 行语法不受支持。")
|
||||
current_list = None
|
||||
continue
|
||||
key, raw_value = field.groups()
|
||||
if key in metadata:
|
||||
errors.append(f"frontmatter 字段重复：{key}。")
|
||||
value = parse_scalar(raw_value or "")
|
||||
if value is None and not (raw_value or "").strip():
|
||||
value = []
|
||||
current_list = key
|
||||
else:
|
||||
current_list = None
|
||||
metadata[key] = value
|
||||
return metadata, "\n".join(lines[end + 1 :]), errors
|
||||
|
||||
|
||||
def is_safe_repo_path(value: str) -> bool:
|
||||
path = PurePosixPath(value)
|
||||
return bool(value) and value == value.strip() and not (
|
||||
path.is_absolute()
|
||||
or ".." in path.parts
|
||||
or "\\" in value
|
||||
or URI_SCHEME.match(value)
|
||||
or "【" in value
|
||||
or any(character in value for character in "*?[]{}")
|
||||
or any(ord(character) < 32 for character in value)
|
||||
)
|
||||
|
||||
|
||||
def normalize_scope(value: str) -> tuple[str, ...]:
|
||||
return tuple(
|
||||
unicodedata.normalize("NFC", part).casefold()
|
||||
for part in PurePosixPath(value.rstrip("/")).parts
|
||||
if part not in {"."}
|
||||
)
|
||||
|
||||
|
||||
def scopes_overlap(left: str, right: str) -> bool:
|
||||
left_parts = normalize_scope(left)
|
||||
right_parts = normalize_scope(right)
|
||||
if not left_parts or not right_parts:
|
||||
return True
|
||||
width = min(len(left_parts), len(right_parts))
|
||||
return left_parts[:width] == right_parts[:width]
|
||||
|
||||
|
||||
def section_content(body: str, heading: str) -> str:
|
||||
pattern = re.compile(
|
||||
rf"(?ms)^##\s+{re.escape(heading)}\s*$\n(.*?)(?=^##\s+|\Z)"
|
||||
)
|
||||
match = pattern.search(body)
|
||||
return "" if match is None else match.group(1).strip()
|
||||
|
||||
|
||||
def validate_tasks(root: Path) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
task_dir = root / "docs" / "tasks"
|
||||
template = task_dir / "_template.md"
|
||||
if template.is_file():
|
||||
metadata, body, errors = parse_frontmatter(template)
|
||||
for message in errors:
|
||||
findings.append(Finding("task-template", relative(template, root), 0, message))
|
||||
missing = sorted(TASK_REQUIRED_FIELDS - set(metadata))
|
||||
if missing:
|
||||
findings.append(
|
||||
Finding(
|
||||
"task-template",
|
||||
relative(template, root),
|
||||
0,
|
||||
"缺少字段：" + ", ".join(missing),
|
||||
)
|
||||
)
|
||||
headings = set(re.findall(r"(?m)^##\s+(.+?)\s*$", body))
|
||||
missing_sections = sorted(TASK_REQUIRED_SECTIONS - headings)
|
||||
if missing_sections:
|
||||
findings.append(
|
||||
Finding(
|
||||
"task-template",
|
||||
relative(template, root),
|
||||
0,
|
||||
"缺少章节：" + ", ".join(missing_sections),
|
||||
)
|
||||
)
|
||||
else:
|
||||
findings.append(Finding("task-template", "docs/tasks/_template.md", 0, "文件不存在。"))
|
||||
|
||||
tasks: dict[str, Task] = {}
|
||||
issue_numbers: dict[int, str] = {}
|
||||
for path in sorted(task_dir.glob("T-*.md")) if task_dir.is_dir() else []:
|
||||
rel = relative(path, root)
|
||||
metadata, body, errors = parse_frontmatter(path)
|
||||
for message in errors:
|
||||
findings.append(Finding("task-frontmatter", rel, 0, message))
|
||||
filename_id = path.stem
|
||||
if not TASK_ID.fullmatch(filename_id):
|
||||
findings.append(Finding("task-id", rel, 0, "文件名必须是 T-<三位编号>[可选小写后缀]。"))
|
||||
missing = sorted(TASK_REQUIRED_FIELDS - set(metadata))
|
||||
if missing:
|
||||
findings.append(
|
||||
Finding("task-frontmatter", rel, 0, "缺少字段：" + ", ".join(missing))
|
||||
)
|
||||
task_id = metadata.get("id")
|
||||
if task_id != filename_id:
|
||||
findings.append(Finding("task-id", rel, 0, "frontmatter id 必须与文件名一致。"))
|
||||
if isinstance(task_id, str) and task_id in tasks:
|
||||
findings.append(Finding("task-id", rel, 0, "任务 ID 重复。"))
|
||||
status = metadata.get("status")
|
||||
if status not in VALID_STATUS:
|
||||
findings.append(Finding("task-status", rel, 0, "status 不在允许枚举中。"))
|
||||
title = metadata.get("title")
|
||||
if not isinstance(title, str) or not title.strip() or "【" in title:
|
||||
findings.append(Finding("task-metadata", rel, 0, "title 必须是已填写的非空字符串。"))
|
||||
phase = metadata.get("phase")
|
||||
if type(phase) is not int or phase < 0:
|
||||
findings.append(Finding("task-metadata", rel, 0, "phase 必须是非负整数。"))
|
||||
created = metadata.get("created")
|
||||
try:
|
||||
if not isinstance(created, str):
|
||||
raise ValueError
|
||||
date.fromisoformat(created)
|
||||
except ValueError:
|
||||
findings.append(Finding("task-metadata", rel, 0, "created 必须是 YYYY-MM-DD。"))
|
||||
deps = metadata.get("deps")
|
||||
if not isinstance(deps, list) or not all(isinstance(dep, str) for dep in deps):
|
||||
findings.append(Finding("task-deps", rel, 0, "deps 必须是任务 ID 数组。"))
|
||||
elif task_id in deps:
|
||||
findings.append(Finding("task-deps", rel, 0, "任务不得依赖自身。"))
|
||||
elif any(not TASK_ID.fullmatch(dep) for dep in deps):
|
||||
findings.append(Finding("task-deps", rel, 0, "deps 含无效任务 ID。"))
|
||||
write_paths = metadata.get("write_paths")
|
||||
if not isinstance(write_paths, list) or not write_paths:
|
||||
findings.append(Finding("task-scope", rel, 0, "write_paths 必须是非空数组。"))
|
||||
else:
|
||||
values = [value for value in write_paths if isinstance(value, str)]
|
||||
if len(values) != len(write_paths) or any(not is_safe_repo_path(value) for value in values):
|
||||
findings.append(Finding("task-scope", rel, 0, "write_paths 含不安全或非字符串路径。"))
|
||||
if len(values) != len(set(values)):
|
||||
findings.append(Finding("task-scope", rel, 0, "write_paths 含重复路径。"))
|
||||
if rel not in values:
|
||||
findings.append(Finding("task-scope", rel, 0, "write_paths 必须包含任务文件自身。"))
|
||||
issue = metadata.get("issue")
|
||||
if issue is not None and (type(issue) is not int or issue <= 0):
|
||||
findings.append(Finding("task-issue", rel, 0, "issue 必须是正整数或 null。"))
|
||||
elif type(issue) is int:
|
||||
if issue in issue_numbers:
|
||||
findings.append(Finding("task-issue", rel, 0, "Issue 编号与其他任务重复。"))
|
||||
issue_numbers[issue] = filename_id
|
||||
context_ref = metadata.get("context_ref")
|
||||
if context_ref is not None and (
|
||||
not isinstance(context_ref, str) or not SHA40.fullmatch(context_ref)
|
||||
):
|
||||
findings.append(Finding("task-claim", rel, 0, "context_ref 必须是 40 位 SHA 或 null。"))
|
||||
claim_branch = metadata.get("claim_branch")
|
||||
if claim_branch is not None and claim_branch != f"claims/{filename_id}":
|
||||
findings.append(Finding("task-claim", rel, 0, "claim_branch 与任务 ID 不一致。"))
|
||||
work_branch = metadata.get("work_branch")
|
||||
if work_branch is not None and (
|
||||
not isinstance(work_branch, str)
|
||||
or not re.fullmatch(rf"agent/[^/]+/{re.escape(filename_id)}", work_branch)
|
||||
):
|
||||
findings.append(Finding("task-claim", rel, 0, "work_branch 格式或任务 ID 不一致。"))
|
||||
if status == "TODO":
|
||||
for key in ("context_ref", "claim_branch", "work_branch"):
|
||||
if metadata.get(key) is not None:
|
||||
findings.append(Finding("task-claim", rel, 0, f"TODO 的 {key} 必须为 null。"))
|
||||
if issue is not None and status in ACTIVE_STATUS:
|
||||
if not isinstance(context_ref, str) or not SHA40.fullmatch(context_ref):
|
||||
findings.append(Finding("task-claim", rel, 0, "Gitea 活跃任务缺少 40 位 context_ref。"))
|
||||
if metadata.get("claim_branch") != f"claims/{filename_id}":
|
||||
findings.append(Finding("task-claim", rel, 0, "claim_branch 与任务 ID 不一致。"))
|
||||
if not isinstance(work_branch, str) or not work_branch.endswith(f"/{filename_id}"):
|
||||
findings.append(Finding("task-claim", rel, 0, "work_branch 与任务 ID 不一致。"))
|
||||
headings = set(re.findall(r"(?m)^##\s+(.+?)\s*$", body))
|
||||
missing_sections = sorted(TASK_REQUIRED_SECTIONS - headings)
|
||||
if missing_sections:
|
||||
findings.append(
|
||||
Finding("task-sections", rel, 0, "缺少章节：" + ", ".join(missing_sections))
|
||||
)
|
||||
if status == "DONE":
|
||||
evidence = section_content(body, "执行记录")
|
||||
if not evidence or "（做完在此记录" in evidence or "【" in evidence:
|
||||
findings.append(Finding("task-evidence", rel, 0, "DONE 缺少真实执行证据。"))
|
||||
if isinstance(task_id, str):
|
||||
tasks[task_id] = Task(path, metadata, body)
|
||||
|
||||
for task_id, task in sorted(tasks.items()):
|
||||
rel = relative(task.path, root)
|
||||
deps = task.metadata.get("deps")
|
||||
if not isinstance(deps, list):
|
||||
continue
|
||||
for dep in deps:
|
||||
if dep not in tasks:
|
||||
findings.append(Finding("task-deps", rel, 0, f"依赖任务不存在：{dep}。"))
|
||||
elif task.status != "TODO" and tasks[dep].status != "DONE":
|
||||
findings.append(Finding("task-deps", rel, 0, f"非 TODO 任务依赖尚未 DONE：{dep}。"))
|
||||
|
||||
visiting: set[str] = set()
|
||||
visited: set[str] = set()
|
||||
|
||||
def visit(task_id: str) -> None:
|
||||
if task_id in visiting:
|
||||
findings.append(
|
||||
Finding("task-deps", relative(tasks[task_id].path, root), 0, "依赖图存在环。")
|
||||
)
|
||||
return
|
||||
if task_id in visited:
|
||||
return
|
||||
visiting.add(task_id)
|
||||
deps = tasks[task_id].metadata.get("deps")
|
||||
if isinstance(deps, list):
|
||||
for dep in deps:
|
||||
if dep in tasks:
|
||||
visit(dep)
|
||||
visiting.remove(task_id)
|
||||
visited.add(task_id)
|
||||
|
||||
for task_id in sorted(tasks):
|
||||
visit(task_id)
|
||||
|
||||
active = [task for task in tasks.values() if task.status in ACTIVE_STATUS]
|
||||
for index, left in enumerate(sorted(active, key=lambda task: task.task_id)):
|
||||
left_paths = left.metadata.get("write_paths", [])
|
||||
if not isinstance(left_paths, list):
|
||||
continue
|
||||
for right in sorted(active, key=lambda task: task.task_id)[index + 1 :]:
|
||||
right_paths = right.metadata.get("write_paths", [])
|
||||
if not isinstance(right_paths, list):
|
||||
continue
|
||||
if any(
|
||||
isinstance(a, str) and isinstance(b, str) and scopes_overlap(a, b)
|
||||
for a in left_paths
|
||||
for b in right_paths
|
||||
):
|
||||
findings.append(
|
||||
Finding(
|
||||
"task-scope-overlap",
|
||||
relative(right.path, root),
|
||||
0,
|
||||
f"活跃任务与 {left.task_id} 的 write_paths 重叠。",
|
||||
)
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def markdown_targets(text: str) -> Iterable[tuple[int, str]]:
|
||||
in_fence = False
|
||||
for line_number, line in enumerate(text.splitlines(), start=1):
|
||||
stripped = line.lstrip()
|
||||
if stripped.startswith("```") or stripped.startswith("~~~"):
|
||||
in_fence = not in_fence
|
||||
continue
|
||||
if in_fence:
|
||||
continue
|
||||
for match in MARKDOWN_LINK.finditer(line):
|
||||
yield line_number, match.group(1).strip()
|
||||
|
||||
|
||||
def clean_link_target(raw: str) -> str | None:
|
||||
if raw.startswith("<") and ">" in raw:
|
||||
target = raw[1 : raw.index(">")]
|
||||
else:
|
||||
target = raw.split(maxsplit=1)[0]
|
||||
target = urllib.parse.unquote(target).split("#", 1)[0].split("?", 1)[0]
|
||||
if (
|
||||
not target
|
||||
or target.startswith("#")
|
||||
or target.startswith("//")
|
||||
or URI_SCHEME.match(target)
|
||||
or "【" in target
|
||||
):
|
||||
return None
|
||||
return target
|
||||
|
||||
|
||||
def validate_markdown_links(root: Path, files: list[Path]) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
for path in files:
|
||||
if path.suffix.lower() != ".md":
|
||||
continue
|
||||
text = read_text(path)
|
||||
if text is None:
|
||||
continue
|
||||
for line, raw in markdown_targets(text):
|
||||
target = clean_link_target(raw)
|
||||
if target is None:
|
||||
continue
|
||||
resolved = root / target.lstrip("/") if target.startswith("/") else path.parent / target
|
||||
try:
|
||||
resolved.resolve().relative_to(root.resolve())
|
||||
except ValueError:
|
||||
findings.append(
|
||||
Finding("markdown-link", relative(path, root), line, "链接逃出仓库根目录。")
|
||||
)
|
||||
continue
|
||||
if not resolved.exists():
|
||||
findings.append(
|
||||
Finding("markdown-link", relative(path, root), line, "本地链接目标不存在。")
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def validate_navigation(root: Path) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
root_readme = read_text(root / "README.md") or ""
|
||||
docs_readme = read_text(root / "docs" / "README.md") or ""
|
||||
for doc in sorted((root / "docs").glob("*.md")):
|
||||
root_target = f"docs/{doc.name}"
|
||||
if root_target not in root_readme:
|
||||
findings.append(Finding("navigation", "README.md", 0, f"未登记 {root_target}。"))
|
||||
if doc.name != "README.md" and f"({doc.name})" not in docs_readme:
|
||||
findings.append(
|
||||
Finding("navigation", "docs/README.md", 0, f"未登记 {doc.name}。")
|
||||
)
|
||||
|
||||
required_root_entries = (
|
||||
"scripts/validate_agent_context.py",
|
||||
"scripts/setup_gitea_labels.py",
|
||||
"scripts/validate_harness_governance.py",
|
||||
"scripts/audit_gitea_coordination.py",
|
||||
"scripts/test_gitea_claim_race.py",
|
||||
"tests/test_governance.py",
|
||||
".gitea/ISSUE_TEMPLATE/task.md",
|
||||
".gitea/PULL_REQUEST_TEMPLATE.md",
|
||||
".gitea/workflows/harness-governance.yml",
|
||||
)
|
||||
for entry in required_root_entries:
|
||||
if entry not in root_readme:
|
||||
findings.append(Finding("navigation", "README.md", 0, f"未登记 {entry}。"))
|
||||
return findings
|
||||
|
||||
|
||||
def safe_token_assignment(value: str) -> bool:
|
||||
value = value.strip().rstrip(",}").strip().strip("'\"")
|
||||
upper = value.upper()
|
||||
return (
|
||||
not value
|
||||
or value.startswith(("【", "<"))
|
||||
or SAFE_VARIABLE_REFERENCE.fullmatch(value) is not None
|
||||
or upper in {"REPLACE", "CHANGEME", "EXAMPLE"}
|
||||
or upper.startswith(("REPLACE_", "CHANGEME_", "EXAMPLE_"))
|
||||
)
|
||||
|
||||
|
||||
def validate_secrets(root: Path, files: list[Path]) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
for path in files:
|
||||
rel = relative(path, root)
|
||||
lower_name = path.name.lower()
|
||||
if lower_name == "gitea.env" or (
|
||||
lower_name.startswith("gitea.env.") and lower_name != "gitea.env.example"
|
||||
):
|
||||
findings.append(Finding("secret-file", rel, 0, "私有 Gitea 环境文件不得被跟踪。"))
|
||||
text = read_text(path)
|
||||
if text is None:
|
||||
if path.suffix.lower() in KNOWN_TEXT_SUFFIXES or lower_name in {
|
||||
".env",
|
||||
"dockerfile",
|
||||
"makefile",
|
||||
}:
|
||||
findings.append(
|
||||
Finding("secret-scan", rel, 0, "已跟踪文本无法安全解码并扫描。")
|
||||
)
|
||||
continue
|
||||
for line_number, line in enumerate(text.splitlines(), start=1):
|
||||
token_assignment = TOKEN_ASSIGNMENT.match(line)
|
||||
cmd_assignment = CMD_TOKEN_ASSIGNMENT.match(line)
|
||||
rules = []
|
||||
if token_assignment and not safe_token_assignment(token_assignment.group(1)):
|
||||
rules.append("GITEA_TOKEN 实值")
|
||||
if cmd_assignment:
|
||||
cmd_value = next(
|
||||
(value for value in cmd_assignment.groups() if value is not None),
|
||||
"",
|
||||
)
|
||||
if not safe_token_assignment(cmd_value):
|
||||
rules.append("Windows 命令 Token 实值")
|
||||
if AUTH_VALUE.search(line):
|
||||
rules.append("Authorization 实值")
|
||||
if URL_CREDENTIAL.search(line):
|
||||
rules.append("URL 内嵌凭据")
|
||||
if GITEA_TOKEN_LITERAL.search(line):
|
||||
rules.append("Gitea Token 字面值")
|
||||
for rule in rules:
|
||||
findings.append(Finding("secret-value", rel, line_number, f"检测到{rule}。"))
|
||||
for match in DOTNET_TOKEN_SETTER.finditer(text):
|
||||
if not safe_token_assignment(match.group(2)):
|
||||
line_number = text.count("\n", 0, match.start()) + 1
|
||||
findings.append(
|
||||
Finding(
|
||||
"secret-value",
|
||||
rel,
|
||||
line_number,
|
||||
"检测到 .NET 环境变量 Token 实值。",
|
||||
)
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def require_markers(root: Path, path_string: str, markers: Iterable[str]) -> list[Finding]:
|
||||
path = root / PurePosixPath(path_string)
|
||||
if not path.is_file():
|
||||
return [Finding("required-artifact", path_string, 0, "文件不存在。")]
|
||||
text = read_text(path) or ""
|
||||
return [
|
||||
Finding("required-artifact", path_string, 0, f"缺少标记：{marker}。")
|
||||
for marker in markers
|
||||
if marker not in text
|
||||
]
|
||||
|
||||
|
||||
def validate_gitea_artifacts(root: Path) -> list[Finding]:
|
||||
findings = []
|
||||
findings.extend(
|
||||
require_markers(
|
||||
root,
|
||||
".gitea/ISSUE_TEMPLATE/task.md",
|
||||
("task_id:", "task_file:", "context_ref:", "write_paths:", "lease_until:"),
|
||||
)
|
||||
)
|
||||
findings.extend(
|
||||
require_markers(
|
||||
root,
|
||||
".gitea/PULL_REQUEST_TEMPLATE.md",
|
||||
("Closes #", "task_file:", "context_ref:", "write_paths:", "验证证据"),
|
||||
)
|
||||
)
|
||||
workflow = ".gitea/workflows/harness-governance.yml"
|
||||
findings.extend(
|
||||
require_markers(
|
||||
root,
|
||||
workflow,
|
||||
(
|
||||
"push:",
|
||||
"pull_request:",
|
||||
"actions/checkout@v4",
|
||||
"permissions: read-all",
|
||||
"persist-credentials: false",
|
||||
"python -m unittest discover",
|
||||
"python scripts/validate_harness_governance.py",
|
||||
),
|
||||
)
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def validate_repository(root: Path) -> list[Finding]:
|
||||
files = candidate_files(root)
|
||||
findings = [
|
||||
Finding("agent-context", "docs/agent-context.json", 0, message)
|
||||
for message in validate_manifest(root)
|
||||
]
|
||||
findings.extend(validate_navigation(root))
|
||||
findings.extend(validate_markdown_links(root, files))
|
||||
findings.extend(validate_tasks(root))
|
||||
findings.extend(validate_secrets(root, files))
|
||||
findings.extend(validate_gitea_artifacts(root))
|
||||
return sorted(set(findings))
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="离线校验 Harness Coding 仓库治理工件。")
|
||||
parser.add_argument(
|
||||
"--root",
|
||||
type=Path,
|
||||
default=Path(__file__).resolve().parents[1],
|
||||
help="仓库根目录；默认取脚本上一级。",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
root = args.root.resolve()
|
||||
if not root.is_dir():
|
||||
print("ERROR: 仓库根目录不存在。", file=sys.stderr)
|
||||
return 2
|
||||
findings = validate_repository(root)
|
||||
if findings:
|
||||
for finding in findings:
|
||||
print(finding.render(), file=sys.stderr)
|
||||
print(f"治理校验失败：{len(findings)} 项不一致。", file=sys.stderr)
|
||||
return 1
|
||||
print("治理校验通过：上下文、导航、链接、任务、模板、工作流与敏感信息均一致。")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user