2026-07-14 13:05:19 +08:00
|
|
|
|
#!/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",
|
2026-07-16 21:18:29 +08:00
|
|
|
|
("task_id:", "task_file:", "context_ref:", "write_paths:", "lease_until:"),
|
2026-07-14 13:05:19 +08:00
|
|
|
|
)
|
|
|
|
|
|
)
|
|
|
|
|
|
findings.extend(
|
|
|
|
|
|
require_markers(
|
|
|
|
|
|
root,
|
|
|
|
|
|
".gitea/PULL_REQUEST_TEMPLATE.md",
|
2026-07-16 21:18:29 +08:00
|
|
|
|
("Closes #", "task_file:", "context_ref:", "write_paths:", "验证证据"),
|
2026-07-14 13:05:19 +08:00
|
|
|
|
)
|
|
|
|
|
|
)
|
|
|
|
|
|
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())
|