mirror of
https://github.com/yeasy/docker_practice.git
synced 2026-08-10 08:27:25 +00:00
yeasy
f82e13017d
起因：blockchain_guide 上曾有一个 claude/novel-content-extraction-r0wys3 分支，4 个提交的 author 和 committer 都是 Claude <noreply@anthropic.com> （内容还是与本书无关的小说抓取脚本）。分支已删除，但暴露出一个缺口： 既有的 .git/hooks/commit-msg 只查提交信息里的 Co-authored-by 尾注， 查不到身份字段；而且本地钩子不随仓库分发，那 4 个提交来自云端会话， 根本不经过本地钩子。 因此把闸门放到服务端：本工作流在 push 与 pull_request 上扫描本次新增 提交的 author/committer 身份与 Co-authored-by 尾注，命中即失败。 本地另配 pre-commit（拦当前身份）与 pre-push（拦 cherry-pick/rebase 带进来的外来提交）作为纵深防御，二者不入库。 判定刻意收窄，避免误伤本仓库群里大量讨论 Claude/Codex 的正文： 只检查身份字段与行首锚定的 Co-authored-by 尾注，绝不扫描自由文本； 姓名要求全等（"Claude Dubois" 这样的真人不受影响），邮箱按厂商域名 （含子域）匹配，并覆盖 claude[bot] 这类 GitHub App 身份。 已验证： - 用被删分支的真实 4 个提交做回归，push / pull_request / 新建分支 三种到达形态全部拦下； - 全部 14 个仓库的完整历史（5,715 个提交，含 docker_practice 1,591 个多人历史与 claude_guide 339 个满是 Claude 的提交）零误报； - dependabot 的 Co-authored-by、真人 Claude Dubois、正文提到 co-authored-by 但非尾注的提交，均正确放行； - 范围解析不出来时 fail closed，只有确实没有 head 提交（删分支）才放行； - tags-ignore 避免发布 tag 触发一次重复的全量审计。
416 lines
16 KiB
Go
416 lines
16 KiB
Go
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
WORKFLOW_DIR = ROOT / ".github" / "workflows"
|
|
FULL_ACTION_SHA = re.compile(r"^[^@\s]+@[0-9a-f]{40}$")
|
|
TEST_REPOSITORY = "owner/repo"
|
|
TEST_SHA = "a" * 40
|
|
GET_REF_COMMAND = [
|
|
"api",
|
|
"--include",
|
|
"--method",
|
|
"GET",
|
|
f"repos/{TEST_REPOSITORY}/git/ref/tags/preview-pdf",
|
|
]
|
|
PATCH_REF_COMMAND = [
|
|
"api",
|
|
"--silent",
|
|
"--method",
|
|
"PATCH",
|
|
f"repos/{TEST_REPOSITORY}/git/refs/tags/preview-pdf",
|
|
"--raw-field",
|
|
f"sha={TEST_SHA}",
|
|
"--field",
|
|
"force=true",
|
|
]
|
|
POST_REF_COMMAND = [
|
|
"api",
|
|
"--silent",
|
|
"--method",
|
|
"POST",
|
|
f"repos/{TEST_REPOSITORY}/git/refs",
|
|
"--raw-field",
|
|
"ref=refs/tags/preview-pdf",
|
|
"--raw-field",
|
|
f"sha={TEST_SHA}",
|
|
]
|
|
EDIT_RELEASE_COMMAND = [
|
|
"release",
|
|
"edit",
|
|
"preview-pdf",
|
|
"--title",
|
|
"Latest Preview PDF",
|
|
"--notes-file",
|
|
"dist/release-notes.md",
|
|
"--prerelease",
|
|
]
|
|
VIEW_RELEASE_COMMAND = ["release", "view", "preview-pdf"]
|
|
CREATE_RELEASE_COMMAND = [
|
|
"release",
|
|
"create",
|
|
"preview-pdf",
|
|
"--title",
|
|
"Latest Preview PDF",
|
|
"--notes-file",
|
|
"dist/release-notes.md",
|
|
"--prerelease",
|
|
"--latest=false",
|
|
"--verify-tag",
|
|
]
|
|
|
|
FAKE_GH = r'''#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
args = sys.argv[1:]
|
|
with open(os.environ["GH_LOG"], "a", encoding="utf-8") as stream:
|
|
stream.write(json.dumps(args) + "\n")
|
|
|
|
scenario = os.environ["GH_SCENARIO"]
|
|
repository = "owner/repo"
|
|
sha = "a" * 40
|
|
reasons = {
|
|
"401": "Unauthorized",
|
|
"403": "Forbidden",
|
|
"404": "Not Found",
|
|
"429": "Too Many Requests",
|
|
"503": "Service Unavailable",
|
|
}
|
|
|
|
def fail_http(code):
|
|
print(f"HTTP/2.0 {code} {reasons[code]}")
|
|
print(f"fake gh HTTP {code}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
get_ref = ["api", "--include", "--method", "GET", f"repos/{repository}/git/ref/tags/preview-pdf"]
|
|
patch_ref = [
|
|
"api", "--silent", "--method", "PATCH",
|
|
f"repos/{repository}/git/refs/tags/preview-pdf",
|
|
"--raw-field", f"sha={sha}", "--field", "force=true",
|
|
]
|
|
post_ref = [
|
|
"api", "--silent", "--method", "POST", f"repos/{repository}/git/refs",
|
|
"--raw-field", "ref=refs/tags/preview-pdf", "--raw-field", f"sha={sha}",
|
|
]
|
|
edit_release = [
|
|
"release", "edit", "preview-pdf", "--title", "Latest Preview PDF",
|
|
"--notes-file", "dist/release-notes.md", "--prerelease",
|
|
]
|
|
create_release = [
|
|
"release", "create", "preview-pdf", "--title", "Latest Preview PDF",
|
|
"--notes-file", "dist/release-notes.md", "--prerelease",
|
|
"--latest=false", "--verify-tag",
|
|
]
|
|
view_release = ["release", "view", "preview-pdf"]
|
|
|
|
if os.environ.get("GH_REPO") != repository:
|
|
print("fake gh requires explicit GH_REPO", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
|
|
if args == get_ref:
|
|
if scenario.startswith("ref_network"):
|
|
print("fake gh network failure", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
for code in reasons:
|
|
if scenario.startswith(f"ref_{code}"):
|
|
fail_http(code)
|
|
print("HTTP/2.0 200 OK")
|
|
print('Content-Type: application/json\n\n{"ref":"refs/tags/preview-pdf"}')
|
|
raise SystemExit(0)
|
|
|
|
if args in (patch_ref, post_ref, edit_release, create_release):
|
|
raise SystemExit(0)
|
|
|
|
if args == view_release:
|
|
if "release_missing" in scenario:
|
|
print("release not found", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
if "release_network" in scenario:
|
|
print("fake release network failure", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
for code in reasons:
|
|
if f"release_{code}" in scenario:
|
|
print(f"fake release HTTP {code}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
raise SystemExit(0)
|
|
|
|
print(f"unexpected gh argv: {args!r}", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
'''
|
|
|
|
|
|
def workflow_step_script(workflow_text, step_name):
|
|
marker = f" - name: {step_name}\n"
|
|
start = workflow_text.index(marker) + len(marker)
|
|
run_marker = " run: |\n"
|
|
script_start = workflow_text.index(run_marker, start) + len(run_marker)
|
|
script_end = workflow_text.find("\n - name:", script_start)
|
|
if script_end < 0:
|
|
script_end = len(workflow_text)
|
|
return textwrap.dedent(workflow_text[script_start:script_end])
|
|
|
|
|
|
def workflow_step_scripts_in_document_order(workflow_text, step_names):
|
|
ordered_names = sorted(
|
|
step_names,
|
|
key=lambda name: workflow_text.index(f" - name: {name}\n"),
|
|
)
|
|
return tuple(workflow_step_script(workflow_text, name) for name in ordered_names)
|
|
|
|
|
|
class WorkflowSecurityTests(unittest.TestCase):
|
|
@staticmethod
|
|
def workflows():
|
|
return sorted(WORKFLOW_DIR.glob("*.y*ml"))
|
|
|
|
def test_all_actions_are_immutable_with_version_comments(self):
|
|
failures = []
|
|
for workflow in self.workflows():
|
|
for number, line in enumerate(workflow.read_text(encoding="utf-8").splitlines(), 1):
|
|
match = re.search(r"\buses:\s*([^\s#]+)(?:\s+#\s*(\S+))?", line)
|
|
if not match:
|
|
continue
|
|
action, version = match.groups()
|
|
if not FULL_ACTION_SHA.fullmatch(action) or not version or not version.startswith("v"):
|
|
failures.append(f"{workflow.name}:{number}: {line.strip()}")
|
|
self.assertEqual(failures, [])
|
|
|
|
def test_checkout_never_persists_credentials(self):
|
|
failures = []
|
|
for workflow in self.workflows():
|
|
lines = workflow.read_text(encoding="utf-8").splitlines()
|
|
for index, line in enumerate(lines):
|
|
if "uses: actions/checkout@" not in line:
|
|
continue
|
|
step = "\n".join(lines[index : index + 8])
|
|
if "persist-credentials: false" not in step:
|
|
failures.append(f"{workflow.name}:{index + 1}")
|
|
self.assertEqual(failures, [])
|
|
|
|
def test_every_workflow_declares_permissions(self):
|
|
failures = []
|
|
for workflow in self.workflows():
|
|
text = workflow.read_text(encoding="utf-8")
|
|
before_jobs = text.split("\njobs:", 1)[0]
|
|
if not re.search(r"(?m)^permissions:", before_jobs):
|
|
failures.append(workflow.name)
|
|
self.assertEqual(failures, [])
|
|
|
|
def test_release_and_preview_separate_read_only_builds_from_writes(self):
|
|
expectations = {
|
|
"auto-release.yml": ("release",),
|
|
"preview-pdf.yml": ("publish",),
|
|
}
|
|
for name, write_jobs in expectations.items():
|
|
text = (WORKFLOW_DIR / name).read_text(encoding="utf-8")
|
|
self.assertRegex(text, r"(?ms)^ build:\n permissions:\n contents: read\b", name)
|
|
for job in write_jobs:
|
|
self.assertRegex(
|
|
text,
|
|
rf"(?ms)^ {job}:.*?permissions:\n contents: write\b.*?needs: build\b",
|
|
name,
|
|
)
|
|
|
|
def test_preview_publish_steps_follow_tag_release_asset_order(self):
|
|
preview = (WORKFLOW_DIR / "preview-pdf.yml").read_text(encoding="utf-8")
|
|
synchronize = preview.index(" - name: Synchronize mutable preview tag\n")
|
|
release = preview.index(" - name: Create or update preview release\n")
|
|
upload = preview.index(" - name: Replace preview assets\n")
|
|
|
|
self.assertLess(synchronize, release)
|
|
self.assertLess(release, upload)
|
|
|
|
def run_preview_scripts(
|
|
self,
|
|
scenario,
|
|
*,
|
|
repository=TEST_REPOSITORY,
|
|
sha=TEST_SHA,
|
|
):
|
|
preview = (WORKFLOW_DIR / "preview-pdf.yml").read_text(encoding="utf-8")
|
|
scripts = workflow_step_scripts_in_document_order(
|
|
preview,
|
|
(
|
|
"Synchronize mutable preview tag",
|
|
"Create or update preview release",
|
|
),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
fake_gh = root / "gh"
|
|
fake_gh.write_text(FAKE_GH, encoding="utf-8")
|
|
fake_gh.chmod(0o755)
|
|
log = root / "commands.jsonl"
|
|
env = os.environ.copy()
|
|
env.update(
|
|
{
|
|
"PATH": f"{root}:{env.get('PATH', '')}",
|
|
"GH_LOG": str(log),
|
|
"GH_SCENARIO": scenario,
|
|
"GH_TOKEN": "test-token",
|
|
"GH_REPO": repository,
|
|
"GITHUB_REPOSITORY": repository,
|
|
"GITHUB_SHA": sha,
|
|
}
|
|
)
|
|
result = None
|
|
for script in scripts:
|
|
result = subprocess.run(
|
|
["/bin/bash", "-c", script],
|
|
cwd=ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
break
|
|
commands = []
|
|
if log.exists():
|
|
commands = [
|
|
json.loads(line)
|
|
for line in log.read_text(encoding="utf-8").splitlines()
|
|
]
|
|
return result, commands
|
|
|
|
def test_mutable_preview_updates_existing_tag_and_release(self):
|
|
result, commands = self.run_preview_scripts("ref_200_release_exists")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(
|
|
commands,
|
|
[GET_REF_COMMAND, PATCH_REF_COMMAND, VIEW_RELEASE_COMMAND, EDIT_RELEASE_COMMAND],
|
|
)
|
|
|
|
def test_mutable_preview_creates_only_on_explicit_not_found(self):
|
|
result, commands = self.run_preview_scripts("ref_404_release_missing")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(
|
|
commands,
|
|
[GET_REF_COMMAND, POST_REF_COMMAND, VIEW_RELEASE_COMMAND, CREATE_RELEASE_COMMAND],
|
|
)
|
|
flattened = [argument for command in commands for argument in command]
|
|
self.assertNotIn("--target", flattened)
|
|
|
|
def test_preview_rejects_invalid_repository_and_sha_before_calling_gh(self):
|
|
cases = (
|
|
{"repository": "owner/repo/extra", "sha": TEST_SHA, "message": "Invalid GITHUB_REPOSITORY"},
|
|
{"repository": TEST_REPOSITORY, "sha": "a" * 39, "message": "Invalid GITHUB_SHA"},
|
|
)
|
|
for case in cases:
|
|
with self.subTest(case=case):
|
|
result, commands = self.run_preview_scripts(
|
|
"ref_200_release_exists",
|
|
repository=case["repository"],
|
|
sha=case["sha"],
|
|
)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(commands, [])
|
|
self.assertIn(case["message"], result.stderr)
|
|
|
|
def test_preview_tag_lookup_fails_closed_on_non_404_errors(self):
|
|
for scenario in ("ref_401", "ref_403", "ref_429", "ref_503", "ref_network"):
|
|
with self.subTest(scenario=scenario):
|
|
result, commands = self.run_preview_scripts(scenario)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(commands, [GET_REF_COMMAND])
|
|
expected = "network failure" if scenario.endswith("network") else scenario.removeprefix("ref_")
|
|
self.assertIn(expected, result.stderr)
|
|
|
|
def test_preview_release_lookup_fails_closed_except_exact_not_found(self):
|
|
scenarios = (
|
|
"ref_200_release_401",
|
|
"ref_200_release_403",
|
|
"ref_200_release_404",
|
|
"ref_200_release_429",
|
|
"ref_200_release_503",
|
|
"ref_200_release_network",
|
|
)
|
|
for scenario in scenarios:
|
|
with self.subTest(scenario=scenario):
|
|
result, commands = self.run_preview_scripts(scenario)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(
|
|
commands,
|
|
[GET_REF_COMMAND, PATCH_REF_COMMAND, VIEW_RELEASE_COMMAND],
|
|
)
|
|
expected = "network failure" if scenario.endswith("network") else scenario.rsplit("release_", 1)[1]
|
|
self.assertIn(expected, result.stderr)
|
|
|
|
def test_preview_publish_has_explicit_repo_context_only_in_write_job(self):
|
|
preview = (WORKFLOW_DIR / "preview-pdf.yml").read_text(encoding="utf-8")
|
|
build, publish = preview.split("\n publish:\n", 1)
|
|
|
|
self.assertNotIn("GH_REPO", build)
|
|
self.assertIn("GH_REPO: ${{ github.repository }}", publish)
|
|
|
|
def test_downloads_dependencies_and_link_checker_are_integrity_pinned(self):
|
|
combined = "\n".join(path.read_text(encoding="utf-8") for path in self.workflows())
|
|
link_text = (WORKFLOW_DIR / "check-link.yml").read_text(encoding="utf-8")
|
|
|
|
self.assertNotRegex(combined, r"npm install\s+-g\s+@mermaid-js/mermaid-cli")
|
|
self.assertIn("PANDOC_SHA256", combined)
|
|
self.assertIn("KUBECONFORM_SHA256", combined)
|
|
self.assertIn("ACTIONLINT_SHA256", combined)
|
|
self.assertRegex(link_text, r"dkhamsing/awesome_bot@sha256:[0-9a-f]{64}")
|
|
|
|
def test_mermaid_is_exact_and_lockfile_backed(self):
|
|
package = json.loads((ROOT / "package.json").read_text(encoding="utf-8"))
|
|
version = package["devDependencies"]["@mermaid-js/mermaid-cli"]
|
|
|
|
self.assertRegex(version, r"^\d+\.\d+\.\d+$")
|
|
self.assertTrue((ROOT / "package-lock.json").is_file())
|
|
ignored = {
|
|
line.strip()
|
|
for line in (ROOT / ".gitignore").read_text(encoding="utf-8").splitlines()
|
|
if line.strip() and not line.lstrip().startswith("#")
|
|
}
|
|
self.assertNotIn("package-lock.json", ignored)
|
|
self.assertTrue(all("npm ci" in path.read_text(encoding="utf-8") for path in self.workflows() if path.name not in {"check-link.yml", "dependabot-automerge.yml", "identity-guard.yaml"}))
|
|
|
|
def test_artifacts_are_smoke_tested_and_html_failures_are_not_silent(self):
|
|
verifier = ROOT / "tools" / "verify_artifacts.py"
|
|
self.assertTrue(verifier.is_file())
|
|
for name in ("auto-release.yml", "ci.yaml", "preview-pdf.yml"):
|
|
text = (WORKFLOW_DIR / name).read_text(encoding="utf-8")
|
|
self.assertIn("tools/verify_artifacts.py", text, name)
|
|
self.assertIn("SHA256SUMS", text, name)
|
|
auto_release = (WORKFLOW_DIR / "auto-release.yml").read_text(encoding="utf-8")
|
|
self.assertNotIn("continue-on-error: true", auto_release)
|
|
|
|
def test_tagged_release_attests_every_formal_artifact_with_minimum_permissions(self):
|
|
release = (WORKFLOW_DIR / "auto-release.yml").read_text(encoding="utf-8")
|
|
preview = (WORKFLOW_DIR / "preview-pdf.yml").read_text(encoding="utf-8")
|
|
build_part, release_job = release.split("\n release:\n", 1)
|
|
|
|
self.assertIn("contents: write", release_job)
|
|
self.assertIn("id-token: write", release_job)
|
|
self.assertIn("attestations: write", release_job)
|
|
self.assertIn(
|
|
"actions/attest-build-provenance@0f67c3f4856b2e3261c31976d6725780e5e4c373 # v4.1.1",
|
|
release_job,
|
|
)
|
|
self.assertIn("subject-path: |", release_job)
|
|
self.assertIn("dist/docker_practice-*.pdf", release_job)
|
|
self.assertIn("dist/docker_practice-*.html", release_job)
|
|
self.assertIn("dist/SHA256SUMS", release_job)
|
|
self.assertNotIn("attest-build-provenance", build_part)
|
|
self.assertNotIn("attestations: write", build_part)
|
|
self.assertNotIn("id-token: write", build_part)
|
|
self.assertNotIn("attest-build-provenance", preview)
|
|
self.assertIn("mutable preview", preview.lower())
|
|
self.assertIn("tagged release", preview.lower())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|