github/dingtalk-workspace-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(eval): bind dispatch markers to workflow artifacts

Browse Source 
Bind each accepted marker to the exact workflow run attempt, immutable artifact, source comment, and current PR head so a historical successful run cannot authorize a different payload.
This commit is contained in:
chichuan
2026-08-11 21:33:10 +08:00
parent aebb75371b
commit 6ddda6f1bf
4 changed files with 1502 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/eval-dispatch.yml
+209 -27
View File
@@ -1,16 +1,17 @@
name: PR Eval Dispatch
# `/eval <products> [sha=<full-head-sha>] [cases=<ref>]` PR 评论 → 触发内网评测流水线，报告由内网 bot 回贴。
# `/eval <products> [sha=<full-head-sha>] [cases=<ref>]` PR 评论 → 生成可验证的评测请求，报告由 bot 回贴。
# 本 workflow 只在默认分支上下文运行，不 checkout、不执行 PR 代码。
# 审核 SHA 规则：评测他人 PR 必须显式携带 sha=（审阅背书凭据，验证
# 其恰为当前 open head）；评测自己创建的 PR 可省略，自动钉住派发时刻
# 的当前 head（作者自背书，无第三方偷换窗口）；内网 CI 另以
# 的当前 head（作者自背书，无第三方偷换窗口）；受控评测执行端另以
# FETCH_HEAD 校验兜底派发后的变更。
# 授权两级：仓库 write/maintain/admin 可派发任意 PR；默认分支
# .github/eval-allowlist.txt 名单内的用户仅可派发自己创建的 PR。
# 触发通道：workflow 校验通过后发结构化评论（HTML 注释内嵌 JSON），
# 内网 Devix 轮询服务每 3 分钟扫描 → 触发 Aone CI 流水线。
# 不再直接 curl 内网（公网 runner 结构性不可达）。
# 触发通道：workflow 先创建占位评论，再上传与本次 run/comment 绑定的
# 不可变 manifest artifact，最后把 artifact 指针写回同一评论。评论仅是
# 不可信通知；受控评测服务必须验证成功 run、artifact 与 manifest，并在
# 触发评测前原子占用 manifest.idempotency_key，重复占用只能 no-op。
on:
issue_comment:
@@ -93,36 +94,217 @@ jobs:
| python3 scripts/ci/eval_dispatch_guard.py head \
>> "$GITHUB_OUTPUT"
- name: Post dispatch marker and acknowledge
- name: Create dispatch placeholder
id: placeholder
env:
GH_TOKEN: ${{ github.token }}
PR_NUMBER: ${{ github.event.issue.number }}
run: |
set -euo pipefail
placeholder_body="🛰️ /eval 请求已通过权限与版本校验，正在生成可验证的评测请求。"
response="$(
jq -n --arg body "$placeholder_body" '{body: $body}' \
| curl --fail --silent --show-error \
-X POST \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
--data @- \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/comments"
)"
comment_id="$(
printf '%s' "$response" \
| jq -er \
--arg issue_url "https://api.github.com/repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}" \
'select(.issue_url == $issue_url) | .id | tostring | select(test("^[1-9][0-9]*$"))'
)"
printf 'comment_id=%s\n' "$comment_id" >> "$GITHUB_OUTPUT"
- name: Build dispatch request manifest
env:
REPOSITORY_ID: '1187709537'
REPOSITORY: ${{ github.repository }}
WORKFLOW_ID: '331725458'
WORKFLOW_PATH: .github/workflows/eval-dispatch.yml
RUN_ID: ${{ github.run_id }}
RUN_ATTEMPT: ${{ github.run_attempt }}
SOURCE_COMMENT_ID: ${{ github.event.comment.id }}
DISPATCH_COMMENT_ID: ${{ steps.placeholder.outputs.comment_id }}
ACTOR_ID: ${{ github.event.comment.user.id }}
ACTOR_LOGIN: ${{ github.event.comment.user.login }}
PR_NUMBER: ${{ github.event.issue.number }}
PR_HEAD_SHA: ${{ steps.pr.outputs.head_sha }}
PRODUCTS: ${{ steps.parse.outputs.products }}
CASES_REF: ${{ steps.parse.outputs.cases_ref }}
RUN_ID: ${{ github.run_id }}
SOURCE_BODY: ${{ github.event.comment.body }}
MANIFEST_PATH: ${{ runner.temp }}/eval-dispatch-request.json
run: |
set -euo pipefail
if [ "$REPOSITORY" != "DingTalk-Real-AI/dingtalk-workspace-cli" ]; then
echo "unexpected repository: ${REPOSITORY}" >&2
exit 1
fi
for value in \
"$REPOSITORY_ID" \
"$WORKFLOW_ID" \
"$RUN_ID" \
"$RUN_ATTEMPT" \
"$SOURCE_COMMENT_ID" \
"$DISPATCH_COMMENT_ID" \
"$ACTOR_ID" \
"$PR_NUMBER"; do
if [[ ! "$value" =~ ^[1-9][0-9]*$ ]]; then
echo "dispatch manifest contains a non-canonical identifier" >&2
exit 1
fi
done
if [[ ! "$PR_HEAD_SHA" =~ ^[0-9a-f]{40}$ ]]; then
echo "dispatch manifest contains an invalid PR head SHA" >&2
exit 1
fi
hash_output="$(printf '%s' "$SOURCE_BODY" | sha256sum)"
source_body_sha256="${hash_output%% *}"
if [[ ! "$source_body_sha256" =~ ^[0-9a-f]{64}$ ]]; then
echo "failed to hash source comment" >&2
exit 1
fi
idempotency_key="${REPOSITORY_ID}:${SOURCE_COMMENT_ID}"
umask 077
jq -n \
--arg repository_id "$REPOSITORY_ID" \
--arg repository "$REPOSITORY" \
--arg workflow_id "$WORKFLOW_ID" \
--arg workflow_path "$WORKFLOW_PATH" \
--arg run_id "$RUN_ID" \
--arg run_attempt "$RUN_ATTEMPT" \
--arg source_comment_id "$SOURCE_COMMENT_ID" \
--arg dispatch_comment_id "$DISPATCH_COMMENT_ID" \
--arg actor_id "$ACTOR_ID" \
--arg actor_login "$ACTOR_LOGIN" \
--arg pr_number "$PR_NUMBER" \
--arg pr_head_sha "$PR_HEAD_SHA" \
--arg products "$PRODUCTS" \
--arg cases_ref "$CASES_REF" \
--arg source_body_sha256 "$source_body_sha256" \
--arg idempotency_key "$idempotency_key" \
'{
schema_version: 1,
repository_id: $repository_id,
repository: $repository,
workflow_id: $workflow_id,
workflow_path: $workflow_path,
run_id: $run_id,
run_attempt: $run_attempt,
source_comment_id: $source_comment_id,
dispatch_comment_id: $dispatch_comment_id,
actor_id: $actor_id,
actor_login: $actor_login,
pr_number: $pr_number,
pr_head_sha: $pr_head_sha,
products: $products,
cases_ref: $cases_ref,
source_body_sha256: $source_body_sha256,
idempotency_key: $idempotency_key
}' > "$MANIFEST_PATH"
- name: Upload dispatch request manifest
id: artifact
uses: actions/upload-artifact@v4
with:
name: eval-dispatch-request-${{ github.run_id }}-${{ github.run_attempt }}-${{ steps.placeholder.outputs.comment_id }}
path: ${{ runner.temp }}/eval-dispatch-request.json
if-no-files-found: error
retention-days: 1
overwrite: false
- name: Finalize dispatch marker
env:
GH_TOKEN: ${{ github.token }}
DISPATCH_COMMENT_ID: ${{ steps.placeholder.outputs.comment_id }}
REPOSITORY_ID: '1187709537'
WORKFLOW_ID: '331725458'
WORKFLOW_PATH: .github/workflows/eval-dispatch.yml
RUN_ID: ${{ github.run_id }}
RUN_ATTEMPT: ${{ github.run_attempt }}
ARTIFACT_ID: ${{ steps.artifact.outputs.artifact-id }}
ARTIFACT_DIGEST: ${{ steps.artifact.outputs.artifact-digest }}
PR_HEAD_SHA: ${{ steps.pr.outputs.head_sha }}
PRODUCTS: ${{ steps.parse.outputs.products }}
CASES_REF: ${{ steps.parse.outputs.cases_ref }}
run: |
set -euo pipefail
if [[ ! "$DISPATCH_COMMENT_ID" =~ ^[1-9][0-9]*$ ]] || \
[[ ! "$ARTIFACT_ID" =~ ^[1-9][0-9]*$ ]]; then
echo "artifact marker contains a non-canonical identifier" >&2
exit 1
fi
artifact_digest="${ARTIFACT_DIGEST,,}"
if [[ "$artifact_digest" != sha256:* ]]; then
artifact_digest="sha256:${artifact_digest}"
fi
if [[ ! "$artifact_digest" =~ ^sha256:[0-9a-f]{64}$ ]]; then
echo "artifact marker contains an invalid digest" >&2
exit 1
fi
marker_json="$(
jq -nc \
--arg repository_id "$REPOSITORY_ID" \
--arg workflow_id "$WORKFLOW_ID" \
--arg workflow_path "$WORKFLOW_PATH" \
--arg run_id "$RUN_ID" \
--arg run_attempt "$RUN_ATTEMPT" \
--arg dispatch_comment_id "$DISPATCH_COMMENT_ID" \
--arg artifact_id "$ARTIFACT_ID" \
--arg artifact_digest "$artifact_digest" \
'{
schema_version: 1,
repository_id: $repository_id,
workflow_id: $workflow_id,
workflow_path: $workflow_path,
run_id: $run_id,
run_attempt: $run_attempt,
dispatch_comment_id: $dispatch_comment_id,
artifact_id: $artifact_id,
artifact_digest: $artifact_digest
}'
)"
cases_note=""
if [ -n "$CASES_REF" ]; then
cases_note="，用例版本 \`${CASES_REF}\`"
fi
# 安全模型：结构化标记供内网轮询服务提取并触发 Aone CI。
# 防伪造保证由消费端三重校验（见 scripts/ci/eval_poll_validate.py）：
# 1. comment.user.login == 'github-actions[bot]'（平台级身份，不可冒充）
# 2. comment.performed_via_github_app.slug == 'github-actions'（App 签名）
# 3. payload.run_id 对应真实成功的 workflow run（GET /actions/runs/{id}）
# 任意一层不通过即丢弃，普通用户手写评论无法通过上述校验。
dispatch_payload=$(jq -nc \
--arg pr "$PR_NUMBER" \
--arg sha "$PR_HEAD_SHA" \
--arg products "$PRODUCTS" \
--arg cases "$CASES_REF" \
--arg run_id "$RUN_ID" \
'{pr_number: $pr, pr_head_sha: $sha, products: $products, cases_ref: $cases, run_id: $run_id}')
body="<!-- eval-dispatch: ${dispatch_payload} -->"$'\n'"🛰️ /eval 已受理：产品集 \`${PRODUCTS}\`${cases_note}，评测对象 \`${PR_HEAD_SHA}\`。"$'\n'"内网评测流水线将在数分钟内由轮询服务触发，运行结束后由 bot 回贴报告（首行为基线对比头条）。"
jq -n --arg body "$body" '{body: $body}' | curl --fail --silent --show-error \
-X POST \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
--data @- \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/comments" > /dev/null
body="<!-- eval-dispatch: ${marker_json} -->"$'\n'"🛰️ /eval 已受理：产品集 \`${PRODUCTS}\`${cases_note}，评测对象 \`${PR_HEAD_SHA}\`。"$'\n'"受控评测服务将在数分钟内处理，完成后由 bot 回贴报告。"
response="$(
jq -n --arg body "$body" '{body: $body}' \
| curl --fail --silent --show-error \
-X PATCH \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
--data @- \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/issues/comments/${DISPATCH_COMMENT_ID}"
)"
printf '%s' "$response" \
| jq -e \
--arg comment_id "$DISPATCH_COMMENT_ID" \
--arg body "$body" \
'((.id | tostring) == $comment_id) and (.body == $body)' \
> /dev/null
- name: Mark dispatch preparation failure
if: ${{ failure() && steps.placeholder.outputs.comment_id != '' }}
env:
GH_TOKEN: ${{ github.token }}
DISPATCH_COMMENT_ID: ${{ steps.placeholder.outputs.comment_id }}
run: |
failure_body="❌ /eval 请求准备失败，未生成可消费的评测请求。请稍后重试。"
jq -n --arg body "$failure_body" '{body: $body}' \
| curl --fail --silent --show-error \
-X PATCH \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
--data @- \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/issues/comments/${DISPATCH_COMMENT_ID}" \
> /dev/null \
|| true
scripts/ci/eval_poll_validate.py
+478 -109
View File
@@ -1,144 +1,513 @@
"""eval_poll_validate.py — 消费端校验 eval-dispatch 结构化评论的合法性。
#!/usr/bin/env python3
"""验证 GitHub Actions 产出的 eval-dispatch artifact 请求。
三重防伪造校验：
1. comment.user.login == 'github-actions[bot]' — 平台级身份，普通用户不可冒充
2. comment.performed_via_github_app.slug == 'github-actions' — GitHub App 签名
3. payload.run_id 对应本仓库真实成功完成的 workflow run
任何一层不通过即返回 False，拒绝触发内网流水线。
评论只是索引。真正的授权记录是绑定到精确 workflow run、run attempt 和
dispatch comment ID 的不可变 artifact。任何缺失、冲突或异常都必须拒绝，
不能让单条不可信评论终止轮询进程。校验成功时返回稳定的
idempotency_key；调用方必须在触发评测前持久化、原子占用该 key，重复
占用只能 no-op。本脚本无状态，不以 reaction 代替持久去重。
"""
from __future__ import annotations
import hashlib
import io
import json
import re
import stat
import subprocess
import sys
from typing import Optional
import zipfile
from typing import Any, Optional, Protocol
from eval_comment_parse import parse as parse_eval_comment
REPO = "DingTalk-Real-AI/dingtalk-workspace-cli"
REPOSITORY = "DingTalk-Real-AI/dingtalk-workspace-cli"
REPOSITORY_ID = "1187709537"
WORKFLOW_ID = "331725458"
WORKFLOW_PATH = ".github/workflows/eval-dispatch.yml"
DEFAULT_BRANCH = "main"
TRUSTED_BOT_LOGIN = "github-actions[bot]"
TRUSTED_APP_SLUG = "github-actions"
ARTIFACT_FILENAME = "eval-dispatch-request.json"
SCHEMA_VERSION = 1
def validate_comment_author(comment: dict) -> bool:
"""校验评论由 github-actions[bot] 通过 GitHub Actions App 发出。"""
user = comment.get("user", {})
if user.get("login") != TRUSTED_BOT_LOGIN:
return False
if user.get("type") != "Bot":
return False
app = comment.get("performed_via_github_app") or {}
if app.get("slug") != TRUSTED_APP_SLUG:
return False
return True
def validate_run_id(run_id, repo: str = REPO) -> bool:
"""校验 run_id 对应真实存在且成功完成的 workflow run。"""
if not isinstance(run_id, str) or not run_id.isdigit():
return False
result = subprocess.run(
["gh", "api", f"repos/{repo}/actions/runs/{run_id}",
"--jq", ".status + \"|\" + .conclusion + \"|\" + .name"],
capture_output=True, text=True
)
if result.returncode != 0:
return False
parts = result.stdout.strip().split("|")
if len(parts) < 3:
return False
status, conclusion, name = parts[0], parts[1], parts[2]
if status != "completed" or conclusion != "success":
return False
if "Eval Dispatch" not in name:
return False
return True
def validate_pr_head(pr_number: str, expected_sha: str, repo: str = REPO) -> bool:
"""二次校验 PR 仍然 open 且 head SHA 一致（防 TOCTOU）。"""
result = subprocess.run(
["gh", "api", f"repos/{repo}/pulls/{pr_number}",
"--jq", ".state + \"|\" + .head.sha"],
capture_output=True, text=True
)
if result.returncode != 0:
return False
parts = result.stdout.strip().split("|")
if len(parts) < 2:
return False
state, current_sha = parts[0], parts[1]
if state != "open":
return False
if current_sha != expected_sha:
return False
return True
MAX_COMMENT_BYTES = 64 * 1024
MAX_API_JSON_BYTES = 1024 * 1024
MAX_ARCHIVE_BYTES = 64 * 1024
MAX_MANIFEST_BYTES = 64 * 1024
MAX_PRODUCTS_BYTES = 512
MAX_CASES_REF_BYTES = 1024
_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
_PRODUCTS_RE = re.compile(r"^[a-zA-Z0-9_,.-]+$")
_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
_DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
_PRODUCTS_RE = re.compile(r"^[a-z0-9][a-z0-9-]*(?:,[a-z0-9][a-z0-9-]*)*$")
_REF_CHARSET_RE = re.compile(r"^[A-Za-z0-9._/-]+$")
_LOGIN_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?$")
_MARKER_RE = re.compile(r"<!-- eval-dispatch: ([^\r\n]*?) -->")
_MARKER_FIELDS = {
"schema_version",
"repository_id",
"workflow_id",
"workflow_path",
"run_id",
"run_attempt",
"dispatch_comment_id",
"artifact_id",
"artifact_digest",
}
_MANIFEST_FIELDS = {
"schema_version",
"repository_id",
"repository",
"workflow_id",
"workflow_path",
"run_id",
"run_attempt",
"source_comment_id",
"dispatch_comment_id",
"actor_id",
"actor_login",
"pr_number",
"pr_head_sha",
"products",
"cases_ref",
"source_body_sha256",
"idempotency_key",
}
def extract_payload(body: str) -> Optional[dict]:
"""从评论 body 提取 eval-dispatch JSON payload，严格要求为合法 dict 且字段类型正确。"""
match = re.search(r"<!-- eval-dispatch: ({.*?}) -->", body)
if not match:
class GitHubClient(Protocol):
"""consumer 所需的最小 GitHub 只读边界。"""
def get_run_attempt(self, run_id: str, run_attempt: str) -> Any: ...
def get_artifact(self, artifact_id: str) -> Any: ...
def download_artifact(self, artifact_id: str) -> bytes: ...
def get_comment(self, comment_id: str) -> Any: ...
def get_pull_request(self, pr_number: str) -> Any: ...
class GitHubCLIClient:
"""通过已认证的 gh CLI 读取 GitHub REST API。"""
def _json(self, endpoint: str) -> Any:
result = subprocess.run(
["gh", "api", endpoint],
capture_output=True,
text=True,
timeout=20,
)
if result.returncode != 0:
raise RuntimeError("GitHub API 请求失败")
if len(result.stdout.encode("utf-8")) > MAX_API_JSON_BYTES:
raise ValueError("GitHub API JSON 响应过大")
return _loads_no_duplicates(result.stdout)
def get_run_attempt(self, run_id: str, run_attempt: str) -> Any:
return self._json(
f"repos/{REPOSITORY}/actions/runs/{run_id}/attempts/{run_attempt}"
)
def get_artifact(self, artifact_id: str) -> Any:
return self._json(f"repos/{REPOSITORY}/actions/artifacts/{artifact_id}")
def download_artifact(self, artifact_id: str) -> bytes:
result = subprocess.run(
["gh", "api", f"repos/{REPOSITORY}/actions/artifacts/{artifact_id}/zip"],
capture_output=True,
timeout=30,
)
if result.returncode != 0:
raise RuntimeError("下载 GitHub artifact 失败")
if len(result.stdout) > MAX_ARCHIVE_BYTES:
raise ValueError("artifact 压缩包过大")
return result.stdout
def get_comment(self, comment_id: str) -> Any:
return self._json(f"repos/{REPOSITORY}/issues/comments/{comment_id}")
def get_pull_request(self, pr_number: str) -> Any:
return self._json(f"repos/{REPOSITORY}/pulls/{pr_number}")
def _loads_no_duplicates(raw: str) -> Any:
def reject_duplicates(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"JSON 字段重复: {key}")
result[key] = value
return result
return json.loads(raw, object_pairs_hook=reject_duplicates)
def _is_decimal_string(value: Any, *, max_length: int = 20) -> bool:
return (
type(value) is str
and 1 <= len(value) <= max_length
and value.isascii()
and value.isdigit()
and value[0] != "0"
)
def _as_decimal_string(value: Any) -> Optional[str]:
if type(value) is int and value > 0:
return str(value)
if _is_decimal_string(value):
return value
return None
def _is_valid_cases_ref(value: Any) -> bool:
if type(value) is not str or len(value.encode("utf-8")) > MAX_CASES_REF_BYTES:
return False
if value == "":
return True
if not _REF_CHARSET_RE.fullmatch(value):
return False
if value.startswith(("-", "/")) or value.endswith(("/", ".")):
return False
if ".." in value or "//" in value:
return False
return all(
component
and not component.startswith(".")
and not component.endswith(".lock")
for component in value.split("/")
)
def _is_string(value: Any, *, max_bytes: int, allow_empty: bool = False) -> bool:
return (
type(value) is str
and (allow_empty or bool(value))
and len(value.encode("utf-8")) <= max_bytes
)
def validate_comment_author(comment: Any) -> bool:
"""校验 dispatch 评论由 GitHub Actions App 发出。"""
if type(comment) is not dict:
return False
user = comment.get("user")
app = comment.get("performed_via_github_app")
return (
type(user) is dict
and user.get("login") == TRUSTED_BOT_LOGIN
and user.get("type") == "Bot"
and type(app) is dict
and app.get("slug") == TRUSTED_APP_SLUG
)
def _extract_marker(body: Any) -> Optional[dict[str, Any]]:
if not _is_string(body, max_bytes=MAX_COMMENT_BYTES):
return None
if body.count("<!-- eval-dispatch:") != 1:
return None
matches = _MARKER_RE.findall(body)
if len(matches) != 1 or len(matches[0].encode("utf-8")) > 8 * 1024:
return None
try:
data = json.loads(match.group(1))
except (json.JSONDecodeError, ValueError):
marker = _loads_no_duplicates(matches[0])
except (json.JSONDecodeError, TypeError, ValueError):
return None
if not isinstance(data, dict):
if type(marker) is not dict or set(marker) != _MARKER_FIELDS:
return None
pr_number = data.get("pr_number")
if not isinstance(pr_number, str) or not pr_number.isdigit():
if type(marker.get("schema_version")) is not int or marker["schema_version"] != SCHEMA_VERSION:
return None
pr_head_sha = data.get("pr_head_sha")
if not isinstance(pr_head_sha, str) or not _SHA_RE.match(pr_head_sha):
for field, expected in (
("repository_id", REPOSITORY_ID),
("workflow_id", WORKFLOW_ID),
("workflow_path", WORKFLOW_PATH),
):
if marker.get(field) != expected:
return None
for field in ("run_id", "run_attempt", "dispatch_comment_id", "artifact_id"):
if not _is_decimal_string(marker.get(field)):
return None
if type(marker.get("artifact_digest")) is not str or not _DIGEST_RE.fullmatch(
marker["artifact_digest"]
):
return None
products = data.get("products")
if not isinstance(products, str) or not _PRODUCTS_RE.match(products):
return None
run_id = data.get("run_id")
if not isinstance(run_id, str) or not run_id.isdigit():
return None
cases_ref = data.get("cases_ref")
if not isinstance(cases_ref, str):
return None
return data
return marker
def validate_comment(comment: dict, verify_run: bool = True, verify_pr: bool = True) -> Optional[dict]:
"""完整校验流程，通过返回 payload dict，不通过返回 None。"""
def _validate_run(run: Any, marker: dict[str, Any]) -> bool:
if type(run) is not dict:
return False
repository = run.get("repository")
head_repository = run.get("head_repository")
return (
_as_decimal_string(run.get("id")) == marker["run_id"]
and _as_decimal_string(run.get("run_attempt")) == marker["run_attempt"]
and _as_decimal_string(run.get("workflow_id")) == WORKFLOW_ID
and run.get("path") == WORKFLOW_PATH
and run.get("event") == "issue_comment"
and run.get("head_branch") == DEFAULT_BRANCH
and run.get("status") == "completed"
and run.get("conclusion") == "success"
and type(repository) is dict
and _as_decimal_string(repository.get("id")) == REPOSITORY_ID
and repository.get("full_name") == REPOSITORY
and type(head_repository) is dict
and _as_decimal_string(head_repository.get("id")) == REPOSITORY_ID
and head_repository.get("full_name") == REPOSITORY
and type(run.get("head_sha")) is str
and bool(_SHA_RE.fullmatch(run["head_sha"]))
)
def _validate_artifact_metadata(
artifact: Any, marker: dict[str, Any], run: dict[str, Any]
) -> bool:
if type(artifact) is not dict:
return False
workflow_run = artifact.get("workflow_run")
expected_name = (
f"eval-dispatch-request-{marker['run_id']}-{marker['run_attempt']}-"
f"{marker['dispatch_comment_id']}"
)
size = artifact.get("size_in_bytes")
return (
_as_decimal_string(artifact.get("id")) == marker["artifact_id"]
and artifact.get("name") == expected_name
and artifact.get("expired") is False
and artifact.get("digest") == marker["artifact_digest"]
and type(size) is int
and 0 < size <= MAX_ARCHIVE_BYTES
and type(workflow_run) is dict
and _as_decimal_string(workflow_run.get("id")) == marker["run_id"]
and _as_decimal_string(workflow_run.get("repository_id")) == REPOSITORY_ID
and _as_decimal_string(workflow_run.get("head_repository_id")) == REPOSITORY_ID
and workflow_run.get("head_sha") == run.get("head_sha")
)
def _read_manifest(archive_bytes: Any, expected_digest: str) -> Optional[dict[str, Any]]:
if type(archive_bytes) is not bytes or not 0 < len(archive_bytes) <= MAX_ARCHIVE_BYTES:
return None
actual_digest = "sha256:" + hashlib.sha256(archive_bytes).hexdigest()
if actual_digest != expected_digest:
return None
try:
with zipfile.ZipFile(io.BytesIO(archive_bytes), "r") as archive:
infos = archive.infolist()
if len(infos) != 1:
return None
info = infos[0]
if info.filename != ARTIFACT_FILENAME or info.is_dir():
return None
if info.flag_bits & 0x1:
return None
if info.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED):
return None
mode = (info.external_attr >> 16) & 0o177777
file_type = stat.S_IFMT(mode)
if file_type not in (0, stat.S_IFREG):
return None
if not 0 < info.file_size <= MAX_MANIFEST_BYTES:
return None
raw_manifest = archive.read(info)
except (EOFError, OSError, RuntimeError, ValueError, zipfile.BadZipFile):
return None
if not 0 < len(raw_manifest) <= MAX_MANIFEST_BYTES:
return None
try:
manifest = _loads_no_duplicates(raw_manifest.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError):
return None
if type(manifest) is not dict or set(manifest) != _MANIFEST_FIELDS:
return None
return manifest
def _validate_manifest_shape(manifest: dict[str, Any]) -> bool:
if (
type(manifest.get("schema_version")) is not int
or manifest["schema_version"] != SCHEMA_VERSION
):
return False
for field, expected in (
("repository_id", REPOSITORY_ID),
("repository", REPOSITORY),
("workflow_id", WORKFLOW_ID),
("workflow_path", WORKFLOW_PATH),
):
if manifest.get(field) != expected:
return False
for field in (
"run_id",
"run_attempt",
"source_comment_id",
"dispatch_comment_id",
"actor_id",
"pr_number",
):
if not _is_decimal_string(manifest.get(field)):
return False
if type(manifest.get("actor_login")) is not str or not _LOGIN_RE.fullmatch(
manifest["actor_login"]
):
return False
if type(manifest.get("pr_head_sha")) is not str or not _SHA_RE.fullmatch(
manifest["pr_head_sha"]
):
return False
products = manifest.get("products")
if (
type(products) is not str
or len(products.encode("utf-8")) > MAX_PRODUCTS_BYTES
or not _PRODUCTS_RE.fullmatch(products)
):
return False
if not _is_valid_cases_ref(manifest.get("cases_ref")):
return False
if type(manifest.get("source_body_sha256")) is not str or not _SHA256_RE.fullmatch(
manifest["source_body_sha256"]
):
return False
expected_key = f"{REPOSITORY_ID}:{manifest['source_comment_id']}"
return manifest.get("idempotency_key") == expected_key
def _expected_issue_url(pr_number: str) -> str:
return f"https://api.github.com/repos/{REPOSITORY}/issues/{pr_number}"
def _validate_source_comment(source: Any, manifest: dict[str, Any]) -> bool:
if type(source) is not dict:
return False
user = source.get("user")
body = source.get("body")
if (
_as_decimal_string(source.get("id")) != manifest["source_comment_id"]
or source.get("issue_url") != _expected_issue_url(manifest["pr_number"])
or type(user) is not dict
or _as_decimal_string(user.get("id")) != manifest["actor_id"]
or user.get("login") != manifest["actor_login"]
or not _is_string(body, max_bytes=MAX_COMMENT_BYTES)
or hashlib.sha256(body.encode("utf-8")).hexdigest()
!= manifest["source_body_sha256"]
):
return False
try:
products, cases_ref, reviewed_sha = parse_eval_comment(body)
except (TypeError, ValueError):
return False
return (
products == manifest["products"]
and cases_ref == manifest["cases_ref"]
and (not reviewed_sha or reviewed_sha == manifest["pr_head_sha"])
)
def _validate_run_actor(run: dict[str, Any], manifest: dict[str, Any]) -> bool:
actor = run.get("actor")
return (
type(actor) is dict
and _as_decimal_string(actor.get("id")) == manifest["actor_id"]
and actor.get("login") == manifest["actor_login"]
)
def _validate_pull_request(pr: Any, manifest: dict[str, Any]) -> bool:
if type(pr) is not dict:
return False
head = pr.get("head")
return (
_as_decimal_string(pr.get("number")) == manifest["pr_number"]
and pr.get("state") == "open"
and type(head) is dict
and head.get("sha") == manifest["pr_head_sha"]
)
def _validate_comment(comment: Any, client: GitHubClient) -> Optional[dict[str, str]]:
if not validate_comment_author(comment):
print(f" REJECT: 评论作者不是可信 bot (id={comment.get('id')})")
return None
dispatch_comment_id = _as_decimal_string(comment.get("id"))
marker = _extract_marker(comment.get("body"))
if dispatch_comment_id is None or marker is None:
return None
if marker["dispatch_comment_id"] != dispatch_comment_id:
return None
payload = extract_payload(comment.get("body", ""))
if not payload:
run = client.get_run_attempt(marker["run_id"], marker["run_attempt"])
if not _validate_run(run, marker):
return None
if verify_run:
run_id = payload.get("run_id", "")
if not validate_run_id(run_id):
print(f" REJECT: run_id={run_id} 不是有效的成功 workflow run")
return None
artifact = client.get_artifact(marker["artifact_id"])
if not _validate_artifact_metadata(artifact, marker, run):
return None
archive = client.download_artifact(marker["artifact_id"])
manifest = _read_manifest(archive, marker["artifact_digest"])
if manifest is None or not _validate_manifest_shape(manifest):
return None
if verify_pr:
pr_number = payload.get("pr_number", "")
pr_head_sha = payload.get("pr_head_sha", "")
if not validate_pr_head(pr_number, pr_head_sha):
print(f" REJECT: PR #{pr_number} 非 open 或 head 已变更")
for field in (
"schema_version",
"repository_id",
"workflow_id",
"workflow_path",
"run_id",
"run_attempt",
"dispatch_comment_id",
):
if manifest.get(field) != marker.get(field):
return None
if comment.get("issue_url") != _expected_issue_url(manifest["pr_number"]):
return None
if not _validate_run_actor(run, manifest):
return None
return payload
source_comment = client.get_comment(manifest["source_comment_id"])
if not _validate_source_comment(source_comment, manifest):
return None
pr = client.get_pull_request(manifest["pr_number"])
if not _validate_pull_request(pr, manifest):
return None
return {
"pr_number": manifest["pr_number"],
"pr_head_sha": manifest["pr_head_sha"],
"products": manifest["products"],
"cases_ref": manifest["cases_ref"],
"idempotency_key": manifest["idempotency_key"],
}
def validate_comment(
comment: Any, client: Optional[GitHubClient] = None
) -> Optional[dict[str, str]]:
"""完整校验一条 dispatch 评论；调用方须原子占用返回的幂等键。"""
try:
return _validate_comment(comment, client or GitHubCLIClient())
except Exception:
return None
def main() -> int:
try:
comment = _loads_no_duplicates(sys.stdin.read())
except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError):
return 1
result = validate_comment(comment)
if result is None:
return 1
print(json.dumps(result, ensure_ascii=False, separators=(",", ":"), sort_keys=True))
return 0
if __name__ == "__main__":
comment_json = json.load(sys.stdin)
result = validate_comment(comment_json)
if result:
print(json.dumps(result))
sys.exit(0)
else:
sys.exit(1)
sys.exit(main())
scripts/ci/test_eval_poll_validate.py
+720 -131
View File
@@ -1,173 +1,762 @@
"""test_eval_poll_validate.py — 验证消费端拒绝伪造的 eval-dispatch 评论。"""
"""eval-dispatch 消费端安全协议的无网络行为测试。"""
from __future__ import annotations
import copy
import hashlib
import io
import json
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "scripts", "ci"))
from eval_poll_validate import validate_comment_author, extract_payload, validate_run_id
import stat
import sys
import unittest
import warnings
import zipfile
from unittest import mock
def test_rejects_regular_user_comment():
"""普通用户手写 eval-dispatch 标记的评论必须被拒绝。"""
comment = {
"id": 1234,
"user": {"login": "malicious-user", "type": "User"},
"body": '<!-- eval-dispatch: {"pr_number":"899","pr_head_sha":"a"*40,"products":"drive","cases_ref":"","run_id":"99999"} -->',
"performed_via_github_app": None,
sys.path.insert(0, os.path.dirname(__file__))
from eval_poll_validate import GitHubCLIClient, main as validate_main, validate_comment
REPOSITORY = "DingTalk-Real-AI/dingtalk-workspace-cli"
REPOSITORY_ID = "1187709537"
WORKFLOW_ID = "331725458"
WORKFLOW_PATH = ".github/workflows/eval-dispatch.yml"
RUN_ID = "31490000000"
RUN_ATTEMPT = "1"
SOURCE_COMMENT_ID = "5250000001"
DISPATCH_COMMENT_ID = "5250000002"
ARTIFACT_ID = "9100000001"
ACTOR_ID = "30925823"
ACTOR_LOGIN = "trusted-reviewer"
PR_NUMBER = "952"
PR_HEAD_SHA = "a" * 40
DEFAULT_BRANCH_SHA = "b" * 40
PRODUCTS = "drive,doc"
CASES_REF = "fixtures/v1"
SOURCE_BODY = f"/eval {PRODUCTS} sha={PR_HEAD_SHA} cases={CASES_REF}"
ARTIFACT_FILENAME = "eval-dispatch-request.json"
def _compact_json(value: dict) -> str:
return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
def _zip_single_file(name: str, content: bytes) -> bytes:
output = io.BytesIO()
with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive:
archive.writestr(name, content)
return output.getvalue()
def _zip_files(
entries: tuple[tuple[str, bytes], ...], *, compression=zipfile.ZIP_DEFLATED
) -> bytes:
output = io.BytesIO()
with warnings.catch_warnings():
warnings.simplefilter("ignore", UserWarning)
with zipfile.ZipFile(output, "w", compression=compression) as archive:
for name, content in entries:
archive.writestr(name, content)
return output.getvalue()
def _zip_symlink(name: str, target: str) -> bytes:
output = io.BytesIO()
with zipfile.ZipFile(output, "w") as archive:
info = zipfile.ZipInfo(name)
info.create_system = 3
info.external_attr = (stat.S_IFLNK | 0o777) << 16
archive.writestr(info, target.encode("utf-8"))
return output.getvalue()
def _set_marker(comment: dict, marker: dict) -> None:
comment["body"] = f"<!-- eval-dispatch: {_compact_json(marker)} -->\n已受理。"
def _install_archive(comment: dict, client, marker: dict, archive: bytes) -> None:
digest = "sha256:" + hashlib.sha256(archive).hexdigest()
client.archive = archive
client.artifact["size_in_bytes"] = len(archive)
client.artifact["digest"] = digest
marker["artifact_digest"] = digest
_set_marker(comment, marker)
def _install_manifest(
comment: dict, client, manifest: dict, marker: dict
) -> None:
archive = _zip_single_file(
ARTIFACT_FILENAME,
(_compact_json(manifest) + "\n").encode("utf-8"),
)
_install_archive(comment, client, marker, archive)
def _set_nested(value: dict, path: tuple[str, ...], replacement) -> None:
current = value
for component in path[:-1]:
current = current[component]
current[path[-1]] = replacement
class FakeGitHubClient:
"""只模拟 consumer 的 GitHub 外部边界，不模拟内部校验步骤。"""
def __init__(self, *, run, artifact, archive, source_comment, pr):
self.run = run
self.artifact = artifact
self.archive = archive
self.source_comment = source_comment
self.pr = pr
self.calls = []
def get_run_attempt(self, run_id: str, run_attempt: str):
self.calls.append(("get_run_attempt", run_id, run_attempt))
return copy.deepcopy(self.run)
def get_artifact(self, artifact_id: str):
self.calls.append(("get_artifact", artifact_id))
return copy.deepcopy(self.artifact)
def download_artifact(self, artifact_id: str):
self.calls.append(("download_artifact", artifact_id))
return self.archive
def get_comment(self, comment_id: str):
self.calls.append(("get_comment", comment_id))
return copy.deepcopy(self.source_comment)
def get_pull_request(self, pr_number: str):
self.calls.append(("get_pull_request", pr_number))
return copy.deepcopy(self.pr)
def valid_fixture():
manifest = {
"schema_version": 1,
"repository_id": REPOSITORY_ID,
"repository": REPOSITORY,
"workflow_id": WORKFLOW_ID,
"workflow_path": WORKFLOW_PATH,
"run_id": RUN_ID,
"run_attempt": RUN_ATTEMPT,
"source_comment_id": SOURCE_COMMENT_ID,
"dispatch_comment_id": DISPATCH_COMMENT_ID,
"actor_id": ACTOR_ID,
"actor_login": ACTOR_LOGIN,
"pr_number": PR_NUMBER,
"pr_head_sha": PR_HEAD_SHA,
"products": PRODUCTS,
"cases_ref": CASES_REF,
"source_body_sha256": hashlib.sha256(SOURCE_BODY.encode("utf-8")).hexdigest(),
"idempotency_key": f"{REPOSITORY_ID}:{SOURCE_COMMENT_ID}",
}
assert validate_comment_author(comment) is False
def test_rejects_wrong_bot_login():
"""非 github-actions[bot] 的 Bot 账号必须被拒绝。"""
comment = {
"id": 1235,
"user": {"login": "some-other-bot[bot]", "type": "Bot"},
"body": '<!-- eval-dispatch: {"pr_number":"899","pr_head_sha":"a"*40,"products":"drive","cases_ref":"","run_id":"99999"} -->',
"performed_via_github_app": {"slug": "some-other-app"},
archive = _zip_single_file(
ARTIFACT_FILENAME,
(_compact_json(manifest) + "\n").encode("utf-8"),
)
artifact_digest = "sha256:" + hashlib.sha256(archive).hexdigest()
marker = {
"schema_version": 1,
"repository_id": REPOSITORY_ID,
"workflow_id": WORKFLOW_ID,
"workflow_path": WORKFLOW_PATH,
"run_id": RUN_ID,
"run_attempt": RUN_ATTEMPT,
"dispatch_comment_id": DISPATCH_COMMENT_ID,
"artifact_id": ARTIFACT_ID,
"artifact_digest": artifact_digest,
}
assert validate_comment_author(comment) is False
def test_rejects_bot_without_app_signature():
"""github-actions[bot] 但缺少 performed_via_github_app 签名。"""
issue_url = f"https://api.github.com/repos/{REPOSITORY}/issues/{PR_NUMBER}"
comment = {
"id": 1236,
"id": int(DISPATCH_COMMENT_ID),
"issue_url": issue_url,
"body": f"<!-- eval-dispatch: {_compact_json(marker)} -->\n已受理。",
"user": {"login": "github-actions[bot]", "type": "Bot"},
"body": '<!-- eval-dispatch: {"pr_number":"899"} -->',
"performed_via_github_app": None,
}
assert validate_comment_author(comment) is False
def test_rejects_bot_with_wrong_app_slug():
"""github-actions[bot] 但 App slug 不匹配。"""
comment = {
"id": 1237,
"user": {"login": "github-actions[bot]", "type": "Bot"},
"body": '<!-- eval-dispatch: {"pr_number":"899"} -->',
"performed_via_github_app": {"slug": "dependabot"},
}
assert validate_comment_author(comment) is False
def test_accepts_legitimate_github_actions_comment():
"""正确的 github-actions[bot] + github-actions App 签名通过身份校验。"""
comment = {
"id": 1238,
"user": {"login": "github-actions[bot]", "type": "Bot"},
"body": '<!-- eval-dispatch: {"pr_number":"899","pr_head_sha":"abcd1234","products":"drive","cases_ref":"","run_id":"12345"} -->',
"performed_via_github_app": {"slug": "github-actions"},
}
assert validate_comment_author(comment) is True
run = {
"id": int(RUN_ID),
"run_attempt": int(RUN_ATTEMPT),
"workflow_id": int(WORKFLOW_ID),
"path": WORKFLOW_PATH,
"event": "issue_comment",
"head_branch": "main",
"head_sha": DEFAULT_BRANCH_SHA,
"status": "completed",
"conclusion": "success",
"repository": {"id": int(REPOSITORY_ID), "full_name": REPOSITORY},
"head_repository": {"id": int(REPOSITORY_ID), "full_name": REPOSITORY},
"actor": {"id": int(ACTOR_ID), "login": ACTOR_LOGIN},
}
artifact = {
"id": int(ARTIFACT_ID),
"name": f"eval-dispatch-request-{RUN_ID}-{RUN_ATTEMPT}-{DISPATCH_COMMENT_ID}",
"size_in_bytes": len(archive),
"expired": False,
"digest": artifact_digest,
"workflow_run": {
"id": int(RUN_ID),
"repository_id": int(REPOSITORY_ID),
"head_repository_id": int(REPOSITORY_ID),
"head_sha": DEFAULT_BRANCH_SHA,
},
}
source_comment = {
"id": int(SOURCE_COMMENT_ID),
"issue_url": issue_url,
"body": SOURCE_BODY,
"user": {"id": int(ACTOR_ID), "login": ACTOR_LOGIN},
}
pr = {"number": int(PR_NUMBER), "state": "open", "head": {"sha": PR_HEAD_SHA}}
client = FakeGitHubClient(
run=run,
artifact=artifact,
archive=archive,
source_comment=source_comment,
pr=pr,
)
return comment, client, manifest, marker
def test_extract_payload_valid():
body = '<!-- eval-dispatch: {"pr_number":"899","pr_head_sha":"a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2","products":"drive","cases_ref":"","run_id":"123"} -->\nsome text'
payload = extract_payload(body)
assert payload is not None
assert payload["pr_number"] == "899"
assert payload["run_id"] == "123"
assert payload["pr_head_sha"] == "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2"
class EvalPollValidateTests(unittest.TestCase):
def assert_rejected(self, comment, client):
self.assertIsNone(validate_comment(comment, client=client))
def test_accepts_fully_bound_dispatch(self):
comment, client, _, _ = valid_fixture()
def test_extract_payload_no_marker():
body = "just a regular comment"
assert extract_payload(body) is None
self.assertEqual(
validate_comment(comment, client=client),
{
"pr_number": PR_NUMBER,
"pr_head_sha": PR_HEAD_SHA,
"products": PRODUCTS,
"cases_ref": CASES_REF,
"idempotency_key": f"{REPOSITORY_ID}:{SOURCE_COMMENT_ID}",
},
)
self.assertEqual(
client.calls,
[
("get_run_attempt", RUN_ID, RUN_ATTEMPT),
("get_artifact", ARTIFACT_ID),
("download_artifact", ARTIFACT_ID),
("get_comment", SOURCE_COMMENT_ID),
("get_pull_request", PR_NUMBER),
],
)
def test_rejects_untrusted_or_malformed_dispatch_comment(self):
mutations = (
("非字典", lambda comment: None),
("缺少用户", lambda comment: comment.pop("user")),
("伪造机器人", lambda comment: comment["user"].update(login="attacker")),
("用户类型错误", lambda comment: comment["user"].update(type="User")),
("缺少 GitHub App", lambda comment: comment.pop("performed_via_github_app")),
(
"GitHub App 错误",
lambda comment: comment["performed_via_github_app"].update(
slug="untrusted-app"
),
),
("body 非字符串", lambda comment: comment.update(body=None)),
("comment id 非十进制", lambda comment: comment.update(id="01")),
(
"重复 marker",
lambda comment: comment.update(body=comment["body"] + "\n" + comment["body"]),
),
)
for label, mutate in mutations:
with self.subTest(label=label):
comment, client, _, _ = valid_fixture()
replacement = mutate(comment)
if label == "非字典":
comment = replacement
self.assert_rejected(comment, client)
def test_extract_payload_malformed_json():
body = "<!-- eval-dispatch: {invalid json} -->"
assert extract_payload(body) is None
def test_rejects_marker_schema_type_and_binding_drift(self):
mutations = (
("schema 版本", "schema_version", 2),
("schema 类型", "schema_version", "1"),
("仓库 ID", "repository_id", "999"),
("仓库 ID 类型", "repository_id", int(REPOSITORY_ID)),
("workflow ID", "workflow_id", "999"),
("workflow path", "workflow_path", ".github/workflows/other.yml"),
("run ID 类型", "run_id", int(RUN_ID)),
("attempt 前导零", "run_attempt", "01"),
("comment ID", "dispatch_comment_id", "999"),
("artifact ID 类型", "artifact_id", int(ARTIFACT_ID)),
("digest 无前缀", "artifact_digest", "0" * 64),
("digest 大写", "artifact_digest", "sha256:" + "A" * 64),
)
for label, field, value in mutations:
with self.subTest(label=label):
comment, client, _, marker = valid_fixture()
marker[field] = value
_set_marker(comment, marker)
self.assert_rejected(comment, client)
for label, mutate in (
("marker 缺字段", lambda marker: marker.pop("artifact_id")),
("marker 多字段", lambda marker: marker.update(payload={"products": "all"})),
):
with self.subTest(label=label):
comment, client, _, marker = valid_fixture()
mutate(marker)
_set_marker(comment, marker)
self.assert_rejected(comment, client)
def test_extract_payload_rejects_non_dict_integer():
"""JSON 整数不是合法 payload。"""
body = "<!-- eval-dispatch: 1 -->"
assert extract_payload(body) is None
def test_rejects_marker_copied_to_different_comment(self):
comment, client, _, _ = valid_fixture()
comment["id"] = int(DISPATCH_COMMENT_ID) + 1
self.assert_rejected(comment, client)
def test_extract_payload_rejects_non_dict_array():
"""JSON 数组不是合法 payload。"""
body = '<!-- eval-dispatch: [1, 2, 3] -->'
assert extract_payload(body) is None
def test_emits_stable_idempotency_key_across_run_attempts(self):
first_comment, first_client, _, _ = valid_fixture()
first_result = validate_comment(first_comment, client=first_client)
second_comment, second_client, manifest, marker = valid_fixture()
second_attempt = "2"
second_dispatch_comment_id = str(int(DISPATCH_COMMENT_ID) + 1)
second_artifact_id = str(int(ARTIFACT_ID) + 1)
second_comment["id"] = int(second_dispatch_comment_id)
marker.update(
run_attempt=second_attempt,
dispatch_comment_id=second_dispatch_comment_id,
artifact_id=second_artifact_id,
)
manifest.update(
run_attempt=second_attempt,
dispatch_comment_id=second_dispatch_comment_id,
)
second_client.run["run_attempt"] = int(second_attempt)
second_client.artifact.update(
id=int(second_artifact_id),
name=(
f"eval-dispatch-request-{RUN_ID}-{second_attempt}-"
f"{second_dispatch_comment_id}"
),
)
_install_manifest(second_comment, second_client, manifest, marker)
def test_extract_payload_rejects_non_dict_string():
"""JSON 字符串不是合法 payload。"""
body = '<!-- eval-dispatch: "hello" -->'
assert extract_payload(body) is None
second_result = validate_comment(second_comment, client=second_client)
self.assertIsNotNone(first_result)
self.assertIsNotNone(second_result)
self.assertEqual(
first_result["idempotency_key"],
second_result["idempotency_key"],
)
def test_extract_payload_rejects_non_dict_null():
"""JSON null 不是合法 payload。"""
body = "<!-- eval-dispatch: null -->"
assert extract_payload(body) is None
def test_rejects_cross_workflow_and_run_attempt_drift(self):
mutations = (
("run id", ("id",), int(RUN_ID) + 1),
("run attempt", ("run_attempt",), int(RUN_ATTEMPT) + 1),
("workflow id", ("workflow_id",), int(WORKFLOW_ID) + 1),
("workflow path", ("path",), ".github/workflows/other.yml"),
("event", ("event",), "workflow_dispatch"),
("branch", ("head_branch",), "feature/eval"),
("status", ("status",), "in_progress"),
("conclusion", ("conclusion",), "failure"),
("repository id", ("repository", "id"), int(REPOSITORY_ID) + 1),
("repository name", ("repository", "full_name"), "attacker/fork"),
(
"head repository id",
("head_repository", "id"),
int(REPOSITORY_ID) + 1,
),
(
"head repository name",
("head_repository", "full_name"),
"attacker/fork",
),
("head sha 类型", ("head_sha",), None),
)
for label, path, value in mutations:
with self.subTest(label=label):
comment, client, _, _ = valid_fixture()
_set_nested(client.run, path, value)
self.assert_rejected(comment, client)
def test_rejects_artifact_metadata_drift(self):
mutations = (
("artifact id", ("id",), int(ARTIFACT_ID) + 1),
("artifact name", ("name",), "eval-dispatch-request-forged"),
("expired", ("expired",), True),
("expired 类型", ("expired",), 0),
("digest", ("digest",), "sha256:" + "0" * 64),
("size 零", ("size_in_bytes",), 0),
("size 类型", ("size_in_bytes",), "123"),
(
"workflow run id",
("workflow_run", "id"),
int(RUN_ID) + 1,
),
(
"workflow repository id",
("workflow_run", "repository_id"),
int(REPOSITORY_ID) + 1,
),
(
"workflow head repository id",
("workflow_run", "head_repository_id"),
int(REPOSITORY_ID) + 1,
),
("workflow head sha", ("workflow_run", "head_sha"), "c" * 40),
)
for label, path, value in mutations:
with self.subTest(label=label):
comment, client, _, _ = valid_fixture()
_set_nested(client.artifact, path, value)
self.assert_rejected(comment, client)
def test_extract_payload_rejects_numeric_run_id():
"""run_id 为数值类型时拒绝。"""
body = '<!-- eval-dispatch: {"pr_number":"899","pr_head_sha":"a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2","products":"drive","cases_ref":"","run_id":12345} -->'
assert extract_payload(body) is None
comment, client, _, _ = valid_fixture()
client.artifact["workflow_run"] = None
self.assert_rejected(comment, client)
def test_rejects_download_digest_mismatch(self):
comment, client, _, _ = valid_fixture()
client.archive += b"tampered-after-digest"
def test_extract_payload_rejects_numeric_pr_number():
"""pr_number 为数值类型时拒绝。"""
body = '<!-- eval-dispatch: {"pr_number":899,"pr_head_sha":"a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2","products":"drive","cases_ref":"","run_id":"123"} -->'
assert extract_payload(body) is None
self.assert_rejected(comment, client)
def test_rejects_tampered_manifest_payload(self):
for field, value in (
("products", "sheet"),
("cases_ref", "fixtures/other"),
("pr_head_sha", "c" * 40),
):
with self.subTest(field=field):
comment, client, manifest, marker = valid_fixture()
manifest[field] = value
_install_manifest(comment, client, manifest, marker)
self.assert_rejected(comment, client)
def test_extract_payload_rejects_invalid_sha_format():
"""SHA 非 40 位十六进制时拒绝。"""
body = '<!-- eval-dispatch: {"pr_number":"899","pr_head_sha":"short","products":"drive","cases_ref":"","run_id":"123"} -->'
assert extract_payload(body) is None
def test_rejects_cross_workflow_historical_run_reuse(self):
comment, client, manifest, marker = valid_fixture()
historical_run_id = str(int(RUN_ID) - 100)
marker["run_id"] = historical_run_id
manifest["run_id"] = historical_run_id
client.run["id"] = int(historical_run_id)
client.run["workflow_id"] = int(WORKFLOW_ID) + 1
client.artifact["name"] = (
f"eval-dispatch-request-{historical_run_id}-{RUN_ATTEMPT}-"
f"{DISPATCH_COMMENT_ID}"
)
client.artifact["workflow_run"]["id"] = int(historical_run_id)
_install_manifest(comment, client, manifest, marker)
self.assert_rejected(comment, client)
def test_extract_payload_rejects_invalid_products():
"""products 含非法字符时拒绝。"""
body = '<!-- eval-dispatch: {"pr_number":"899","pr_head_sha":"a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2","products":"drive; rm -rf /","cases_ref":"","run_id":"123"} -->'
assert extract_payload(body) is None
def test_rejects_unsafe_or_ambiguous_zip(self):
comment, client, manifest, marker = valid_fixture()
manifest_bytes = (_compact_json(manifest) + "\n").encode("utf-8")
unsafe_archives = (
("非 ZIP", b"not-a-zip"),
(
"路径穿越",
_zip_single_file("../eval-dispatch-request.json", manifest_bytes),
),
(
"绝对路径",
_zip_single_file("/eval-dispatch-request.json", manifest_bytes),
),
(
"多文件",
_zip_files(
(
(ARTIFACT_FILENAME, manifest_bytes),
("second.json", b"{}"),
)
),
),
(
"重复文件名",
_zip_files(
(
(ARTIFACT_FILENAME, manifest_bytes),
(ARTIFACT_FILENAME, manifest_bytes),
)
),
),
("符号链接", _zip_symlink(ARTIFACT_FILENAME, "target.json")),
(
"不允许的压缩算法",
_zip_files(
((ARTIFACT_FILENAME, manifest_bytes),),
compression=zipfile.ZIP_BZIP2,
),
),
(
"解压后 manifest 超限",
_zip_single_file(ARTIFACT_FILENAME, b"x" * (64 * 1024 + 1)),
),
(
"压缩包字节超限",
_zip_files(
((ARTIFACT_FILENAME, b"x" * (64 * 1024)),),
compression=zipfile.ZIP_STORED,
),
),
)
for label, archive in unsafe_archives:
with self.subTest(label=label):
test_comment, test_client, _, test_marker = valid_fixture()
_install_archive(test_comment, test_client, test_marker, archive)
self.assert_rejected(test_comment, test_client)
def test_rejects_duplicate_manifest_json_keys(self):
comment, client, manifest, marker = valid_fixture()
raw_manifest = (
_compact_json(manifest)[:-1] + ',"products":"sheet"}\n'
).encode("utf-8")
archive = _zip_single_file(ARTIFACT_FILENAME, raw_manifest)
_install_archive(comment, client, marker, archive)
def test_extract_payload_rejects_missing_required_field():
"""缺少必要字段时拒绝。"""
body = '<!-- eval-dispatch: {"pr_number":"899","products":"drive","run_id":"123"} -->'
assert extract_payload(body) is None
self.assert_rejected(comment, client)
def test_rejects_duplicate_marker_json_keys(self):
comment, client, _, marker = valid_fixture()
raw_marker = _compact_json(marker)[:-1] + ',"run_id":"1"}'
comment["body"] = f"<!-- eval-dispatch: {raw_marker} -->\n已受理。"
def test_validate_run_id_rejects_non_string():
"""run_id 为非字符串类型时拒绝。"""
assert validate_run_id(12345) is False
assert validate_run_id(None) is False
assert validate_run_id(["123"]) is False
self.assert_rejected(comment, client)
def test_rejects_manifest_exact_field_and_type_violations(self):
mutations = (
("schema 版本", lambda manifest: manifest.update(schema_version=2)),
("schema bool", lambda manifest: manifest.update(schema_version=True)),
(
"repository ID 类型",
lambda manifest: manifest.update(repository_id=int(REPOSITORY_ID)),
),
(
"repository name",
lambda manifest: manifest.update(repository="attacker/fork"),
),
(
"workflow ID 类型",
lambda manifest: manifest.update(workflow_id=int(WORKFLOW_ID)),
),
(
"workflow path",
lambda manifest: manifest.update(
workflow_path=".github/workflows/other.yml"
),
),
("run ID 类型", lambda manifest: manifest.update(run_id=int(RUN_ID))),
("attempt 前导零", lambda manifest: manifest.update(run_attempt="01")),
("source comment 零", lambda manifest: manifest.update(source_comment_id="0")),
(
"dispatch comment 超长",
lambda manifest: manifest.update(dispatch_comment_id="1" * 21),
),
("actor ID 类型", lambda manifest: manifest.update(actor_id=int(ACTOR_ID))),
("actor login", lambda manifest: manifest.update(actor_login="bad_login")),
("actor login 超长", lambda manifest: manifest.update(actor_login="a" * 40)),
("PR number 类型", lambda manifest: manifest.update(pr_number=int(PR_NUMBER))),
("PR SHA 大写", lambda manifest: manifest.update(pr_head_sha="A" * 40)),
("products 空", lambda manifest: manifest.update(products="")),
("products 非法", lambda manifest: manifest.update(products="drive,Doc")),
("products 超长", lambda manifest: manifest.update(products="a" * 513)),
(
"source hash 带前缀",
lambda manifest: manifest.update(
source_body_sha256="sha256:" + "0" * 64
),
),
(
"idempotency 前缀错误",
lambda manifest: manifest.update(
idempotency_key=(
f"eval-dispatch:{REPOSITORY_ID}:{SOURCE_COMMENT_ID}"
)
),
),
("缺字段", lambda manifest: manifest.pop("products")),
("多字段", lambda manifest: manifest.update(payload={})),
)
for label, mutate in mutations:
with self.subTest(label=label):
comment, client, manifest, marker = valid_fixture()
mutate(manifest)
_install_manifest(comment, client, manifest, marker)
self.assert_rejected(comment, client)
def test_validate_run_id_rejects_non_digit_string():
"""run_id 含非数字字符时拒绝。"""
assert validate_run_id("abc") is False
assert validate_run_id("123abc") is False
assert validate_run_id("") is False
def test_rejects_invalid_cases_refs(self):
invalid_refs = (
"-option",
"/absolute",
"trailing/",
"trailing.",
"a..b",
"a//b",
".hidden",
"a/.hidden",
"a.lock",
"a/b.lock",
"contains space",
"中文",
"a" * 1025,
)
for cases_ref in invalid_refs:
with self.subTest(cases_ref=cases_ref[:40]):
comment, client, manifest, marker = valid_fixture()
manifest["cases_ref"] = cases_ref
_install_manifest(comment, client, manifest, marker)
self.assert_rejected(comment, client)
def test_accepts_empty_cases_ref(self):
comment, client, manifest, marker = valid_fixture()
source_body = f"/eval {PRODUCTS} sha={PR_HEAD_SHA}"
manifest["cases_ref"] = ""
manifest["source_body_sha256"] = hashlib.sha256(
source_body.encode("utf-8")
).hexdigest()
client.source_comment["body"] = source_body
_install_manifest(comment, client, manifest, marker)
result = validate_comment(comment, client=client)
self.assertIsNotNone(result)
self.assertEqual(result["cases_ref"], "")
def test_rejects_source_comment_and_actor_drift(self):
mutations = (
("source id", ("source_comment", "id"), int(SOURCE_COMMENT_ID) + 1),
(
"source issue URL",
("source_comment", "issue_url"),
f"https://api.github.com/repos/{REPOSITORY}/issues/951",
),
("source user", ("source_comment", "user"), None),
(
"source actor id",
("source_comment", "user", "id"),
int(ACTOR_ID) + 1,
),
(
"source actor login",
("source_comment", "user", "login"),
"attacker",
),
("source body 类型", ("source_comment", "body"), None),
("source body hash", ("source_comment", "body"), SOURCE_BODY + "\nchanged"),
("run actor", ("run", "actor"), None),
("run actor id", ("run", "actor", "id"), int(ACTOR_ID) + 1),
("run actor login", ("run", "actor", "login"), "attacker"),
)
for label, path, value in mutations:
with self.subTest(label=label):
comment, client, _, _ = valid_fixture()
target = client.source_comment if path[0] == "source_comment" else client.run
_set_nested(target, path[1:], value)
self.assert_rejected(comment, client)
def test_rejects_source_command_payload_mismatch_even_with_matching_hash(self):
comment, client, manifest, marker = valid_fixture()
changed_body = f"/eval sheet sha={PR_HEAD_SHA} cases={CASES_REF}"
client.source_comment["body"] = changed_body
manifest["source_body_sha256"] = hashlib.sha256(
changed_body.encode("utf-8")
).hexdigest()
_install_manifest(comment, client, manifest, marker)
self.assert_rejected(comment, client)
def test_rejects_dispatch_issue_url_drift(self):
comment, client, _, _ = valid_fixture()
comment["issue_url"] = f"https://api.github.com/repos/{REPOSITORY}/issues/951"
self.assert_rejected(comment, client)
def test_rejects_pull_request_drift(self):
mutations = (
("PR number", ("number",), int(PR_NUMBER) + 1),
("PR closed", ("state",), "closed"),
("PR head", ("head", "sha"), "c" * 40),
("PR head 缺失", ("head",), None),
)
for label, path, value in mutations:
with self.subTest(label=label):
comment, client, _, _ = valid_fixture()
_set_nested(client.pr, path, value)
self.assert_rejected(comment, client)
def test_api_errors_and_invalid_response_shapes_fail_closed(self):
def explode(*_args, **_kwargs):
raise RuntimeError("simulated API failure")
for method_name in (
"get_run_attempt",
"get_artifact",
"download_artifact",
"get_comment",
"get_pull_request",
):
with self.subTest(method=method_name):
comment, client, _, _ = valid_fixture()
setattr(client, method_name, explode)
self.assert_rejected(comment, client)
for label, attribute, value in (
("run list", "run", []),
("artifact null", "artifact", None),
("archive text", "archive", "not bytes"),
("source list", "source_comment", []),
("PR list", "pr", []),
):
with self.subTest(label=label):
comment, client, _, _ = valid_fixture()
setattr(client, attribute, value)
self.assert_rejected(comment, client)
def test_cli_client_uses_attempt_and_artifact_id_endpoints(self):
client = GitHubCLIClient()
with mock.patch("eval_poll_validate.subprocess.run") as run:
run.return_value = mock.Mock(returncode=0, stdout="{}")
client.get_run_attempt(RUN_ID, RUN_ATTEMPT)
run.assert_called_once_with(
[
"gh",
"api",
(
f"repos/{REPOSITORY}/actions/runs/{RUN_ID}/"
f"attempts/{RUN_ATTEMPT}"
),
],
capture_output=True,
text=True,
timeout=20,
)
with mock.patch("eval_poll_validate.subprocess.run") as run:
run.return_value = mock.Mock(returncode=0, stdout=b"zip")
self.assertEqual(client.download_artifact(ARTIFACT_ID), b"zip")
run.assert_called_once_with(
[
"gh",
"api",
f"repos/{REPOSITORY}/actions/artifacts/{ARTIFACT_ID}/zip",
],
capture_output=True,
timeout=30,
)
def test_main_rejects_duplicate_stdin_json_keys_without_network(self):
with mock.patch(
"eval_poll_validate.sys.stdin",
io.StringIO('{"body":{},"body":{}}'),
), mock.patch("eval_poll_validate.subprocess.run") as run:
self.assertEqual(validate_main(), 1)
run.assert_not_called()
if __name__ == "__main__":
test_rejects_regular_user_comment()
test_rejects_wrong_bot_login()
test_rejects_bot_without_app_signature()
test_rejects_bot_with_wrong_app_slug()
test_accepts_legitimate_github_actions_comment()
test_extract_payload_valid()
test_extract_payload_no_marker()
test_extract_payload_malformed_json()
test_extract_payload_rejects_non_dict_integer()
test_extract_payload_rejects_non_dict_array()
test_extract_payload_rejects_non_dict_string()
test_extract_payload_rejects_non_dict_null()
test_extract_payload_rejects_numeric_run_id()
test_extract_payload_rejects_numeric_pr_number()
test_extract_payload_rejects_invalid_sha_format()
test_extract_payload_rejects_invalid_products()
test_extract_payload_rejects_missing_required_field()
test_validate_run_id_rejects_non_string()
test_validate_run_id_rejects_non_digit_string()
print("All eval_poll_validate tests passed.")
unittest.main(verbosity=2)
test/scripts/eval_dispatch_workflow_test.go
+95 -2
View File
@@ -31,8 +31,6 @@ func TestEvalDispatchWorkflowUsesRepositoryPermissionAndReviewedSHA(t *testing.T
"EVAL_ALLOWLIST_PATH: .github/eval-allowlist.txt",
"REVIEWED_SHA: ${{ steps.parse.outputs.reviewed_sha }}",
"eval_dispatch_guard.py head",
"RUN_ID: ${{ github.run_id }}",
"eval_poll_validate.py",
} {
if !strings.Contains(workflow, want) {
t.Errorf("eval-dispatch workflow missing security contract %q", want)
@@ -46,6 +44,101 @@ func TestEvalDispatchWorkflowUsesRepositoryPermissionAndReviewedSHA(t *testing.T
}
}
func TestEvalDispatchWorkflowPublishesArtifactBoundRequest(t *testing.T) {
t.Parallel()
root, err := filepath.Abs(filepath.Join("..", ".."))
if err != nil {
t.Fatalf("resolve repository root: %v", err)
}
data, err := os.ReadFile(filepath.Join(root, ".github", "workflows", "eval-dispatch.yml"))
if err != nil {
t.Fatalf("read eval-dispatch workflow: %v", err)
}
workflow := string(data)
orderedSteps := []string{
"- name: Create dispatch placeholder",
"- name: Build dispatch request manifest",
"- name: Upload dispatch request manifest",
"- name: Finalize dispatch marker",
}
previous := -1
for _, step := range orderedSteps {
index := strings.Index(workflow, step)
if index < 0 {
t.Fatalf("eval-dispatch workflow missing step %q", step)
}
if index <= previous {
t.Fatalf("eval-dispatch workflow step %q is out of order", step)
}
previous = index
}
placeholderStart := strings.Index(workflow, orderedSteps[0])
manifestStart := strings.Index(workflow, orderedSteps[1])
if strings.Contains(workflow[placeholderStart:manifestStart], "<!-- eval-dispatch:") {
t.Fatal("dispatch placeholder must not expose a consumable marker before the manifest exists")
}
if count := strings.Count(workflow, "<!-- eval-dispatch:"); count != 1 {
t.Fatalf("eval-dispatch workflow marker count = %d, want exactly one finalized marker", count)
}
for _, want := range []string{
"REPOSITORY_ID: '1187709537'",
"WORKFLOW_ID: '331725458'",
"WORKFLOW_PATH: .github/workflows/eval-dispatch.yml",
"RUN_ID: ${{ github.run_id }}",
"RUN_ATTEMPT: ${{ github.run_attempt }}",
"SOURCE_COMMENT_ID: ${{ github.event.comment.id }}",
"DISPATCH_COMMENT_ID: ${{ steps.placeholder.outputs.comment_id }}",
"idempotency_key: $idempotency_key",
"actions/upload-artifact@v4",
"name: eval-dispatch-request-${{ github.run_id }}-${{ github.run_attempt }}-${{ steps.placeholder.outputs.comment_id }}",
"path: ${{ runner.temp }}/eval-dispatch-request.json",
"if-no-files-found: error",
"retention-days: 1",
"overwrite: false",
"ARTIFACT_ID: ${{ steps.artifact.outputs.artifact-id }}",
"ARTIFACT_DIGEST: ${{ steps.artifact.outputs.artifact-digest }}",
"<!-- eval-dispatch: ${marker_json} -->",
"-X PATCH",
"issues/comments/${DISPATCH_COMMENT_ID}",
} {
if !strings.Contains(workflow, want) {
t.Errorf("eval-dispatch workflow missing artifact contract %q", want)
}
}
for _, forbidden := range []string{
"EVAL_TRIGGER_URL",
"EVAL_TRIGGER_TOKEN",
} {
if strings.Contains(workflow, forbidden) {
t.Errorf("eval-dispatch workflow exposes retired direct-trigger detail %q", forbidden)
}
}
}
func TestEvalPollValidatePython(t *testing.T) {
t.Parallel()
root, err := filepath.Abs(filepath.Join("..", ".."))
if err != nil {
t.Fatalf("resolve repository root: %v", err)
}
cmd := exec.Command(
"python3",
"-B",
filepath.Join(root, "scripts", "ci", "test_eval_poll_validate.py"),
)
cmd.Dir = root
output, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("eval poll validator tests failed: %v\n%s", err, output)
}
}
func TestEvalDispatchRejectsLowRepositoryPermissions(t *testing.T) {
t.Parallel()