laily
12e00bd594
Keep pouch naming and .pouch/ack project state, and bring in ACK regression mode, deployer test-environment binding, and manage-release updates from main.
904 lines
36 KiB
Python
Executable File
904 lines
36 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""校验 ACK 项目交付契约。
|
||
|
||
权威结构位于 templates/delivery.schema.json。jsonschema 是可选依赖；内置规则始终
|
||
检查引用、步骤顺序、默认 profile 安全边界、敏感信息和仓库内入口路径。
|
||
|
||
用法:
|
||
python3 validate_delivery.py .pouch/ack/delivery.yaml
|
||
python3 validate_delivery.py .pouch/ack/delivery.yaml \
|
||
--tasks .pouch/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",
|
||
"intents",
|
||
"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", "validation_ready", "review_ready", "released"}
|
||
INTENT_FIELDS = {"testEnvironment", "release"}
|
||
INTENT_STOP_AT = {
|
||
"release": "released",
|
||
}
|
||
TEST_ENVIRONMENT_FIELDS = {"via", "env"}
|
||
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()
|
||
checked_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}"
|
||
)
|
||
else:
|
||
checked_environments.add(environment_id)
|
||
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 stop_at == "validation_ready":
|
||
if not deployed_environments:
|
||
errors.append(f"{where}: validation_ready 必须至少部署一个环境")
|
||
missing_health = deployed_environments - checked_environments
|
||
if missing_health:
|
||
errors.append(
|
||
f"{where}: validation_ready 的部署环境必须全部完成 health-check: "
|
||
f"{sorted(missing_health)}"
|
||
)
|
||
|
||
if profile_id == default_profile:
|
||
if stop_at not in {"validation_ready", "review_ready"}:
|
||
errors.append(
|
||
f"{where}: defaultProfile 必须停在 validation_ready 或 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_test_environment_intent(
|
||
value: Any,
|
||
errors: list[str],
|
||
project_root: Path | None,
|
||
) -> None:
|
||
if value is None:
|
||
return
|
||
if isinstance(value, str):
|
||
errors.append(
|
||
"intents.testEnvironment: 已改为 deployer 绑定 "
|
||
"{via: deployer, env: <env>}，不能再使用 profile ID "
|
||
f"{value!r}。请迁移后由 ACK 内部调用 deployer skill"
|
||
)
|
||
return
|
||
if not _mapping(value):
|
||
errors.append("intents.testEnvironment: 必须是 null 或 {via, env} 对象")
|
||
return
|
||
_reject_unknown(value, TEST_ENVIRONMENT_FIELDS, "intents.testEnvironment", errors)
|
||
if value.get("via") != "deployer":
|
||
errors.append("intents.testEnvironment.via 必须是 'deployer'")
|
||
env = value.get("env")
|
||
if not isinstance(env, str) or ID_RE.fullmatch(env) is None:
|
||
errors.append("intents.testEnvironment.env 必须是小写连字符环境名")
|
||
return
|
||
if project_root is None:
|
||
return
|
||
env_dir = project_root / ".pouch" / "deployer" / env
|
||
if not env_dir.is_dir():
|
||
env_dir = project_root / ".skiff" / "deployer" / env
|
||
if not env_dir.is_dir():
|
||
errors.append(
|
||
"intents.testEnvironment.env: 找不到 "
|
||
f".pouch/deployer/{env}；先按 deployer skill 配置项目测试环境"
|
||
)
|
||
|
||
|
||
def _validate_intents(
|
||
values: Any,
|
||
profiles: dict[str, Any],
|
||
errors: list[str],
|
||
project_root: Path | None = None,
|
||
) -> None:
|
||
if values is None:
|
||
return
|
||
if not _mapping(values):
|
||
errors.append("intents: 必须是对象")
|
||
return
|
||
_reject_unknown(values, INTENT_FIELDS, "intents", errors)
|
||
if "testEnvironment" not in values:
|
||
errors.append("intents.testEnvironment: 必填")
|
||
else:
|
||
_validate_test_environment_intent(
|
||
values.get("testEnvironment"), errors, project_root
|
||
)
|
||
if "release" not in values:
|
||
errors.append("intents.release: 必填")
|
||
return
|
||
profile_id = values["release"]
|
||
if profile_id is None:
|
||
return
|
||
if not isinstance(profile_id, str) or ID_RE.fullmatch(profile_id) is None:
|
||
errors.append("intents.release: 必须是 null 或小写连字符 profile ID")
|
||
return
|
||
profile = profiles.get(profile_id)
|
||
if profile is None:
|
||
errors.append(f"intents.release: 未定义 profile {profile_id!r}")
|
||
return
|
||
if _mapping(profile) and profile.get("stopAt") != INTENT_STOP_AT["release"]:
|
||
errors.append(
|
||
f"intents.release: profile {profile_id!r} 必须 stopAt released"
|
||
)
|
||
|
||
|
||
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,
|
||
)
|
||
_validate_intents(data.get("intents"), profiles, errors, project_root)
|
||
|
||
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") != ".pouch/ack/delivery.yaml":
|
||
errors.append("tasks.project.deliveryFile 必须固定为 .pouch/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=".pouch/ack/delivery.yaml")
|
||
parser.add_argument("--tasks", help="关联的 .pouch/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())
|