feat(ack): add project delivery workflow
This commit is contained in:
Executable
+809
@@ -0,0 +1,809 @@
|
||||
#!/usr/bin/env python3
|
||||
"""校验 ACK 项目交付契约。
|
||||
|
||||
权威结构位于 templates/delivery.schema.json。jsonschema 是可选依赖；内置规则始终
|
||||
检查引用、步骤顺序、默认 profile 安全边界、敏感信息和仓库内入口路径。
|
||||
|
||||
用法:
|
||||
python3 validate_delivery.py docs/ack/delivery.yaml
|
||||
python3 validate_delivery.py docs/ack/delivery.yaml \
|
||||
--tasks docs/ack/tasks.yaml --project-root <project-root>
|
||||
|
||||
退出码: 0 通过 / 1 校验失败 / 2 环境或用法错误。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import stat
|
||||
import sys
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any
|
||||
|
||||
from yaml_subset import (
|
||||
DuplicateKeyError,
|
||||
YamlSubsetError,
|
||||
load_json_unique,
|
||||
load_yaml_subset,
|
||||
make_unique_pyyaml_loader,
|
||||
)
|
||||
|
||||
|
||||
ID_RE = re.compile(r"^[a-z][a-z0-9-]{0,63}$")
|
||||
RELATIVE_PATH_RE = re.compile(r"^[A-Za-z0-9._/*?+-]+$")
|
||||
PLATFORM_RE = re.compile(r"^[a-z0-9]+/[A-Za-z0-9._-]+$")
|
||||
SECRET_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,127}$")
|
||||
REMOTE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
|
||||
|
||||
TOP_LEVEL_FIELDS = {
|
||||
"version",
|
||||
"updatedAt",
|
||||
"project",
|
||||
"enabled",
|
||||
"defaultProfile",
|
||||
"entrypoints",
|
||||
"artifacts",
|
||||
"destinations",
|
||||
"environments",
|
||||
"profiles",
|
||||
}
|
||||
ENTRYPOINT_FIELDS = {
|
||||
"kind",
|
||||
"target",
|
||||
"function",
|
||||
"path",
|
||||
"args",
|
||||
"requiredSecrets",
|
||||
"workingDirectory",
|
||||
"timeoutSeconds",
|
||||
}
|
||||
ARTIFACT_FIELDS = {"type", "build", "outputs", "image", "platforms"}
|
||||
DESTINATION_FIELDS = {
|
||||
"type",
|
||||
"channel",
|
||||
"registry",
|
||||
"repository",
|
||||
"endpoint",
|
||||
"artifactName",
|
||||
"upload",
|
||||
}
|
||||
ENVIRONMENT_FIELDS = {
|
||||
"type",
|
||||
"classification",
|
||||
"target",
|
||||
"deploy",
|
||||
"healthCheck",
|
||||
"rollback",
|
||||
"mutex",
|
||||
}
|
||||
PROFILE_FIELDS = {"stopAt", "steps"}
|
||||
STEP_FIELDS = {
|
||||
"id",
|
||||
"action",
|
||||
"entrypoint",
|
||||
"artifact",
|
||||
"destination",
|
||||
"environment",
|
||||
"gate",
|
||||
"draft",
|
||||
"remote",
|
||||
"baseBranch",
|
||||
}
|
||||
|
||||
ENTRYPOINT_KINDS = {"make", "just", "task", "dagger", "script"}
|
||||
ARTIFACT_TYPES = {"deb", "oci-image", "file"}
|
||||
DESTINATION_TYPES = {"apt-repository", "oci-registry", "ci-artifact"}
|
||||
CHANNELS = {"preview", "staging", "stable"}
|
||||
ENVIRONMENT_TYPES = {"ssh-host", "docker-compose", "kubernetes", "custom"}
|
||||
CLASSIFICATIONS = {"development", "staging", "production"}
|
||||
STOP_POINTS = {"verified", "review_ready", "released"}
|
||||
ACTIONS = {
|
||||
"verify",
|
||||
"pull-request",
|
||||
"build",
|
||||
"publish",
|
||||
"deploy",
|
||||
"health-check",
|
||||
"approval",
|
||||
"mark-ready",
|
||||
}
|
||||
ACTION_FIELDS = {
|
||||
"verify": {"entrypoint"},
|
||||
"pull-request": {"draft", "remote", "baseBranch"},
|
||||
"build": {"artifact"},
|
||||
"publish": {"artifact", "destination"},
|
||||
"deploy": {"artifact", "environment"},
|
||||
"health-check": {"environment"},
|
||||
"approval": {"gate"},
|
||||
"mark-ready": set(),
|
||||
}
|
||||
|
||||
SECRET_PATTERNS = (
|
||||
("private key", re.compile(r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----")),
|
||||
("GitHub token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b")),
|
||||
("OpenAI-style token", re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b")),
|
||||
("AWS access key", re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b")),
|
||||
("URL credentials", re.compile(r"https?://[^/\s:@]+:[^/\s@]+@")),
|
||||
(
|
||||
"inline secret assignment",
|
||||
re.compile(
|
||||
r"(?i)\b(?:api[_-]?key|access[_-]?token|password|secret|token)"
|
||||
r"\s*[:=]\s*[\"']?[^\s,\"']{8,}"
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _nonempty(value: Any) -> bool:
|
||||
return isinstance(value, str) and bool(value.strip())
|
||||
|
||||
|
||||
def _mapping(value: Any) -> bool:
|
||||
return isinstance(value, dict)
|
||||
|
||||
|
||||
def _reject_unknown(
|
||||
value: dict[str, Any],
|
||||
allowed: set[str],
|
||||
where: str,
|
||||
errors: list[str],
|
||||
) -> None:
|
||||
for field in sorted(set(value) - allowed):
|
||||
errors.append(f"{where}: 未知字段 {field!r}")
|
||||
|
||||
|
||||
def _load_document(path: Path, label: str) -> dict[str, Any]:
|
||||
try:
|
||||
content = path.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
sys.stderr.write(f"{label}读取失败: {exc}\n")
|
||||
raise SystemExit(1)
|
||||
|
||||
if path.suffix.lower() == ".json":
|
||||
try:
|
||||
data = load_json_unique(content)
|
||||
except (json.JSONDecodeError, DuplicateKeyError) as exc:
|
||||
sys.stderr.write(f"{label} JSON 解析失败: {exc}\n")
|
||||
raise SystemExit(1)
|
||||
else:
|
||||
try:
|
||||
import yaml # type: ignore
|
||||
except ImportError:
|
||||
try:
|
||||
data = load_yaml_subset(content)
|
||||
except YamlSubsetError as exc:
|
||||
sys.stderr.write(f"{label} YAML 子集解析失败: {exc}\n")
|
||||
raise SystemExit(1)
|
||||
else:
|
||||
try:
|
||||
data = yaml.load(content, Loader=make_unique_pyyaml_loader(yaml))
|
||||
except yaml.YAMLError as exc: # type: ignore
|
||||
sys.stderr.write(f"{label} YAML 解析失败: {exc}\n")
|
||||
raise SystemExit(1)
|
||||
|
||||
if not isinstance(data, dict):
|
||||
sys.stderr.write(f"{label}顶层必须是对象（mapping）\n")
|
||||
raise SystemExit(1)
|
||||
return data
|
||||
|
||||
|
||||
def _safe_relative_path(value: Any, *, allow_glob: bool = False) -> bool:
|
||||
if not _nonempty(value) or value.startswith("/") or "\\" in value:
|
||||
return False
|
||||
if not RELATIVE_PATH_RE.fullmatch(value):
|
||||
return False
|
||||
if not allow_glob and any(marker in value for marker in "*?"):
|
||||
return False
|
||||
parts = PurePosixPath(value).parts
|
||||
return ".." not in parts and all(part not in {"", "/"} for part in parts)
|
||||
|
||||
|
||||
def _safe_branch_name(value: Any) -> bool:
|
||||
if not _nonempty(value) or len(value) > 255:
|
||||
return False
|
||||
if value == "@" or value.startswith(("/", ".", "-")):
|
||||
return False
|
||||
if value.endswith(("/", ".", ".lock")):
|
||||
return False
|
||||
if "@{" in value or ".." in value or "//" in value:
|
||||
return False
|
||||
return re.search(r"[\x00-\x20\x7f~^:?*\[\\]", value) is None
|
||||
|
||||
|
||||
def _validate_path_binding(
|
||||
project_root: Path,
|
||||
relative_path: str,
|
||||
where: str,
|
||||
*,
|
||||
expected: str,
|
||||
) -> list[str]:
|
||||
errors: list[str] = []
|
||||
current = project_root
|
||||
parts = PurePosixPath(relative_path).parts
|
||||
if relative_path == ".":
|
||||
parts = ()
|
||||
for index, part in enumerate(parts):
|
||||
current = current / part
|
||||
try:
|
||||
metadata = current.lstat()
|
||||
except FileNotFoundError:
|
||||
return [f"{where}: 路径不存在: {relative_path!r}"]
|
||||
except OSError as exc:
|
||||
return [f"{where}: 路径不可访问: {relative_path!r}: {exc}"]
|
||||
if stat.S_ISLNK(metadata.st_mode):
|
||||
return [f"{where}: 路径不能包含 symlink: {relative_path!r}"]
|
||||
if index < len(parts) - 1 and not stat.S_ISDIR(metadata.st_mode):
|
||||
return [f"{where}: 中间路径不是目录: {relative_path!r}"]
|
||||
|
||||
try:
|
||||
metadata
|
||||
except UnboundLocalError:
|
||||
metadata = project_root.lstat()
|
||||
if expected == "directory" and not stat.S_ISDIR(metadata.st_mode):
|
||||
errors.append(f"{where}: 必须指向目录: {relative_path!r}")
|
||||
if expected == "executable":
|
||||
if not stat.S_ISREG(metadata.st_mode):
|
||||
errors.append(f"{where}: 必须指向普通文件: {relative_path!r}")
|
||||
elif metadata.st_mode & 0o111 == 0:
|
||||
errors.append(f"{where}: 脚本不可执行: {relative_path!r}")
|
||||
return errors
|
||||
|
||||
|
||||
def _scan_secrets(value: Any, where: str, errors: list[str]) -> None:
|
||||
if isinstance(value, dict):
|
||||
for key, item in value.items():
|
||||
_scan_secrets(item, f"{where}.{key}", errors)
|
||||
return
|
||||
if isinstance(value, list):
|
||||
for index, item in enumerate(value):
|
||||
_scan_secrets(item, f"{where}[{index}]", errors)
|
||||
return
|
||||
if not isinstance(value, str):
|
||||
return
|
||||
for label, pattern in SECRET_PATTERNS:
|
||||
if pattern.search(value):
|
||||
errors.append(f"{where}: 疑似包含敏感信息（{label}）")
|
||||
|
||||
|
||||
def _validate_ids(values: Any, where: str, errors: list[str]) -> dict[str, Any]:
|
||||
if not isinstance(values, dict):
|
||||
errors.append(f"{where}: 必须是对象")
|
||||
return {}
|
||||
for key in values:
|
||||
if not isinstance(key, str) or ID_RE.fullmatch(key) is None:
|
||||
errors.append(f"{where}: ID {key!r} 必须使用小写连字符格式")
|
||||
return values
|
||||
|
||||
|
||||
def _validate_entrypoints(
|
||||
values: dict[str, Any],
|
||||
errors: list[str],
|
||||
project_root: Path | None,
|
||||
) -> None:
|
||||
for entrypoint_id, value in values.items():
|
||||
where = f"entrypoints.{entrypoint_id}"
|
||||
if not _mapping(value):
|
||||
errors.append(f"{where}: 必须是对象")
|
||||
continue
|
||||
_reject_unknown(value, ENTRYPOINT_FIELDS, where, errors)
|
||||
kind = value.get("kind")
|
||||
if kind not in ENTRYPOINT_KINDS:
|
||||
errors.append(f"{where}.kind: 必须是 {sorted(ENTRYPOINT_KINDS)}")
|
||||
required_selector = {
|
||||
"make": "target",
|
||||
"just": "target",
|
||||
"task": "target",
|
||||
"dagger": "function",
|
||||
"script": "path",
|
||||
}.get(kind)
|
||||
for selector in ("target", "function", "path"):
|
||||
if selector == required_selector:
|
||||
if not _nonempty(value.get(selector)):
|
||||
errors.append(f"{where}.{selector}: {kind} 入口必须填写非空值")
|
||||
elif selector in value:
|
||||
errors.append(f"{where}.{selector}: kind={kind!r} 不允许此字段")
|
||||
|
||||
args = value.get("args")
|
||||
if not isinstance(args, list) or any(not isinstance(item, str) for item in args):
|
||||
errors.append(f"{where}.args: 必须是字符串列表")
|
||||
required_secrets = value.get("requiredSecrets")
|
||||
if (
|
||||
not isinstance(required_secrets, list)
|
||||
or any(
|
||||
not isinstance(item, str) or SECRET_NAME_RE.fullmatch(item) is None
|
||||
for item in required_secrets
|
||||
)
|
||||
or (
|
||||
isinstance(required_secrets, list)
|
||||
and len(required_secrets) != len(set(required_secrets))
|
||||
)
|
||||
):
|
||||
errors.append(
|
||||
f"{where}.requiredSecrets: 必须是唯一的大写 secret 名称列表"
|
||||
)
|
||||
working_directory = value.get("workingDirectory")
|
||||
if not _safe_relative_path(working_directory):
|
||||
errors.append(f"{where}.workingDirectory: 必须是安全的仓库内相对路径")
|
||||
timeout = value.get("timeoutSeconds")
|
||||
if (
|
||||
not isinstance(timeout, int)
|
||||
or isinstance(timeout, bool)
|
||||
or not 1 <= timeout <= 86400
|
||||
):
|
||||
errors.append(f"{where}.timeoutSeconds: 必须是 1..86400 的整数")
|
||||
|
||||
if kind == "script" and not _safe_relative_path(value.get("path")):
|
||||
errors.append(f"{where}.path: 必须是安全的仓库内相对路径")
|
||||
if project_root is not None:
|
||||
if _safe_relative_path(working_directory):
|
||||
errors.extend(
|
||||
_validate_path_binding(
|
||||
project_root,
|
||||
working_directory,
|
||||
f"{where}.workingDirectory",
|
||||
expected="directory",
|
||||
)
|
||||
)
|
||||
if kind == "script" and _safe_relative_path(value.get("path")):
|
||||
errors.extend(
|
||||
_validate_path_binding(
|
||||
project_root,
|
||||
value["path"],
|
||||
f"{where}.path",
|
||||
expected="executable",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _validate_artifacts(
|
||||
values: dict[str, Any],
|
||||
entrypoints: dict[str, Any],
|
||||
errors: list[str],
|
||||
) -> None:
|
||||
for artifact_id, value in values.items():
|
||||
where = f"artifacts.{artifact_id}"
|
||||
if not _mapping(value):
|
||||
errors.append(f"{where}: 必须是对象")
|
||||
continue
|
||||
_reject_unknown(value, ARTIFACT_FIELDS, where, errors)
|
||||
artifact_type = value.get("type")
|
||||
if artifact_type not in ARTIFACT_TYPES:
|
||||
errors.append(f"{where}.type: 必须是 {sorted(ARTIFACT_TYPES)}")
|
||||
build = value.get("build")
|
||||
if build not in entrypoints:
|
||||
errors.append(f"{where}.build: 未定义 entrypoint {build!r}")
|
||||
outputs = value.get("outputs")
|
||||
if artifact_type in {"deb", "file"}:
|
||||
if (
|
||||
not isinstance(outputs, list)
|
||||
or not outputs
|
||||
or any(not _safe_relative_path(item, allow_glob=True) for item in outputs)
|
||||
):
|
||||
errors.append(f"{where}.outputs: deb/file 必须填写安全的产物路径列表")
|
||||
if "image" in value or "platforms" in value:
|
||||
errors.append(f"{where}: deb/file 不允许 image 或 platforms")
|
||||
if artifact_type == "oci-image":
|
||||
if not _nonempty(value.get("image")):
|
||||
errors.append(f"{where}.image: oci-image 必须填写镜像名")
|
||||
platforms = value.get("platforms")
|
||||
if (
|
||||
not isinstance(platforms, list)
|
||||
or not platforms
|
||||
or any(not isinstance(item, str) or PLATFORM_RE.fullmatch(item) is None for item in platforms)
|
||||
or len(platforms) != len(set(platforms))
|
||||
):
|
||||
errors.append(f"{where}.platforms: 必须是唯一的 os/arch 列表")
|
||||
if "outputs" in value:
|
||||
errors.append(f"{where}: oci-image 不允许 outputs")
|
||||
|
||||
|
||||
def _validate_destinations(
|
||||
values: dict[str, Any],
|
||||
entrypoints: dict[str, Any],
|
||||
errors: list[str],
|
||||
) -> None:
|
||||
for destination_id, value in values.items():
|
||||
where = f"destinations.{destination_id}"
|
||||
if not _mapping(value):
|
||||
errors.append(f"{where}: 必须是对象")
|
||||
continue
|
||||
_reject_unknown(value, DESTINATION_FIELDS, where, errors)
|
||||
destination_type = value.get("type")
|
||||
if destination_type not in DESTINATION_TYPES:
|
||||
errors.append(f"{where}.type: 必须是 {sorted(DESTINATION_TYPES)}")
|
||||
type_fields = {
|
||||
"apt-repository": {"endpoint", "repository"},
|
||||
"oci-registry": {"registry", "repository"},
|
||||
"ci-artifact": {"artifactName"},
|
||||
}.get(destination_type, set())
|
||||
allowed_fields = {"type", "channel", "upload"} | type_fields
|
||||
for field in sorted(set(value) - allowed_fields):
|
||||
errors.append(f"{where}.{field}: type={destination_type!r} 不允许此字段")
|
||||
if value.get("channel") not in CHANNELS:
|
||||
errors.append(f"{where}.channel: 必须是 {sorted(CHANNELS)}")
|
||||
upload = value.get("upload")
|
||||
if upload is not None and upload not in entrypoints:
|
||||
errors.append(f"{where}.upload: 未定义 entrypoint {upload!r}")
|
||||
if destination_type == "apt-repository":
|
||||
if not _nonempty(value.get("endpoint")):
|
||||
errors.append(f"{where}.endpoint: APT 目标必须填写服务地址")
|
||||
if not _nonempty(value.get("repository")):
|
||||
errors.append(f"{where}.repository: APT 目标必须填写仓库名")
|
||||
if destination_type == "oci-registry":
|
||||
for field in ("registry", "repository"):
|
||||
if not _nonempty(value.get(field)):
|
||||
errors.append(f"{where}.{field}: OCI 目标必须填写非空值")
|
||||
if destination_type == "ci-artifact" and not _nonempty(value.get("artifactName")):
|
||||
errors.append(f"{where}.artifactName: CI artifact 必须填写名称")
|
||||
|
||||
|
||||
def _validate_environments(
|
||||
values: dict[str, Any],
|
||||
entrypoints: dict[str, Any],
|
||||
errors: list[str],
|
||||
) -> None:
|
||||
for environment_id, value in values.items():
|
||||
where = f"environments.{environment_id}"
|
||||
if not _mapping(value):
|
||||
errors.append(f"{where}: 必须是对象")
|
||||
continue
|
||||
_reject_unknown(value, ENVIRONMENT_FIELDS, where, errors)
|
||||
if value.get("type") not in ENVIRONMENT_TYPES:
|
||||
errors.append(f"{where}.type: 必须是 {sorted(ENVIRONMENT_TYPES)}")
|
||||
classification = value.get("classification")
|
||||
if classification not in CLASSIFICATIONS:
|
||||
errors.append(f"{where}.classification: 必须是 {sorted(CLASSIFICATIONS)}")
|
||||
if not _nonempty(value.get("target")):
|
||||
errors.append(f"{where}.target: 必须是非空目标别名")
|
||||
for field in ("deploy", "healthCheck"):
|
||||
reference = value.get(field)
|
||||
if reference not in entrypoints:
|
||||
errors.append(f"{where}.{field}: 未定义 entrypoint {reference!r}")
|
||||
rollback = value.get("rollback")
|
||||
if rollback is not None and rollback not in entrypoints:
|
||||
errors.append(f"{where}.rollback: 未定义 entrypoint {rollback!r}")
|
||||
if classification == "production" and rollback is None:
|
||||
errors.append(f"{where}.rollback: production 环境必须提供回滚入口")
|
||||
if not _nonempty(value.get("mutex")):
|
||||
errors.append(f"{where}.mutex: 必须填写部署互斥锁 ID")
|
||||
|
||||
|
||||
def _artifact_destination_compatible(artifact_type: str, destination_type: str) -> bool:
|
||||
return destination_type in {
|
||||
"deb": {"apt-repository", "ci-artifact"},
|
||||
"oci-image": {"oci-registry", "ci-artifact"},
|
||||
"file": {"ci-artifact"},
|
||||
}.get(artifact_type, set())
|
||||
|
||||
|
||||
def _validate_profiles(
|
||||
values: dict[str, Any],
|
||||
*,
|
||||
default_profile: Any,
|
||||
entrypoints: dict[str, Any],
|
||||
artifacts: dict[str, Any],
|
||||
destinations: dict[str, Any],
|
||||
environments: dict[str, Any],
|
||||
errors: list[str],
|
||||
) -> None:
|
||||
for profile_id, value in values.items():
|
||||
where = f"profiles.{profile_id}"
|
||||
if not _mapping(value):
|
||||
errors.append(f"{where}: 必须是对象")
|
||||
continue
|
||||
_reject_unknown(value, PROFILE_FIELDS, where, errors)
|
||||
stop_at = value.get("stopAt")
|
||||
if stop_at not in STOP_POINTS:
|
||||
errors.append(f"{where}.stopAt: 必须是 {sorted(STOP_POINTS)}")
|
||||
steps = value.get("steps")
|
||||
if not isinstance(steps, list):
|
||||
errors.append(f"{where}.steps: 必须是列表")
|
||||
continue
|
||||
|
||||
seen_step_ids: set[str] = set()
|
||||
built_artifacts: set[str] = set()
|
||||
published_artifacts: set[str] = set()
|
||||
deployed_environments: set[str] = set()
|
||||
approvals: set[str] = set()
|
||||
has_pull_request = False
|
||||
has_mark_ready = False
|
||||
|
||||
for index, step in enumerate(steps):
|
||||
step_where = f"{where}.steps[{index}]"
|
||||
if not _mapping(step):
|
||||
errors.append(f"{step_where}: 必须是对象")
|
||||
continue
|
||||
_reject_unknown(step, STEP_FIELDS, step_where, errors)
|
||||
step_id = step.get("id")
|
||||
if not isinstance(step_id, str) or ID_RE.fullmatch(step_id) is None:
|
||||
errors.append(f"{step_where}.id: 必须使用小写连字符格式")
|
||||
elif step_id in seen_step_ids:
|
||||
errors.append(f"{step_where}.id: 不能重复 {step_id!r}")
|
||||
else:
|
||||
seen_step_ids.add(step_id)
|
||||
|
||||
action = step.get("action")
|
||||
if action not in ACTIONS:
|
||||
errors.append(f"{step_where}.action: 必须是 {sorted(ACTIONS)}")
|
||||
continue
|
||||
required_fields = ACTION_FIELDS[action]
|
||||
for field in sorted(required_fields):
|
||||
if field not in step:
|
||||
errors.append(f"{step_where}.{field}: action={action!r} 时必填")
|
||||
allowed_fields = {"id", "action"} | required_fields
|
||||
for field in sorted(set(step) - allowed_fields):
|
||||
errors.append(f"{step_where}.{field}: action={action!r} 不允许此字段")
|
||||
|
||||
if action == "verify" and step.get("entrypoint") not in entrypoints:
|
||||
errors.append(
|
||||
f"{step_where}.entrypoint: 未定义 entrypoint {step.get('entrypoint')!r}"
|
||||
)
|
||||
if action == "pull-request":
|
||||
if not isinstance(step.get("draft"), bool):
|
||||
errors.append(f"{step_where}.draft: 必须是布尔值")
|
||||
if (
|
||||
not isinstance(step.get("remote"), str)
|
||||
or REMOTE_RE.fullmatch(step["remote"]) is None
|
||||
):
|
||||
errors.append(f"{step_where}.remote: 必须是安全的 Git remote 名称")
|
||||
if not _safe_branch_name(step.get("baseBranch")):
|
||||
errors.append(f"{step_where}.baseBranch: 必须是安全的 Git 分支名")
|
||||
has_pull_request = True
|
||||
if action == "build":
|
||||
artifact_id = step.get("artifact")
|
||||
if artifact_id not in artifacts:
|
||||
errors.append(f"{step_where}.artifact: 未定义 artifact {artifact_id!r}")
|
||||
else:
|
||||
built_artifacts.add(artifact_id)
|
||||
if action == "publish":
|
||||
artifact_id = step.get("artifact")
|
||||
destination_id = step.get("destination")
|
||||
if artifact_id not in artifacts:
|
||||
errors.append(f"{step_where}.artifact: 未定义 artifact {artifact_id!r}")
|
||||
elif artifact_id not in built_artifacts:
|
||||
errors.append(f"{step_where}: publish 前必须先 build {artifact_id!r}")
|
||||
if destination_id not in destinations:
|
||||
errors.append(
|
||||
f"{step_where}.destination: 未定义 destination {destination_id!r}"
|
||||
)
|
||||
elif artifact_id in artifacts:
|
||||
artifact_type = artifacts[artifact_id].get("type")
|
||||
destination_type = destinations[destination_id].get("type")
|
||||
if not _artifact_destination_compatible(artifact_type, destination_type):
|
||||
errors.append(
|
||||
f"{step_where}: artifact {artifact_type!r} 不能发布到 "
|
||||
f"{destination_type!r}"
|
||||
)
|
||||
if destinations[destination_id].get("channel") == "stable" and "release" not in approvals:
|
||||
errors.append(f"{step_where}: stable 发布前必须有 release approval")
|
||||
published_artifacts.add(artifact_id)
|
||||
if action == "deploy":
|
||||
artifact_id = step.get("artifact")
|
||||
environment_id = step.get("environment")
|
||||
if artifact_id not in artifacts:
|
||||
errors.append(f"{step_where}.artifact: 未定义 artifact {artifact_id!r}")
|
||||
elif artifact_id not in built_artifacts:
|
||||
errors.append(f"{step_where}: deploy 前必须先 build {artifact_id!r}")
|
||||
if environment_id not in environments:
|
||||
errors.append(
|
||||
f"{step_where}.environment: 未定义 environment {environment_id!r}"
|
||||
)
|
||||
else:
|
||||
classification = environments[environment_id].get("classification")
|
||||
if classification == "production" and "production" not in approvals:
|
||||
errors.append(f"{step_where}: production 部署前必须有 production approval")
|
||||
if (
|
||||
classification == "production"
|
||||
and artifact_id not in published_artifacts
|
||||
):
|
||||
errors.append(f"{step_where}: production 部署前必须先 publish 同一产物")
|
||||
deployed_environments.add(environment_id)
|
||||
if action == "health-check":
|
||||
environment_id = step.get("environment")
|
||||
if environment_id not in environments:
|
||||
errors.append(
|
||||
f"{step_where}.environment: 未定义 environment {environment_id!r}"
|
||||
)
|
||||
elif environment_id not in deployed_environments:
|
||||
errors.append(
|
||||
f"{step_where}: health-check 前必须先 deploy {environment_id!r}"
|
||||
)
|
||||
if action == "approval":
|
||||
gate = step.get("gate")
|
||||
if gate not in {"release", "production"}:
|
||||
errors.append(f"{step_where}.gate: 必须是 release/production")
|
||||
else:
|
||||
if gate == "release" and not built_artifacts:
|
||||
errors.append(f"{step_where}: release approval 前必须先 build 产物")
|
||||
if gate == "production" and not published_artifacts:
|
||||
errors.append(f"{step_where}: production approval 前必须先 publish 产物")
|
||||
approvals.add(gate)
|
||||
if action == "mark-ready":
|
||||
if not has_pull_request:
|
||||
errors.append(f"{step_where}: mark-ready 前必须先创建 pull-request")
|
||||
has_mark_ready = True
|
||||
if index != len(steps) - 1:
|
||||
errors.append(f"{step_where}: mark-ready 必须是 profile 最后一步")
|
||||
|
||||
if stop_at in {"review_ready", "released"} and (
|
||||
not has_pull_request or not has_mark_ready
|
||||
):
|
||||
errors.append(
|
||||
f"{where}: {stop_at} 必须包含 pull-request 和末尾 mark-ready"
|
||||
)
|
||||
if stop_at == "released" and not ({"release", "production"} & approvals):
|
||||
errors.append(f"{where}: released profile 必须包含 release 或 production approval")
|
||||
|
||||
if profile_id == default_profile:
|
||||
if stop_at != "review_ready":
|
||||
errors.append(f"{where}: defaultProfile 必须停在 review_ready")
|
||||
used_destinations = {
|
||||
step.get("destination")
|
||||
for step in steps
|
||||
if isinstance(step, dict) and step.get("action") == "publish"
|
||||
}
|
||||
used_environments = {
|
||||
step.get("environment")
|
||||
for step in steps
|
||||
if isinstance(step, dict) and step.get("action") == "deploy"
|
||||
}
|
||||
if any(
|
||||
destinations.get(item, {}).get("channel") == "stable"
|
||||
for item in used_destinations
|
||||
):
|
||||
errors.append(f"{where}: defaultProfile 不能发布 stable 目标")
|
||||
if any(
|
||||
environments.get(item, {}).get("classification") == "production"
|
||||
for item in used_environments
|
||||
):
|
||||
errors.append(f"{where}: defaultProfile 不能部署 production 环境")
|
||||
|
||||
|
||||
def validate_builtin(data: dict[str, Any], project_root: Path | None = None) -> list[str]:
|
||||
errors: list[str] = []
|
||||
_reject_unknown(data, TOP_LEVEL_FIELDS, "<root>", errors)
|
||||
|
||||
if data.get("version") != 1 or isinstance(data.get("version"), bool):
|
||||
errors.append("version 必须是整数 1")
|
||||
if "updatedAt" in data and not _nonempty(data.get("updatedAt")):
|
||||
errors.append("updatedAt 必须是非空字符串")
|
||||
project = data.get("project")
|
||||
if not _mapping(project):
|
||||
errors.append("project 必须是对象")
|
||||
project = {}
|
||||
else:
|
||||
_reject_unknown(project, {"name"}, "project", errors)
|
||||
if not _nonempty(project.get("name")):
|
||||
errors.append("project.name 必须是非空字符串")
|
||||
|
||||
enabled = data.get("enabled")
|
||||
if not isinstance(enabled, bool):
|
||||
errors.append("enabled 必须是布尔值")
|
||||
default_profile = data.get("defaultProfile")
|
||||
if default_profile is not None and (
|
||||
not isinstance(default_profile, str) or ID_RE.fullmatch(default_profile) is None
|
||||
):
|
||||
errors.append("defaultProfile 必须是 null 或小写连字符 ID")
|
||||
|
||||
entrypoints = _validate_ids(data.get("entrypoints"), "entrypoints", errors)
|
||||
artifacts = _validate_ids(data.get("artifacts"), "artifacts", errors)
|
||||
destinations = _validate_ids(data.get("destinations"), "destinations", errors)
|
||||
environments = _validate_ids(data.get("environments"), "environments", errors)
|
||||
profiles = _validate_ids(data.get("profiles"), "profiles", errors)
|
||||
|
||||
_validate_entrypoints(entrypoints, errors, project_root)
|
||||
_validate_artifacts(artifacts, entrypoints, errors)
|
||||
_validate_destinations(destinations, entrypoints, errors)
|
||||
_validate_environments(environments, entrypoints, errors)
|
||||
_validate_profiles(
|
||||
profiles,
|
||||
default_profile=default_profile,
|
||||
entrypoints=entrypoints,
|
||||
artifacts=artifacts,
|
||||
destinations=destinations,
|
||||
environments=environments,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
if enabled:
|
||||
if default_profile not in profiles:
|
||||
errors.append("enabled=true 时 defaultProfile 必须引用已定义 profile")
|
||||
elif not profiles[default_profile].get("steps"):
|
||||
errors.append("enabled=true 时 defaultProfile.steps 不能为空")
|
||||
elif default_profile is not None and default_profile not in profiles:
|
||||
errors.append("defaultProfile 必须引用已定义 profile")
|
||||
|
||||
_scan_secrets(data, "<root>", errors)
|
||||
return errors
|
||||
|
||||
|
||||
def validate_tasks_link(delivery: dict[str, Any], tasks: dict[str, Any]) -> list[str]:
|
||||
errors: list[str] = []
|
||||
project = tasks.get("project")
|
||||
if not isinstance(project, dict):
|
||||
return ["tasks.project 必须是对象"]
|
||||
if project.get("deliveryFile") != "docs/ack/delivery.yaml":
|
||||
errors.append("tasks.project.deliveryFile 必须固定为 docs/ack/delivery.yaml")
|
||||
delivery_project = delivery.get("project")
|
||||
if (
|
||||
isinstance(delivery_project, dict)
|
||||
and _nonempty(delivery_project.get("name"))
|
||||
and _nonempty(project.get("name"))
|
||||
and delivery_project["name"] != project["name"]
|
||||
):
|
||||
errors.append("delivery.project.name 必须与 tasks.project.name 一致")
|
||||
if not isinstance(tasks.get("deliveryRuns"), list):
|
||||
errors.append("引用 deliveryFile 的任务板必须包含 deliveryRuns 列表")
|
||||
return errors
|
||||
|
||||
|
||||
def validate_with_schema(data: dict[str, Any], schema_path: Path) -> list[str]:
|
||||
import jsonschema # type: ignore
|
||||
|
||||
schema = json.loads(schema_path.read_text(encoding="utf-8"))
|
||||
validator = jsonschema.Draft7Validator(schema)
|
||||
errors = []
|
||||
for error in sorted(validator.iter_errors(data), key=lambda item: list(item.path)):
|
||||
location = "/".join(str(part) for part in error.path) or "<root>"
|
||||
errors.append(f"[schema] {location}: {error.message}")
|
||||
return errors
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="校验 ACK 项目交付契约")
|
||||
parser.add_argument("delivery", nargs="?", default="docs/ack/delivery.yaml")
|
||||
parser.add_argument("--tasks", help="关联的 docs/ack/tasks.yaml")
|
||||
parser.add_argument("--project-root", help="项目根目录；提供后检查入口路径")
|
||||
parser.add_argument("--schema", help="delivery.schema.json 路径（默认自动探测）")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
delivery_path = Path(args.delivery)
|
||||
if not delivery_path.is_file():
|
||||
sys.stderr.write(f"找不到交付契约: {delivery_path}\n")
|
||||
return 2
|
||||
project_root = Path(args.project_root).resolve() if args.project_root else None
|
||||
if project_root is not None and not project_root.is_dir():
|
||||
sys.stderr.write(f"项目根目录不存在: {project_root}\n")
|
||||
return 2
|
||||
|
||||
delivery = _load_document(delivery_path, "交付契约")
|
||||
errors = validate_builtin(delivery, project_root)
|
||||
|
||||
if args.tasks:
|
||||
tasks_path = Path(args.tasks)
|
||||
if not tasks_path.is_file():
|
||||
sys.stderr.write(f"找不到任务板: {tasks_path}\n")
|
||||
return 2
|
||||
tasks = _load_document(tasks_path, "任务板")
|
||||
errors.extend(validate_tasks_link(delivery, tasks))
|
||||
|
||||
schema_path = (
|
||||
Path(args.schema)
|
||||
if args.schema
|
||||
else Path(__file__).resolve().parent.parent / "templates" / "delivery.schema.json"
|
||||
)
|
||||
if args.schema and not schema_path.is_file():
|
||||
sys.stderr.write(f"找不到 schema: {schema_path}\n")
|
||||
return 2
|
||||
if schema_path.is_file():
|
||||
try:
|
||||
errors.extend(validate_with_schema(delivery, schema_path))
|
||||
except ImportError:
|
||||
sys.stderr.write("提示: 未安装 jsonschema，仅执行内置语义规则\n")
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
sys.stderr.write(f"schema 读取失败: {exc}\n")
|
||||
return 2
|
||||
|
||||
if errors:
|
||||
for error in errors:
|
||||
sys.stderr.write(f"- {error}\n")
|
||||
return 1
|
||||
|
||||
sys.stdout.write("交付契约校验通过\n")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user