This repository has been archived on 2026-11-08. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
platform-ci/platform_ci/configuration.py
forge-recoveryandClaude Sonnet 5.5 a36f17b857
KofTwentyTwo/ci-policy: ci / fork-guard (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / go-0 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / go-nix-1 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / gradle-2 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / gradle-3 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / java-docker-4 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / java-maven-5 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / manifests-6 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / nix-7 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / node-8 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / python-9 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / rust-10 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / rust-11 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / rust-12 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / source-config-13 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / source-config-14 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-0 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-1 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-2 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-3 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-4 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-5 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-6 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / image-build-7 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / not-enrolled (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / content-validate-0 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / standards-0 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / node-matrix-0 (pull_request) Skipped
KofTwentyTwo/ci-policy: ci / source-config-15 (pull_request) Blocked by required conditions
KofTwentyTwo/ci-policy: Shared source and config CI / verify (pull_request) Blocked by required conditions
KofTwentyTwo/ci-policy: Shared source and config CI / secrets (pull_request) Blocked by required conditions
KofTwentyTwo/ci-policy: Shared source and config CI / gate (pull_request) Waiting for needed jobs
KofTwentyTwo/ci-policy: ci / ci-result (pull_request) Waiting for needed jobs
KofTwentyTwo/ci-policy: native / fork-deny (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-macos-0 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-macos-1 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-macos-2 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-macos-3 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-windows-4 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-windows-5 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-windows-6 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / build-windows-7 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / dotnet-8 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / dotnet-9 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / dotnet-10 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / dotnet-11 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-12 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-13 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-14 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-15 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-16 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-17 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-18 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / swift-19 (pull_request) Skipped
KofTwentyTwo/ci-policy: native / native-result (pull_request) Canceled after 6s
feat(ci): advance shared pipeline capabilities
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-05 10:30:25 +00:00

981 lines
61 KiB
Python

"""Load and strictly validate `.ci.yaml` from the exact checked-out source revision.
`verification.push` selects which ordinary pushes verify; it cannot disable pull request verification.
Only the fields this pipeline implements are accepted. A field the approved design describes but that is not implemented
yet (an artifact `source`) is rejected by name, so a configuration can never appear to request behavior that does not exist.
Delivery fields only REQUEST work. `release.publications` names a component artifact, a logical destination and the channels it
applies to; it carries no address, credential, owner or signer. The trusted publisher resolves the destination through its own
operator-owned bindings, so no source configuration can widen what any identity may deliver.
"""
import re
from collections.abc import Hashable, Sequence
from dataclasses import dataclass, field, replace
from pathlib import Path, PurePosixPath
from typing import cast, override
import yaml
from platform_ci.build_tools import BUILD_TOOLS, IMAGE_PROFILE, IMAGE_RECIPES, PLATFORMS, image_archive_path, image_evidence_paths
from platform_ci.effective_version import DEFAULT_VERSION_PROPERTY, EFFECTIVE_VERSION_RULES, RC_BASELINE_NONE, RC_BASELINES, VERSION_PROPERTY_PATTERN
from platform_ci.source_tools import validate_helper_command
from platform_ci.workspace import PipelineError
CONFIGURATION_FILE_NAME = ".ci.yaml"
SUPPORTED_SCHEMA_VERSION = 1
SUPPORTED_PLATFORMS = tuple(PLATFORMS)
TOOLCHAIN_NAME_BY_BUILD_TOOL = {name: tool.toolchain for name, tool in BUILD_TOOLS.items()}
REPORT_FORMATS = ("junit", "file")
MAVEN_ARTIFACT_FORMAT = "maven"
NPM_ARTIFACT_FORMAT = "npm"
CONTAINER_ARTIFACT_FORMAT = "container"
ARTIFACT_FORMATS = (MAVEN_ARTIFACT_FORMAT, NPM_ARTIFACT_FORMAT, CONTAINER_ARTIFACT_FORMAT)
BUILD_TOOL_BY_PACKAGE_FORMAT = {MAVEN_ARTIFACT_FORMAT: ("maven", "script"), NPM_ARTIFACT_FORMAT: ("node",)}
"""The build tools that may declare an artifact of a package format. A `script` component (a checked-in producer program, such as a native Android build) may declare Maven-format
artifacts; the format is still validated here and the trusted publisher still reads and verifies the artifact itself, so nothing about publication is taken from the script."""
VERSIONING_SNAPSHOT = "maven-snapshot"
VERSIONING_RELEASE = "maven-release"
VERSIONING_DECLARED = "declared"
VERSIONING_RULES = (VERSIONING_SNAPSHOT, VERSIONING_RELEASE, VERSIONING_DECLARED)
TAG_POLICY_STABLE = "stable"
TAG_POLICY_V_PREFIX = "v-prefix"
TAG_POLICY_NONE = "none"
TAG_POLICIES = (TAG_POLICY_STABLE, TAG_POLICY_V_PREFIX, TAG_POLICY_NONE)
_IDENTIFIER_PATTERN = re.compile(r"[a-z][a-z0-9-]{0,39}")
_VERSION_PATTERN = re.compile(r"[0-9]+(\.[0-9]+){0,2}")
_BRANCH_PATTERN = re.compile(r"[A-Za-z0-9*][A-Za-z0-9._/*-]{0,199}")
_STAR_RUN_PATTERN = re.compile(r"\*{3,}")
_PROGRAM_PATTERN = re.compile(r"\./[A-Za-z0-9_][A-Za-z0-9_./-]{0,199}")
_TAG_PATTERN = re.compile(r"[A-Za-z0-9*?][A-Za-z0-9._/*?-]{0,199}")
_DESTINATION_PATTERN = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}")
_VERSION_FILE_PATTERN = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_./-]{0,198}\.properties")
_VERSION_PROPERTY_PATTERN = re.compile(r"[A-Za-z][A-Za-z0-9._-]{0,99}")
_PORT_PATTERN = re.compile(r"[1-9][0-9]{3,4}")
_TARGET_PATTERN = re.compile(r"[A-Za-z0-9._-]*")
MAX_PORT = 65535
PULL_REQUEST_SELECTION = "pull_request"
"""The reserved event-selection name that matches every pull request (kept equal to `event_selection.PULL_REQUEST`)."""
_COMMIT_PATTERN = re.compile(r"[0-9a-f]{40}")
_REPOSITORY_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*")
_BINDING_PATTERN = re.compile(r"[a-z][a-z0-9-]{0,39}")
_CHECKOUT_PATH_PATTERN = re.compile(r"[A-Za-z0-9_][A-Za-z0-9._-]*(/[A-Za-z0-9_][A-Za-z0-9._-]*)*")
class ConfigurationError(PipelineError):
"""Raised when `.ci.yaml` is missing, malformed, or requests something this pipeline does not implement."""
@dataclass(frozen=True)
class Report:
"""One declared report: how it is interpreted, which globs find it, and whether a successful build must produce it."""
format: str
paths: tuple[str, ...]
required: bool = True
@dataclass(frozen=True)
class ImageSpec:
"""One OCI image a component packages from its own verified output, in the same job, without building the source again.
`artifact` names a declared artifact of the SAME component: its files, as recorded in the verification manifest, are the only
build outputs the image may contain. The image is built from either the repository `dockerfile` or a central `recipe`; both are
paths or names, never an address. `context` is the build context inside the component directory.
"""
name: str
artifact: str
dockerfile: str = ""
recipe: str = ""
context: str = "."
target: str = ""
node_version: str = ""
health_port: str = ""
only_on_channels: tuple[str, ...] = ()
not_on_channels: tuple[str, ...] = ()
"""Event selection, as for a component (release channel names or `pull_request`): the image is packaged only for events it selects. Both empty means every
verified event, which is not what a repository that built images on one branch wants: declare the selection rather than dropping the image."""
ignore_unfixed: bool = False
"""The one closed scan-policy field: HIGH and CRITICAL findings still fail, except findings with no available fix (`trivy --ignore-unfixed`), as a repository's
own earlier scan policy may have said. There is no way to turn the scan off, and every other finding remains a failure."""
@dataclass(frozen=True)
class Checkout:
"""One pinned private dependency the platform fetches before the build, with an operator-authorized read binding.
`binding` is a LOGICAL name in the platform's authorization manifest (`runtime/checkout-bindings.yaml`), which owns which repositories may read
which; `repository`, `ref` (a full commit id) and `path` (inside the checkout) are this consumer's own dependency identity. No credential value or
address appears here.
"""
name: str
binding: str
repository: str
ref: str
path: str
@dataclass(frozen=True)
class VersionSource:
"""The explicit native version source of a `script` component: a `.properties` file inside the component directory and the key whose literal value is the version.
The same declaration is read from the checked-out files when the bundle is packaged and, independently, from the forge at the build and release commits; it is part of the
target, so of the build-input identity. There is no expression language: only a literal `key=value`, as `platform_ci.versions.parse_properties_version` defines.
"""
file: str
property: str
@dataclass(frozen=True)
class Component:
"""One independently built part of a repository, already validated.
`build_commands` is the single canonical command sequence. Each command is the argument list handed to the
build tool's own executable; the legacy `build.arguments` form is stored as one command.
"""
name: str
directory: str
platforms: tuple[str, ...]
build_tool: str
toolchain_version: str
build_commands: tuple[tuple[str, ...], ...]
reports: dict[str, Report]
artifacts: dict[str, tuple[str, ...]]
requires: tuple[str, ...] = ()
"""Optional runtime profiles (`browser`, `docker`) the component's tests need; empty selects the standard runtime."""
toolchain: str = ""
"""The declared toolchain name; empty means the build tool's own."""
only_on_channels: tuple[str, ...] = ()
"""Run this component only for events whose branch (the push branch, or a pull request's head branch) matches one of these release channels."""
not_on_channels: tuple[str, ...] = ()
"""Skip this component for events whose branch matches any of these release channels. Empty on both means the component always runs."""
event_facts: bool = False
"""Opt in (`script` tool only) to the fixed variables `CI_SOURCE_REVISION`, `CI_EVENT_NAME` and `CI_REF_NAME`, taken from the authoritative event, not from the
runner's environment. They are part of the target and so of the build-input identity: a component that reads them is not reused across events or commits."""
checkouts: tuple[Checkout, ...] = ()
"""Pinned private dependencies fetched before the build (see `Checkout`)."""
artifact_formats: dict[str, str] = field(default_factory=dict[str, str])
"""The delivery format (`maven`, `npm`, `container`) of each artifact that declares one; only those can be published."""
images: dict[str, ImageSpec] = field(default_factory=dict[str, ImageSpec])
"""Images this component packages from its verified output; a component that declares any requires the `image` runtime profile."""
version_source: VersionSource | None = None
"""The explicit version source (`script` tool only); None keeps the build tool's own (pom.xml, package.json) or no version."""
WRITE_BACK_FILE = "pom.xml"
"""The captured source-version helper rewrites only the root POM's `<properties><revision>`; no other file or property is supported."""
@dataclass(frozen=True)
class WriteBackPolicy:
"""A channel's repository-owned declaration that the verified, published version is written back to the source (`release.channels.<name>.version_write_back`). UNTESTED.
`file` is the file whose `<revision>` is rewritten (only `pom.xml`, as in the captured helper); `target_branch` is the branch the bump is pushed to for a tag-driven channel (a branch-driven
channel always pushes to the branch it released from, so it must not set one); `require_push` makes an unacknowledged push an error, as the captured RC-completion path required. Presence
is the opt-in; nothing here names a repository, a credential or a trigger.
"""
file: str = WRITE_BACK_FILE
target_branch: str = ""
require_push: bool = False
@dataclass(frozen=True)
class ReleaseChannel:
"""A named release channel, the branch names or globs whose pushes make it eligible, and its optional versioning rule.
`versioning` is `maven-snapshot` (the declared version must end in -SNAPSHOT), `maven-release` (it must not) or `declared`
(any literal declared version); empty states no rule. The trusted publisher enforces it against the independently read version.
"""
name: str
push_branches: tuple[str, ...]
versioning: str = ""
tag_policy: str = ""
"""The captured `release_tag_policy` vocabulary, owned by the repository: `stable` (a final `X.Y.Z` package needs the existing `v<version>` tag at the
release commit), `v-prefix` (any non-snapshot package does) or `none`/empty (no tag is required). No branch name decides it."""
push_tags: tuple[str, ...] = ()
"""Tag names or `*`/`?` patterns whose pushes also make this channel match (a tag-driven release kind); selection and verification see them, but the trusted
publisher still releases from protected branch heads only, so a tag-only channel selects work and publishes nothing by itself."""
central_completion: bool = False
"""Maven publications on this channel are complete only when the trusted Central completion receipt of the exact release is terminal (a snapshot UPLOADED, a
release PUBLISHED with a deployment id and the exact coordinates); upload acceptance or a pending state never completes it."""
release_record: bool = False
"""Finalize the forge release record for the existing tag `v<version>` once the publication (and, if required, Central completion) is complete. Needs a
`tag_policy`, because the tag must exist at the verified commit first."""
tag_create: bool = False
"""Opt in to the platform's trusted write phase creating that required tag as an SSH-signed annotated tag, once, after the run's verification
succeeded (`release-tag`). Default false: the owners' own release step makes the tag and the publisher only requires it. Needs a `tag_policy`."""
effective_version: str = ""
"""UNTESTED. One rule of `platform_ci.effective_version` (`develop-next-minor`, `rc-from-branch`, `patch-bump`, `stable-from-tag`, `feature-snapshot`) calculating the release version BEFORE the
single native build and applying it as `-D<version_property>=<version>` to the commands of every Maven component the event selects, as the captured calculator rewrote the one root POM.
Empty means the declared version is the version. No other constraint: a rule is checked against the event only when a push matches the channel."""
version_property: str = ""
"""The Maven property the effective version is applied to (default `revision` when a rule is set)."""
rc_baseline: str = ""
"""`reconciled` states the legacy qbit RC-counter restoration is part of this repository's release lifecycle; `rc-from-branch` then refuses the one case it cannot decide (see `effective_version`)."""
version_write_back: WriteBackPolicy | None = None
"""UNTESTED. Opt in to writing the published effective version back to the source through the isolated `finalize-version` command; absent means nothing is ever written to the source."""
manual: bool = False
"""UNTESTED. `trigger.manual: true`: a manual run (`workflow_dispatch`) also selects this channel, for `only_on_channels` / `not_on_channels` only. A manual run is never a push, so it is never
release-eligible: nothing on this channel is published, tagged, version-calculated or written back by it. A channel with no push trigger is manual-only, which no publication may name."""
@dataclass(frozen=True)
class Publication:
"""A request to deliver one component artifact to one logical destination on the named channels. It holds no secret or address.
`deploy` optionally names a logical GitOps destination whose enrolled image entry is pinned to the published container digest.
"""
name: str
component: str
artifact: str
format: str
platform: str
destination: str
channels: tuple[str, ...]
deploy: str = ""
@dataclass(frozen=True)
class PushSelection:
"""Which pushes ask for verification besides the ones that are always covered.
Pull requests, manual and callable requests, default branch pushes and release-eligible pushes always verify and
cannot be configured away. Branches are exact names or globs where `*` matches within one path segment and `**`
matches across segments (`release/**`); tags are patterns where `*` and `?` match any characters.
"""
branches: tuple[str, ...] = ()
tags: tuple[str, ...] = ("*",)
@dataclass(frozen=True)
class ProjectConfiguration:
"""The validated contents of `.ci.yaml`."""
components: tuple[Component, ...]
channels: tuple[ReleaseChannel, ...]
push: PushSelection = PushSelection()
publications: tuple[Publication, ...] = ()
def load_project_configuration(source_directory: Path) -> ProjectConfiguration:
"""Read `.ci.yaml` from the source checkout and validate it; a missing file is an error, never an empty pass."""
path = source_directory / CONFIGURATION_FILE_NAME
if not path.is_file():
raise ConfigurationError(f"{CONFIGURATION_FILE_NAME} not found in the checked-out revision; a pipeline never passes without configuration")
try:
text = path.read_text(encoding="utf-8")
except UnicodeDecodeError as error:
raise ConfigurationError(f"{CONFIGURATION_FILE_NAME} is not valid UTF-8: {error}") from error
return parse_project_configuration_text(text)
def parse_project_configuration_text(text: str) -> ProjectConfiguration:
"""Validate `.ci.yaml` text, wherever it was read from (a checkout or the forge's contents API)."""
try:
raw = parse_yaml_rejecting_duplicate_keys(text)
except yaml.YAMLError as error:
raise ConfigurationError(f"{CONFIGURATION_FILE_NAME} is not valid YAML: {error}") from error
return parse_project_configuration(raw)
class _UniqueKeyLoader(yaml.SafeLoader):
"""A safe YAML loader that refuses a mapping that repeats a key instead of keeping the last value."""
@override
def construct_mapping(self, node: yaml.MappingNode, deep: bool = False) -> dict[Hashable, object]:
"""Build the mapping after proving that no key, including merged keys, appears twice."""
self.flatten_mapping(node)
seen: set[Hashable] = set()
for key_node, _value_node in node.value:
# The PyYAML stub leaves `construct_object`'s `node` parameter unannotated, so the bound method is partially unknown.
key: object = self.construct_object(key_node, deep=True) # pyright: ignore[reportUnknownMemberType]
if not isinstance(key, Hashable):
continue # the parent class reports an unhashable key with its own message
if key in seen:
raise yaml.constructor.ConstructorError(None, None, f"duplicate key {key!r}", key_node.start_mark)
seen.add(key)
return super().construct_mapping(node, deep=deep)
def parse_yaml_rejecting_duplicate_keys(text: str) -> object:
"""Parse YAML safely; a repeated key at any depth raises `yaml.YAMLError` naming the key and its line."""
return cast("object", yaml.load(text, Loader=_UniqueKeyLoader)) # noqa: S506 - the loader derives from SafeLoader
def parse_project_configuration(raw: object) -> ProjectConfiguration:
"""Validate already-parsed YAML and return the typed configuration, or raise with the offending field path."""
root = _mapping(raw, "configuration", ("schema", "components", "release", "verification", "scoped_delegation"))
schema = _required(root, "schema", "configuration")
if type(schema) is not int or schema != SUPPORTED_SCHEMA_VERSION:
raise ConfigurationError(f"schema: only schema {SUPPORTED_SCHEMA_VERSION} is supported, got {schema!r}")
# Read only by the org dispatcher's opt-in step (the default branch's value delegates the repository to this pipeline); validated here so it cannot be a typo, never used by a plan.
delegation = root.get("scoped_delegation", False)
if not isinstance(delegation, bool):
raise ConfigurationError(f"scoped_delegation: expected true or false, got {delegation!r}")
component_entries = _named_entries(_required(root, "components", "configuration"), "components")
if not component_entries:
raise ConfigurationError("components: at least one component is required")
components = tuple(_parse_component(name, entry) for name, entry in sorted(component_entries.items()))
release = _mapping(root.get("release", {}), "release", ("channels", "publications"))
channels = _parse_release_channels(release.get("channels", {}))
_require_valid_selection(components, channels)
publications = _parse_publications(release.get("publications", {}), components, channels)
_require_publications_selectable(components, publications)
_require_tag_format_for_publications(channels, publications)
_require_completion_for_packages(channels, publications)
_require_write_back_publication(channels, publications)
return ProjectConfiguration(components=components, channels=channels, push=_parse_push_selection(root.get("verification", {})), publications=publications)
def _require_valid_selection(components: tuple[Component, ...], channels: tuple[ReleaseChannel, ...]) -> None:
"""Require every event-selection name to be a declared release channel or `pull_request`."""
known = {channel.name for channel in channels} | {PULL_REQUEST_SELECTION}
for component in components:
selections = [(f"components.{component.name}", component.only_on_channels, component.not_on_channels)]
selections += [(f"components.{component.name}.images.{image.name}", image.only_on_channels, image.not_on_channels) for image in component.images.values()]
for location, only_on, not_on in selections:
for field_name, names in (("only_on_channels", only_on), ("not_on_channels", not_on)):
unknown = sorted(set(names) - known)
if unknown:
raise ConfigurationError(f"{location}.{field_name}: no release channel named {', '.join(unknown)} (use a declared channel or {PULL_REQUEST_SELECTION})")
def _selected_on_channel(only_on: tuple[str, ...], not_on: tuple[str, ...], channel: str) -> bool:
"""Return whether something restricted this way is selected on an event matching exactly this one release channel."""
return (not only_on or channel in only_on) and channel not in not_on
def _require_completion_for_packages(channels: tuple[ReleaseChannel, ...], publications: tuple[Publication, ...]) -> None:
"""Central completion is for Maven packages only, and a release record for a package (never an image, which has no version and no tag)."""
by_name = {channel.name: channel for channel in channels}
for publication in publications:
for channel_name in publication.channels:
channel = by_name[channel_name]
if channel.central_completion and publication.format != MAVEN_ARTIFACT_FORMAT:
raise ConfigurationError(f"release.publications.{publication.name}: channel {channel_name!r} requires Central completion, which applies to Maven publications only")
if channel.release_record and publication.format not in (MAVEN_ARTIFACT_FORMAT, NPM_ARTIFACT_FORMAT):
raise ConfigurationError(f"release.publications.{publication.name}: channel {channel_name!r} finalizes a release record, which applies to Maven and npm publications only")
def _require_tag_format_for_publications(channels: tuple[ReleaseChannel, ...], publications: tuple[Publication, ...]) -> None:
"""Require a tag format only where a tag can release: a tag-driven channel that a publication uses must say which tag format it accepts.
A tag channel used only to select work (a `publish*` kind with no publication) needs no format, so it is not forced to carry a policy that does not fit it.
"""
by_name = {channel.name: channel for channel in channels}
for publication in publications:
for channel_name in publication.channels:
channel = by_name[channel_name]
if channel.push_tags and channel.tag_policy not in (TAG_POLICY_STABLE, TAG_POLICY_V_PREFIX):
raise ConfigurationError(f"release.publications.{publication.name}: channel {channel_name!r} is tag-driven, so it needs a tag_policy of {TAG_POLICY_STABLE} or {TAG_POLICY_V_PREFIX}, the tag format it accepts")
def _require_publications_selectable(components: tuple[Component, ...], publications: tuple[Publication, ...]) -> None:
"""Refuse a publication whose component, or whose image, an event of its own channel would exclude, instead of failing after the build.
The check assumes the event matches that one channel; the trusted publisher applies the full event rule again and refuses a mismatch.
"""
by_name = {component.name: component for component in components}
for publication in publications:
component = by_name[publication.component]
image = next((spec for spec in component.images.values() if publication.artifact == f"image-{spec.name}"), None)
for channel in publication.channels:
if not _selected_on_channel(component.only_on_channels, component.not_on_channels, channel):
raise ConfigurationError(f"release.publications.{publication.name}: component {component.name!r} is not selected on channel {channel!r}, so no bundle would exist to publish")
if image is not None and not _selected_on_channel(image.only_on_channels, image.not_on_channels, channel):
raise ConfigurationError(f"release.publications.{publication.name}: image {image.name!r} is not packaged on channel {channel!r}, so its archive would not exist to publish")
def branch_matches(patterns: Sequence[str], branch: str) -> bool:
"""Return whether a branch name equals a pattern or matches a glob: `*` stays inside a path segment, `**` crosses segments."""
return any(_branch_regex(pattern).fullmatch(branch) is not None for pattern in patterns)
def _branch_regex(pattern: str) -> re.Pattern[str]:
"""Translate one validated branch pattern; every other character, including `.`, matches only itself."""
parts = re.split(r"(\*\*|\*)", pattern)
return re.compile("".join(".*" if part == "**" else "[^/]*" if part == "*" else re.escape(part) for part in parts))
def _require_branch_pattern(branch: str, location: str) -> None:
"""Fail unless the text is a branch name or a glob using only `*` and `**`."""
if _BRANCH_PATTERN.fullmatch(branch) is None or _STAR_RUN_PATTERN.search(branch) is not None:
raise ConfigurationError(f"{location}: expected a branch name or a glob using only * and **, got {branch!r}")
def _parse_push_selection(value: object) -> PushSelection:
"""Validate `verification.push`; omitted fields take the safe defaults (no extra branches, every tag)."""
verification = _mapping(value, "verification", ("push",))
push = _mapping(verification.get("push", {}), "verification.push", ("branches", "tags"))
defaults = PushSelection()
branches = defaults.branches
if "branches" in push:
branches = _text_list(push["branches"], "verification.push.branches", allow_empty=True)
for branch in branches:
_require_branch_pattern(branch, "verification.push.branches")
tags = defaults.tags
if "tags" in push:
tags = _text_list(push["tags"], "verification.push.tags", allow_empty=True)
for tag in tags:
if _TAG_PATTERN.fullmatch(tag) is None:
raise ConfigurationError(f"verification.push.tags: expected a tag name or a pattern using only * and ?, got {tag!r}")
return PushSelection(branches=tuple(sorted(set(branches))), tags=tuple(sorted(set(tags))))
def _parse_artifacts(value: object, tool: str, location: str) -> tuple[dict[str, tuple[str, ...]], dict[str, str]]:
"""Validate a component's artifacts into (paths by name, declared format by name)."""
artifacts: dict[str, tuple[str, ...]] = {}
artifact_formats: dict[str, str] = {}
for artifact_name, entry in _named_entries(value, f"{location}.artifacts").items():
artifact_location = f"{location}.artifacts.{artifact_name}"
artifact = _mapping(entry, artifact_location, ("paths", "format"), ("source",))
paths = _text_list(_required(artifact, "paths", artifact_location), f"{artifact_location}.paths")
artifacts[artifact_name] = tuple(_relative_path(path, f"{artifact_location}.paths") for path in paths)
if "format" in artifact:
artifact_formats[artifact_name] = _parse_artifact_format(artifact["format"], tool, f"{artifact_location}.format")
return artifacts, artifact_formats
def _parse_component(name: str, value: object) -> Component:
"""Validate one component; the toolchain key allowed depends on the build tool it selects."""
location = f"components.{name}"
fields = _mapping(value, location, ("directory", "platforms", "toolchains", "build", "reports", "artifacts", "requires", "images", "only_on_channels", "not_on_channels", "event_facts", "checkouts", "version"))
only_on, not_on = _parse_selection(fields, location)
event_facts = fields.get("event_facts", False)
if not isinstance(event_facts, bool):
raise ConfigurationError(f"{location}.event_facts: expected true or false, got {event_facts!r}")
build = _mapping(_required(fields, "build", location), f"{location}.build", ("tool", "arguments", "commands"))
tool = _text(_required(build, "tool", f"{location}.build"), f"{location}.build.tool")
if tool not in BUILD_TOOLS:
raise ConfigurationError(f"{location}.build.tool: unsupported build tool {tool!r}; supported: {', '.join(BUILD_TOOLS)}")
if event_facts and not BUILD_TOOLS[tool].program_in_arguments:
raise ConfigurationError(f"{location}.event_facts: only the script tool reads the event variables, not {tool!r}")
version_source = _parse_version_source(fields.get("version"), BUILD_TOOLS[tool].program_in_arguments, tool, f"{location}.version")
checkouts = _parse_checkouts(fields.get("checkouts", {}), f"{location}.checkouts")
declared_platforms = fields.get("platforms", [])
if checkouts and isinstance(declared_platforms, list) and PLATFORMS_WITHOUT_CHECKOUT.intersection(cast("list[object]", declared_platforms)):
raise ConfigurationError(f"{location}.checkouts: pinned private checkouts run in container jobs only, not on a host platform")
commands = _parse_build_commands(build, f"{location}.build")
if BUILD_TOOLS[tool].subcommands:
_require_subcommands(tool, commands, fields.get("directory", "."), location)
if BUILD_TOOLS[tool].helper_subcommands:
_require_helper_subcommands(tool, commands, fields.get("directory", "."), location)
if BUILD_TOOLS[tool].program_in_arguments:
for index, command in enumerate(commands):
if _PROGRAM_PATTERN.fullmatch(command[0]) is None or ".." in PurePosixPath(command[0]).parts:
raise ConfigurationError(f"{location}.build.commands[{index}][0]: expected a checked-in program such as ./scripts/check.sh inside the component directory, got {command[0]!r}")
toolchain_name, version = _parse_toolchain(tool, fields, location)
supported_platforms = BUILD_TOOLS[tool].platforms
platforms = _text_list(fields.get("platforms", [supported_platforms[0]]), f"{location}.platforms")
for platform in platforms:
if platform not in supported_platforms:
raise ConfigurationError(f"{location}.platforms: unsupported platform {platform!r} for build tool {tool!r}; supported: {', '.join(supported_platforms)}")
if len(set(platforms)) != len(platforms):
raise ConfigurationError(f"{location}.platforms: duplicate platform")
requires = _parse_requires(fields.get("requires", []), tool, f"{location}.requires")
reports = {report_name: _parse_report(entry, f"{location}.reports.{report_name}") for report_name, entry in _named_entries(fields.get("reports", {}), f"{location}.reports").items()}
artifacts, artifact_formats = _parse_artifacts(fields.get("artifacts", {}), tool, location)
images, requires = _parse_images(fields.get("images", {}), set(artifacts), requires, f"{location}.images")
_require_no_profiles_on_hosts(platforms, requires, location)
component = Component(
name=name,
directory=_relative_path(fields.get("directory", "."), f"{location}.directory"),
platforms=tuple(sorted(platforms)),
build_tool=tool,
toolchain_version=version,
toolchain=toolchain_name,
only_on_channels=only_on,
not_on_channels=not_on,
event_facts=event_facts,
checkouts=checkouts,
version_source=version_source,
build_commands=commands,
reports=reports,
artifacts=artifacts,
requires=requires,
artifact_formats=artifact_formats,
images=images,
)
return _with_image_outputs(component)
def _parse_version_source(value: object, is_script: bool, tool: str, location: str) -> VersionSource | None:
"""Validate the optional explicit version source: a literal `.properties` path inside the component directory and a plain key; script tool only."""
if value is None:
return None
if not is_script:
raise ConfigurationError(f"{location}: an explicit version source is for the script tool; {tool!r} reads its own version file")
fields = _mapping(value, location, ("file", "property"))
path = _relative_path(_required(fields, "file", location), f"{location}.file")
if _VERSION_FILE_PATTERN.fullmatch(path) is None or ".." in PurePosixPath(path).parts or "." in PurePosixPath(path).parts:
raise ConfigurationError(f"{location}.file: expected a literal relative path to a .properties file inside the component directory, got {path!r}")
key = _text(_required(fields, "property", location), f"{location}.property")
if _VERSION_PROPERTY_PATTERN.fullmatch(key) is None:
raise ConfigurationError(f"{location}.property: expected a plain property name such as VERSION_NAME, got {key!r}")
return VersionSource(file=path, property=key)
PLATFORMS_WITHOUT_CHECKOUT = frozenset(name for name, spec in PLATFORMS.items() if spec.execution != "container")
"""Host platforms (Windows, macOS): their jobs have no catalog image and no checkout step, so a pinned private checkout is not offered there."""
def _require_no_profiles_on_hosts(platforms: Sequence[str], requires: tuple[str, ...], location: str) -> None:
"""Runtime profiles (browser, docker, image) are features of the container pools; a host job cannot implement them, and the runtime catalog would otherwise ignore them silently."""
hosts = PLATFORMS_WITHOUT_CHECKOUT.intersection(platforms)
if requires and hosts:
raise ConfigurationError(f"{location}.requires: the runtime profiles {', '.join(requires)} belong to container platforms, not to {', '.join(sorted(hosts))}")
def _parse_selection(fields: dict[str, object], location: str) -> tuple[tuple[str, ...], tuple[str, ...]]:
"""Parse `only_on_channels` and `not_on_channels` (names of release channels, or `pull_request`); a name cannot be in both."""
only_on = tuple(sorted(set(_text_list(fields["only_on_channels"], f"{location}.only_on_channels")))) if "only_on_channels" in fields else ()
not_on = tuple(sorted(set(_text_list(fields["not_on_channels"], f"{location}.not_on_channels")))) if "not_on_channels" in fields else ()
if set(only_on) & set(not_on):
raise ConfigurationError(f"{location}: a name cannot be in both only_on_channels and not_on_channels")
return only_on, not_on
def _parse_checkouts(value: object, location: str) -> tuple[Checkout, ...]:
"""Validate the pinned private dependencies: a logical binding, a repository, a full commit id and a safe relative path; nothing secret."""
checkouts: list[Checkout] = []
paths: list[str] = []
for name, entry in sorted(_named_entries(value, location).items()):
where = f"{location}.{name}"
fields = _mapping(entry, where, ("binding", "repository", "ref", "path"))
binding = _text(_required(fields, "binding", where), f"{where}.binding")
repository = _text(_required(fields, "repository", where), f"{where}.repository")
ref = _text(_required(fields, "ref", where), f"{where}.ref")
path = _text(_required(fields, "path", where), f"{where}.path")
if _BINDING_PATTERN.fullmatch(binding) is None:
raise ConfigurationError(f"{where}.binding: expected a logical binding name such as 'content-read', got {binding!r}")
if _REPOSITORY_PATTERN.fullmatch(repository) is None:
raise ConfigurationError(f"{where}.repository: expected owner/name, got {repository!r}")
if _COMMIT_PATTERN.fullmatch(ref) is None:
raise ConfigurationError(f"{where}.ref: expected a full 40 character commit id (a branch or tag can move), got {ref!r}")
if _CHECKOUT_PATH_PATTERN.fullmatch(path) is None or path.split("/")[0] in (".git", ".platform-ci") or ".." in path.split("/"):
raise ConfigurationError(f"{where}.path: expected a plain relative directory inside the checkout (not .git or .platform-ci), got {path!r}")
if any(path == other or path.startswith(other + "/") or other.startswith(path + "/") for other in paths):
raise ConfigurationError(f"{where}.path: overlaps another checkout path")
paths.append(path)
checkouts.append(Checkout(name=name, binding=binding, repository=repository, ref=ref, path=path))
return tuple(checkouts)
def _parse_toolchain(tool: str, fields: dict[str, object], location: str) -> tuple[str, str]:
"""Return the declared toolchain name and version: the tool's own, or exactly one of its `toolchain_choices`."""
choices = BUILD_TOOLS[tool].toolchain_choices or (BUILD_TOOLS[tool].toolchain,)
toolchains = _mapping(_required(fields, "toolchains", location), f"{location}.toolchains", choices)
if len(toolchains) != 1:
raise ConfigurationError(f"{location}.toolchains: declare exactly one toolchain of {', '.join(choices)}, got {', '.join(toolchains) or 'none'}")
toolchain_name = next(iter(toolchains))
version_location = f"{location}.toolchains.{toolchain_name}"
version = _text(toolchains[toolchain_name], version_location)
if _VERSION_PATTERN.fullmatch(version) is None:
raise ConfigurationError(f'{version_location}: expected a version such as "21" or "22.20.0", got {version!r}')
return toolchain_name, version
def _parse_images(value: object, artifacts: set[str], requires: tuple[str, ...], location: str) -> tuple[dict[str, ImageSpec], tuple[str, ...]]:
"""Validate the images a component packages from its verified output and return them with the component's runtime profiles.
A recipe and a Dockerfile are mutually exclusive. The same job builds and packages, so one runtime must hold the native tool, BuildKit and
the scanner (the `image` profile, which declaring images implies) and also every browser or docker capability the component's own
verification requires.
"""
images: dict[str, ImageSpec] = {}
for name, entry in _named_entries(value, location).items():
image_location = f"{location}.{name}"
fields = _mapping(entry, image_location, ("artifact", "dockerfile", "recipe", "context", "target", "node_version", "health_port", "only_on_channels", "not_on_channels", "ignore_unfixed"))
only_on, not_on = _parse_selection(fields, image_location)
ignore_unfixed = fields.get("ignore_unfixed", False)
if not isinstance(ignore_unfixed, bool):
raise ConfigurationError(f"{image_location}.ignore_unfixed: expected true or false, got {ignore_unfixed!r}")
artifact = _text(_required(fields, "artifact", image_location), f"{image_location}.artifact")
if artifact not in artifacts:
raise ConfigurationError(f"{image_location}.artifact: {artifact!r} is not an artifact of this component; declared: {', '.join(sorted(artifacts)) or 'none'}")
dockerfile = _relative_path(fields["dockerfile"], f"{image_location}.dockerfile") if "dockerfile" in fields else ""
recipe = _text(fields["recipe"], f"{image_location}.recipe") if "recipe" in fields else ""
if bool(dockerfile) == bool(recipe):
raise ConfigurationError(f"{image_location}: declare exactly one of dockerfile and recipe")
node_version = _text(fields["node_version"], f"{image_location}.node_version") if "node_version" in fields else ""
health_port = _text(fields["health_port"], f"{image_location}.health_port") if "health_port" in fields else ""
if recipe and recipe not in IMAGE_RECIPES:
raise ConfigurationError(f"{image_location}.recipe: unknown recipe {recipe!r}; supported: {', '.join(IMAGE_RECIPES)}")
if recipe and (node_version not in ("20", "22", "24") or _PORT_PATTERN.fullmatch(health_port) is None or int(health_port) > MAX_PORT):
raise ConfigurationError(f"{image_location}: a recipe needs node_version 20, 22 or 24 and a health_port from 1000 to 65535")
if not recipe and (node_version or health_port):
raise ConfigurationError(f"{image_location}: node_version and health_port belong to a recipe")
target = _text(fields["target"], f"{image_location}.target") if "target" in fields else ""
if _TARGET_PATTERN.fullmatch(target) is None:
raise ConfigurationError(f"{image_location}.target: expected a stage name, got {target!r}")
images[name] = ImageSpec(
name=name,
artifact=artifact,
dockerfile=dockerfile,
recipe=recipe,
context=_relative_path(fields.get("context", "."), f"{image_location}.context"),
target=target,
node_version=node_version,
health_port=health_port,
only_on_channels=only_on,
not_on_channels=not_on,
ignore_unfixed=ignore_unfixed,
)
return images, _with_image_profile(requires, bool(images), f"{location}.requires")
def _with_image_profile(requires: tuple[str, ...], has_images: bool, location: str) -> tuple[str, ...]:
"""Return the component's runtime profiles with `image` added exactly when it declares images; browser and docker are kept, never dropped.
`image` is implied by `images` and cannot be declared on its own. It composes with the other profiles: the plan then needs one runtime that
has every required capability, which the catalog either lists (`image` and `docker+image` on the same image) or marks null (`browser+image`).
"""
if not has_images:
if IMAGE_PROFILE in requires:
raise ConfigurationError(f"{location}: the {IMAGE_PROFILE!r} profile is implied by `images`; declare images instead of requiring it")
return requires
return tuple(sorted({*requires, IMAGE_PROFILE}))
def _with_image_outputs(component: Component) -> Component:
"""Declare each image's archive and evidence as ordinary artifacts and reports, so the one verified bundle and manifest retain them.
The archive is an artifact `image-<name>` of format `container` (the shape the existing OCI publisher primitive reads: exactly one archive file) and
the scan, SBOM and provenance files are a required `file` report `image-<name>-evidence`. A name already used by the component is refused.
"""
if not component.images:
return component
artifacts, formats, reports = dict(component.artifacts), dict(component.artifact_formats), dict(component.reports)
for name, image in component.images.items():
archive, evidence = f"image-{name}", f"image-{name}-evidence"
if archive in artifacts or evidence in reports:
raise ConfigurationError(f"components.{component.name}.images.{name}: the generated artifact {archive!r} or report {evidence!r} would replace a declared one")
artifacts[archive] = (image_archive_path(name),)
formats[archive] = CONTAINER_ARTIFACT_FORMAT
reports[evidence] = Report(format="file", paths=image_evidence_paths(name, recipe=bool(image.recipe)), required=True)
return replace(component, artifacts=artifacts, artifact_formats=formats, reports=reports)
def _require_subcommands(tool: str, commands: tuple[tuple[str, ...], ...], directory: object, location: str) -> None:
"""Accept only the tool's closed subcommand vocabulary: each command is exactly one known name, run in the repository root."""
known = [name for name, _program in BUILD_TOOLS[tool].subcommands]
for index, command in enumerate(commands):
if len(command) != 1 or command[0] not in known:
raise ConfigurationError(f"{location}.build.commands[{index}]: expected exactly one subcommand of {tool!r} ({', '.join(known)}), got {list(command)!r}")
if directory != ".":
raise ConfigurationError(f"{location}.directory: the {tool!r} tool checks the repository root, so the directory must be '.', got {directory!r}")
def _require_helper_subcommands(tool: str, commands: tuple[tuple[str, ...], ...], directory: object, location: str) -> None:
"""Accept only the helper tool's closed vocabulary: a subcommand name plus validated relative paths where it takes them, in the repository root."""
for index, command in enumerate(commands):
problem = validate_helper_command(command, f"{location}.build.commands[{index}]")
if problem:
raise ConfigurationError(problem)
if len(set(commands)) != len(commands):
raise ConfigurationError(f"{location}.build.commands: a {tool!r} check is declared twice; each check runs once")
if directory != ".":
raise ConfigurationError(f"{location}.directory: the {tool!r} tool checks the repository root, so the directory must be '.', got {directory!r}")
def _parse_artifact_format(value: object, tool: str, location: str) -> str:
"""Validate an artifact's delivery format; a package format needs the build tool that produces that package."""
artifact_format = _text(value, location)
if artifact_format not in ARTIFACT_FORMATS:
raise ConfigurationError(f"{location}: unsupported artifact format {artifact_format!r}; supported: {', '.join(ARTIFACT_FORMATS)}")
allowed_tools = BUILD_TOOL_BY_PACKAGE_FORMAT.get(artifact_format)
if allowed_tools is not None and tool not in allowed_tools:
raise ConfigurationError(f"{location}: format {artifact_format!r} needs build tool {' or '.join(map(repr, allowed_tools))}, not {tool!r}")
return artifact_format
def _parse_requires(value: object, tool: str, location: str) -> tuple[str, ...]:
"""Validate the optional runtime profiles: only ones the build tool defines. Every tool with profiles runs only in a catalog image."""
requires = _text_list(value, location, allow_empty=True)
supported = BUILD_TOOLS[tool].profiles
for profile in requires:
if profile not in supported:
raise ConfigurationError(f"{location}: unsupported runtime profile {profile!r} for build tool {tool!r}; supported: {', '.join(supported) or 'none'}")
if len(set(requires)) != len(requires):
raise ConfigurationError(f"{location}: duplicate runtime profile")
return tuple(sorted(requires))
def _parse_build_commands(build: dict[str, object], location: str) -> tuple[tuple[str, ...], ...]:
"""Normalize `arguments` (one command) or `commands` (a list of argument lists) into one canonical sequence."""
if "arguments" in build and "commands" in build:
raise ConfigurationError(f"{location}: arguments and commands are mutually exclusive; declare exactly one")
if "arguments" in build:
return (_text_list(build["arguments"], f"{location}.arguments"),)
if "commands" not in build:
raise ConfigurationError(f"{location}: one of arguments or commands is required")
commands = build["commands"]
if not isinstance(commands, list) or not commands:
raise ConfigurationError(f"{location}.commands: expected a non-empty list of argument lists")
return tuple(_text_list(command, f"{location}.commands[{index}]") for index, command in enumerate(cast("list[object]", commands)))
def _parse_report(value: object, location: str) -> Report:
"""Normalize a legacy glob string (`junit` names JUnit, any other name is a plain file) or a typed report object."""
if isinstance(value, str):
name = location.rsplit(".", 1)[-1]
return Report(format="junit" if name == "junit" else "file", paths=(_relative_path(value, location),))
fields = _mapping(value, location, ("format", "paths", "required"))
report_format = _text(_required(fields, "format", location), f"{location}.format")
if report_format not in REPORT_FORMATS:
raise ConfigurationError(f"{location}.format: unsupported report format {report_format!r}; supported: {', '.join(REPORT_FORMATS)}")
paths = tuple(_relative_path(path, f"{location}.paths[{index}]") for index, path in enumerate(_text_list(_required(fields, "paths", location), f"{location}.paths")))
required = fields.get("required", True)
if not isinstance(required, bool):
raise ConfigurationError(f"{location}.required: expected true or false, got {required!r}")
return Report(format=report_format, paths=paths, required=required)
def _parse_channel_trigger(value: object, location: str) -> tuple[bool, tuple[str, ...], tuple[str, ...]]:
"""Validate a channel trigger into (manual, push branch patterns, push tag patterns)."""
trigger = _mapping(value, location, ("push_branches", "push_tags", "manual"))
manual = _optional_flag(trigger, "manual", location)
if "push_branches" not in trigger and "push_tags" not in trigger and not manual:
raise ConfigurationError(f"{location}: declare push_branches, push_tags or both, or manual: true")
branches = _text_list(trigger["push_branches"], f"{location}.push_branches") if "push_branches" in trigger else ()
for branch in branches:
_require_branch_pattern(branch, f"{location}.push_branches")
tags = _text_list(trigger["push_tags"], f"{location}.push_tags") if "push_tags" in trigger else ()
for tag in tags:
if _TAG_PATTERN.fullmatch(tag) is None:
raise ConfigurationError(f"{location}.push_tags: expected a tag name or a pattern using only * and ?, got {tag!r}")
return manual, branches, tags
def _optional_flag(fields: dict[str, object], key: str, location: str) -> bool:
"""Read an optional boolean that defaults to false; anything else is an error naming it."""
flag = fields.get(key, False)
if not isinstance(flag, bool):
raise ConfigurationError(f"{location}.{key}: expected true or false, got {flag!r}")
return flag
def _parse_release_channels(value: object) -> tuple[ReleaseChannel, ...]:
"""Validate release channels: their eligibility triggers and optional versioning rules. Nothing here delivers anything."""
channels: list[ReleaseChannel] = []
for name, entry in sorted(_named_entries(value, "release.channels").items()):
location = f"release.channels.{name}"
fields = _mapping(entry, location, ("trigger", "versioning", "tag_policy", "tag_create", "central_completion", "release_record", "effective_version", "version_write_back"))
manual, branches, tags = _parse_channel_trigger(_required(fields, "trigger", location), f"{location}.trigger")
versioning = _text(fields["versioning"], f"{location}.versioning") if "versioning" in fields else ""
if versioning and versioning not in VERSIONING_RULES:
raise ConfigurationError(f"{location}.versioning: unsupported rule {versioning!r}; supported: {', '.join(VERSIONING_RULES)}")
tag_policy = _text(fields["tag_policy"], f"{location}.tag_policy") if "tag_policy" in fields else ""
if tag_policy and tag_policy not in TAG_POLICIES:
raise ConfigurationError(f"{location}.tag_policy: unsupported policy {tag_policy!r}; supported: {', '.join(TAG_POLICIES)}")
central_completion = _optional_flag(fields, "central_completion", location)
release_record = _optional_flag(fields, "release_record", location)
if release_record and tag_policy not in (TAG_POLICY_STABLE, TAG_POLICY_V_PREFIX):
raise ConfigurationError(f"{location}.release_record: needs a tag_policy of {TAG_POLICY_STABLE} or {TAG_POLICY_V_PREFIX}, because the record belongs to an existing tag")
tag_create = _optional_flag(fields, "tag_create", location)
if tag_create and tag_policy not in (TAG_POLICY_STABLE, TAG_POLICY_V_PREFIX):
raise ConfigurationError(f"{location}.tag_create: needs a tag_policy of {TAG_POLICY_STABLE} or {TAG_POLICY_V_PREFIX} to say which tag to create")
rule, version_property, rc_baseline = _parse_effective_version(fields.get("effective_version"), f"{location}.effective_version")
write_back = _parse_write_back(fields["version_write_back"], f"{location}.version_write_back", branches=bool(branches), tags=bool(tags)) if "version_write_back" in fields else None
channels.append(
ReleaseChannel(
name=name,
push_branches=tuple(sorted(set(branches))),
versioning=versioning,
tag_policy=tag_policy,
tag_create=tag_create,
push_tags=tuple(sorted(set(tags))),
central_completion=central_completion,
release_record=release_record,
effective_version=rule,
version_property=version_property,
rc_baseline=rc_baseline,
version_write_back=write_back,
manual=manual,
)
)
return tuple(channels)
def _parse_effective_version(value: object, location: str) -> tuple[str, str, str]:
"""Validate `effective_version: {rule, property, rc_baseline}` into (rule, property, rc_baseline); absent is ("", "", "") (UNTESTED)."""
if value is None:
return "", "", ""
fields = _mapping(value, location, ("rule", "property", "rc_baseline"))
rule = _text(_required(fields, "rule", location), f"{location}.rule")
if rule not in EFFECTIVE_VERSION_RULES:
raise ConfigurationError(f"{location}.rule: unsupported rule {rule!r}; supported: {', '.join(EFFECTIVE_VERSION_RULES)}")
version_property = _text(fields["property"], f"{location}.property") if "property" in fields else DEFAULT_VERSION_PROPERTY
if VERSION_PROPERTY_PATTERN.fullmatch(version_property) is None:
raise ConfigurationError(f"{location}.property: expected a Maven property name, got {version_property!r}")
rc_baseline = _text(fields["rc_baseline"], f"{location}.rc_baseline") if "rc_baseline" in fields else RC_BASELINE_NONE
if rc_baseline not in RC_BASELINES:
raise ConfigurationError(f"{location}.rc_baseline: unsupported value {rc_baseline!r}; supported: {', '.join(RC_BASELINES)}")
return rule, version_property, rc_baseline
_WRITE_BACK_BRANCH = re.compile(r"[A-Za-z0-9][A-Za-z0-9._/-]*")
def _parse_write_back(value: object, location: str, *, branches: bool, tags: bool) -> WriteBackPolicy:
"""Validate `version_write_back: {file, target_branch, require_push}` for a channel that is branch-driven or tag-driven, never both (UNTESTED)."""
fields = _mapping(value, location, ("file", "target_branch", "require_push"))
if branches == tags:
raise ConfigurationError(f"{location}: a channel that writes a version back must be triggered by push_branches only or by push_tags only")
file_name = _text(fields["file"], f"{location}.file") if "file" in fields else WRITE_BACK_FILE
if file_name != WRITE_BACK_FILE:
raise ConfigurationError(f"{location}.file: only {WRITE_BACK_FILE!r} is supported")
target = _text(fields["target_branch"], f"{location}.target_branch") if "target_branch" in fields else ""
if tags and _WRITE_BACK_BRANCH.fullmatch(target) is None:
raise ConfigurationError(f"{location}.target_branch: a tag-driven channel must name the branch the bump is pushed to")
if branches and target:
raise ConfigurationError(f"{location}.target_branch: a branch-driven channel pushes to the branch it released from")
require_push = fields.get("require_push", False)
if not isinstance(require_push, bool):
raise ConfigurationError(f"{location}.require_push: expected true or false, got {require_push!r}")
return WriteBackPolicy(file=WRITE_BACK_FILE, target_branch=target, require_push=require_push)
def _require_write_back_publication(channels: tuple[ReleaseChannel, ...], publications: tuple[Publication, ...]) -> None:
"""Require a Maven publication on every channel that writes a version back, because only published Maven evidence justifies it (UNTESTED)."""
for channel in channels:
if channel.version_write_back is not None and not any(channel.name in publication.channels and publication.format == MAVEN_ARTIFACT_FORMAT for publication in publications):
raise ConfigurationError(f"release.channels.{channel.name}.version_write_back: needs a Maven publication on this channel, whose registry and completion evidence the write waits for")
def _parse_publications(value: object, components: tuple[Component, ...], channels: tuple[ReleaseChannel, ...]) -> tuple[Publication, ...]:
"""Validate every publication against the declared components, artifact formats and channels; ambiguity is an error."""
by_component = {component.name: component for component in components}
by_channel = {channel.name: channel for channel in channels}
publications: list[Publication] = []
for name, entry in sorted(_named_entries(value, "release.publications").items()):
location = f"release.publications.{name}"
fields = _mapping(entry, location, ("component", "artifact", "destination", "channels", "platform", "deploy"))
component_name = _text(_required(fields, "component", location), f"{location}.component")
component = by_component.get(component_name)
if component is None:
raise ConfigurationError(f"{location}.component: no component named {component_name!r}")
artifact = _text(_required(fields, "artifact", location), f"{location}.artifact")
if artifact not in component.artifacts:
raise ConfigurationError(f"{location}.artifact: component {component_name!r} declares no artifact {artifact!r}")
artifact_format = component.artifact_formats.get(artifact)
if artifact_format is None:
raise ConfigurationError(f"{location}.artifact: artifact {artifact!r} declares no format, so it cannot be published")
channel_names = _text_list(_required(fields, "channels", location), f"{location}.channels")
if len(set(channel_names)) != len(channel_names) or any(channel not in by_channel for channel in channel_names):
raise ConfigurationError(f"{location}.channels: expected distinct names of declared release channels, got {list(channel_names)}")
unreleasable = sorted(channel for channel in channel_names if not by_channel[channel].push_branches and not by_channel[channel].push_tags)
if unreleasable:
# every channel used to have a push trigger, so a publication could always become eligible; a manual-only channel must not make one silently unreachable
raise ConfigurationError(f"{location}.channels: {unreleasable} have no push trigger (manual only), so nothing could ever be published on them")
deploy = _logical_destination(fields["deploy"], f"{location}.deploy") if "deploy" in fields else ""
if deploy and artifact_format != CONTAINER_ARTIFACT_FORMAT:
raise ConfigurationError(f"{location}.deploy: only a container artifact can be pinned in a GitOps destination")
publications.append(
Publication(
name=name,
component=component_name,
artifact=artifact,
format=artifact_format,
platform=_publication_platform(fields, component, location),
destination=_logical_destination(_required(fields, "destination", location), f"{location}.destination"),
channels=tuple(sorted(channel_names)),
deploy=deploy,
)
)
return tuple(publications)
def _publication_platform(fields: dict[str, object], component: Component, location: str) -> str:
"""Return the platform whose build supplies the artifact: the stated one, or the component's only platform."""
if "platform" in fields:
platform = _text(fields["platform"], f"{location}.platform")
if platform not in component.platforms:
raise ConfigurationError(f"{location}.platform: component {component.name!r} does not build for {platform!r}")
return platform
if len(component.platforms) != 1:
raise ConfigurationError(f"{location}.platform: component {component.name!r} builds for several platforms ({', '.join(component.platforms)}); name one")
return component.platforms[0]
def _logical_destination(value: object, location: str) -> str:
"""Require a logical destination name, never a URL, host or credential."""
destination = _text(value, location)
if _DESTINATION_PATTERN.fullmatch(destination) is None:
raise ConfigurationError(f"{location}: expected a logical destination name (lowercase letters, digits, dots, hyphens), got {destination!r}")
return destination
def _mapping(
value: object,
location: str,
supported: tuple[str, ...],
approved_but_unimplemented: tuple[str, ...] = (),
) -> dict[str, object]:
"""Require a mapping whose keys are all supported, naming approved-but-unimplemented fields explicitly."""
if not isinstance(value, dict):
raise ConfigurationError(f"{location}: expected a mapping")
mapping = cast("dict[object, object]", value)
for key in sorted(mapping, key=str):
if key in approved_but_unimplemented:
raise ConfigurationError(f"{location}.{key}: described by the approved design but not implemented in this pipeline slice")
if key not in supported:
raise ConfigurationError(f"{location}.{key}: unsupported field; supported fields: {', '.join(supported)}")
return {str(key): item for key, item in mapping.items()}
def _named_entries(value: object, location: str) -> dict[str, object]:
"""Require a mapping of stable semantic identifiers to values."""
if not isinstance(value, dict):
raise ConfigurationError(f"{location}: expected a mapping")
entries: dict[str, object] = {}
for key, item in cast("dict[object, object]", value).items():
if not isinstance(key, str) or _IDENTIFIER_PATTERN.fullmatch(key) is None:
raise ConfigurationError(f"{location}.{key}: names are lowercase letters, digits and hyphens, starting with a letter")
entries[key] = item
return entries
def _required(mapping: dict[str, object], key: str, location: str) -> object:
"""Return a required field or fail naming it."""
if key not in mapping:
raise ConfigurationError(f"{location}.{key}: required field is missing")
return mapping[key]
def _text(value: object, location: str) -> str:
"""Require a non-empty string; versions must be quoted so YAML does not turn them into numbers."""
if not isinstance(value, str) or not value.strip() or "\x00" in value:
raise ConfigurationError(f"{location}: expected a non-empty quoted string")
return value
def _text_list(value: object, location: str, *, allow_empty: bool = False) -> tuple[str, ...]:
"""Require a list of strings (non-empty unless allowed), kept as separate tokens and never joined into a command line."""
if not isinstance(value, list) or (not value and not allow_empty):
raise ConfigurationError(f"{location}: expected a non-empty list" if not allow_empty else f"{location}: expected a list")
return tuple(_text(item, f"{location}[{index}]") for index, item in enumerate(cast("list[object]", value)))
def _relative_path(value: object, location: str) -> str:
"""Require a relative path or glob that stays inside the repository."""
text = _text(value, location)
if "\\" in text or PurePosixPath(text).is_absolute() or ".." in PurePosixPath(text).parts:
raise ConfigurationError(f"{location}: expected a relative path inside the repository, got {text!r}")
return text