Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 10 additions & 6 deletions .github/workflows/pack-distribution.yml
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
# Verified pack supply-chain rehearsal (issue #191, ADR 0037).
#
# Manually triggered end-to-end integration for the distribution workflow. It
# stands up an ephemeral OCI registry (registry:2), publishes TechVault with its
# generated SBOM + provenance, signs the validated set-digest subject with
# keyless Sigstore (the GitHub Actions OIDC identity — the one thing a developer
# box cannot provide), then pulls, verifies signature + content, installs, and
# checks update/rollback guidance. Nothing here is standing infrastructure: the
# registry and the signing identity are spun up per run.
# stands up an ephemeral OCI registry (registry:2), publishes the pack directory
# named by the `pack` input (TechVault by default) with its generated SBOM +
# provenance, signs the validated set-digest subject with keyless Sigstore (the
# GitHub Actions OIDC identity — the one thing a developer box cannot provide),
# then pulls, verifies signature + content, installs, and checks update/rollback
# guidance. Nothing here is standing infrastructure: the registry and the signing
# identity are spun up per run.
#
# The workflow *shape* (triggers, permissions, pinning, step order) is guarded by
# tests/test_pack_distribution_workflow.py, and the real acceptance criterion —
Expand Down Expand Up @@ -68,4 +69,7 @@ jobs:
python -m pip install --require-hashes -r requirements/runtime.txt
python -m pip install --no-deps -e .
- name: Run the pack-distribution rehearsal
env:
# The dispatch value reaches the script as data, never as shell text.
REHEARSAL_PACK: ${{ inputs.pack }}
run: python tests_integration/distribution_rehearsal.py
3 changes: 2 additions & 1 deletion docs/public/distribution.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,8 @@ fails the build.
Signing and pushing to a registry use established tooling (cosign, oras) and a
keyless Sigstore identity, so they run in CI, not on a laptop. The
`Pack distribution rehearsal` workflow does the full publish → sign → install →
verify pass against an ephemeral registry.
verify pass against an ephemeral registry, for the pack directory named by its
`pack` input (default `packs/techvault`).

## Verify

Expand Down
90 changes: 71 additions & 19 deletions tests/test_distribution_rehearsal.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
"""Regression coverage for the pack-distribution rehearsal harness (#452).
"""Regression coverage for the pack-distribution rehearsal harness (#452, #451).

``tests_integration/distribution_rehearsal.py`` used to read ``techvault-0.1.0``
paths, so it stopped working once TechVault's ``pack.yaml`` moved past 0.1.0.
These tests run it at whatever version ``pack.yaml`` declares: the offline
rehearsal through the script's own entry point, and the registry rehearsal
against stub ``oras`` and ``cosign`` executables that keep pushed files in a
local store and serve them back on pull.
paths, so it stopped working once TechVault's ``pack.yaml`` moved past 0.1.0,
and it always rehearsed TechVault whatever pack the workflow's ``pack`` input
named. These tests run it at whatever version ``pack.yaml`` declares, for
TechVault and for the pack ``REHEARSAL_PACK`` names: the offline rehearsal
through the script's own entry point, and the registry rehearsal against stub
``oras`` and ``cosign`` executables that keep pushed files in a local store and
serve them back on pull.
"""

from __future__ import annotations
Expand All @@ -26,8 +28,18 @@

ROOT = Path(__file__).resolve().parents[1]
REHEARSAL = ROOT / "tests_integration" / "distribution_rehearsal.py"
PACK = yaml.safe_load((ROOT / "packs" / "techvault" / "pack.yaml").read_text(encoding="utf-8"))
STEM = f"{PACK['name']}-{PACK['version']}"
TECHVAULT = ROOT / "packs" / "techvault"
# A second hosted pack, so a rehearsal that ignores REHEARSAL_PACK is caught.
STUDY = ROOT / "packs" / "techvault-participant-study"


def _manifest(pack_dir: Path) -> dict:
return yaml.safe_load((pack_dir / "pack.yaml").read_text(encoding="utf-8"))


def _stem(pack_dir: Path) -> str:
manifest = _manifest(pack_dir)
return f"{manifest['name']}-{manifest['version']}"

# One stub stands in for both tools. It logs every call, and cosign calls just
# succeed, as a keyless signature from the GitHub OIDC identity would in CI.
Expand All @@ -50,6 +62,18 @@
"""


def _rehearse(pack: str | None) -> subprocess.CompletedProcess:
"""Run the script as the workflow does, with ``REHEARSAL_PACK`` set or unset."""

env = {key: value for key, value in os.environ.items()
if key not in ("REHEARSAL_REGISTRY", "REHEARSAL_PACK")}
if pack is not None:
env["REHEARSAL_PACK"] = pack
return subprocess.run(
[sys.executable, str(REHEARSAL)], cwd=ROOT, env=env,
capture_output=True, text=True, timeout=600)


def _load_rehearsal():
spec = importlib.util.spec_from_file_location("distribution_rehearsal", REHEARSAL)
module = importlib.util.module_from_spec(spec)
Expand All @@ -61,17 +85,37 @@ def _load_rehearsal():

class DistributionRehearsalTests(unittest.TestCase):
def test_offline_rehearsal_passes_at_the_pack_yaml_version(self) -> None:
env = {key: value for key, value in os.environ.items()
if key != "REHEARSAL_REGISTRY"}
completed = subprocess.run(
[sys.executable, str(REHEARSAL)], cwd=ROOT, env=env,
capture_output=True, text=True, timeout=600)
completed = _rehearse(None)

self.assertEqual(completed.returncode, 0, completed.stderr or completed.stdout)
self.assertIn(f"[rehearsal] published {STEM} ", completed.stdout)
self.assertIn(f"[rehearsal] published {_stem(TECHVAULT)} ", completed.stdout)
self.assertIn("[rehearsal] OFFLINE REHEARSAL PASSED", completed.stdout)

def test_registry_rehearsal_pushes_the_built_release_under_the_pack_version(self) -> None:
def test_offline_rehearsal_rehearses_the_pack_the_dispatch_names(self) -> None:
completed = _rehearse("packs/techvault-participant-study")

self.assertEqual(completed.returncode, 0, completed.stderr or completed.stdout)
self.assertIn(f"[rehearsal] published {_stem(STUDY)} ", completed.stdout)
self.assertIn("[rehearsal] OFFLINE REHEARSAL PASSED", completed.stdout)

def test_a_directory_that_is_not_a_pack_stops_before_publishing(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
# A pack.yaml whose name is not its directory's name.
mismatched = Path(tmp, "not-techvault")
mismatched.mkdir()
(mismatched / "pack.yaml").write_text(
(TECHVAULT / "pack.yaml").read_text(encoding="utf-8"), encoding="utf-8")
for pack in ("docs", "packs/no-such-pack", str(mismatched)):
with self.subTest(pack=pack):
completed = _rehearse(pack)

self.assertNotEqual(completed.returncode, 0, completed.stdout)
self.assertIn(f"rehearsal stopped: {pack} (", completed.stderr)
self.assertIn("is not a pack", completed.stderr)
self.assertNotIn("[rehearsal] published", completed.stdout)

def _assert_registry_rehearsal(self, pack_dir: Path) -> None:
manifest, stem = _manifest(pack_dir), _stem(pack_dir)
rehearsal = _load_rehearsal()
with tempfile.TemporaryDirectory() as tmp:
store, stubs = Path(tmp, "store"), Path(tmp, "bin")
Expand All @@ -85,19 +129,27 @@ def test_registry_rehearsal_pushes_the_built_release_under_the_pack_version(self
"REHEARSAL_STUB_STORE": str(store)}
output = io.StringIO()
with mock.patch.dict(os.environ, env), contextlib.redirect_stdout(output):
rehearsal.registry_rehearsal("registry.invalid", str(Path(tmp, "work")))
rehearsal.registry_rehearsal(
"registry.invalid", str(Path(tmp, "work")), str(pack_dir))
calls = [json.loads(line) for line in
(store / "calls.jsonl").read_text(encoding="utf-8").splitlines()]

push = next(call for call in calls if call[:2] == ["oras", "push"])
self.assertEqual(push[3], f"registry.invalid/{PACK['name']}:{PACK['version']}")
self.assertEqual(push[3], f"registry.invalid/{manifest['name']}:{manifest['version']}")
self.assertEqual(
[spec.split(":", 1)[0] for spec in push[4:]],
[f"{STEM}.tar.gz", "release.yaml", f"{STEM}.cdx.json", f"{STEM}.provenance.json"])
[f"{stem}.tar.gz", "release.yaml", f"{stem}.cdx.json", f"{stem}.provenance.json"])
self.assertIn(["cosign", "sign", "--yes", "--allow-http-registry",
f"registry.invalid/{PACK['name']}@sha256:{'0' * 64}"], calls)
f"registry.invalid/{manifest['name']}@sha256:{'0' * 64}"], calls)
# Passing means the pack-named consumer directories passed static validation.
self.assertIn("[rehearsal] REGISTRY REHEARSAL PASSED", output.getvalue())

def test_registry_rehearsal_pushes_the_built_release_under_the_pack_version(self) -> None:
self._assert_registry_rehearsal(TECHVAULT)

def test_registry_rehearsal_names_the_repository_and_consumers_after_the_pack(self) -> None:
self._assert_registry_rehearsal(STUDY)


if __name__ == "__main__":
unittest.main()
14 changes: 14 additions & 0 deletions tests/test_pack_distribution_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,20 @@ def index(predicate) -> int:
self.assertLess(cosign, rehearsal, "cosign must be installed before the rehearsal")
self.assertLess(oras, rehearsal, "oras must be installed before the rehearsal")

def test_rehearsal_step_passes_the_pack_input_through_env(self) -> None:
inputs = _triggers(self.data)["workflow_dispatch"]["inputs"]
self.assertEqual(inputs["pack"].get("default"), "packs/techvault")
step = next(s for s in self.steps if "distribution_rehearsal.py" in str(s.get("run", "")))
self.assertEqual(step.get("env", {}).get("REHEARSAL_PACK"), "${{ inputs.pack }}",
"the rehearsal must receive the pack input through env:")

def test_no_run_body_interpolates_an_expression(self) -> None:
# GitHub's script-injection guidance: expressions go through env:, so a
# dispatch value is never pasted into the generated shell script.
for step in self.steps:
self.assertNotIn("${{", str(step.get("run", "")),
f"step {step.get('name')!r} interpolates an expression in run:")


if __name__ == "__main__":
unittest.main()
71 changes: 50 additions & 21 deletions tests_integration/distribution_rehearsal.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@
Two rehearsals, one script:

* The **offline** rehearsal runs everywhere with no external tooling: it
publishes TechVault, exports a deterministic archive, proves clean
publishes the pack named by ``REHEARSAL_PACK`` (default ``packs/techvault``),
exports a deterministic archive, proves clean
reproduction, then installs, verifies, updates, and reads rollback guidance
through the archive route. This is the inner-loop confidence check.
* The **registry** rehearsal runs only when a real OCI registry is reachable and
Expand Down Expand Up @@ -34,15 +35,40 @@
sys.path.insert(0, os.path.join(_REPO, "src"))

from raes_env_packs import distribution as dist # noqa: E402
from raes_env_packs import release, verify # noqa: E402
from raes_env_packs import release, validation, verify # noqa: E402

_TECHVAULT = os.path.join(_REPO, "packs", "techvault")
_DEFAULT_PACK = "packs/techvault"


def _log(message: str) -> None:
print(f"[rehearsal] {message}", flush=True)


def _pack_root() -> tuple[str, str]:
"""Return the pack directory ``REHEARSAL_PACK`` names, as given and resolved.

The workflow passes its ``pack`` dispatch input here through ``env:``. An
unset or empty value means the default pack; a relative value is resolved
against the repository root, where the workflow runs.
"""

requested = os.environ.get("REHEARSAL_PACK") or _DEFAULT_PACK
return requested, os.path.join(_REPO, requested)


def _require_pack(requested: str, pack_root: str) -> None:
"""Stop before anything is published unless ``pack_root`` is a pack.

``validate_pack`` is the one static-validation authority (ADR 0013), so a
directory it rejects never reaches the publish gates (ADR 0037).
"""

result = validation.validate_pack(pack_root)
if not result.ok:
raise SystemExit(f"rehearsal stopped: {requested} ({pack_root}) is not a pack: "
f"{'; '.join(result.errors)}")


class _Published(NamedTuple):
"""Where one publish wrote its release, named from the build's own metadata."""

Expand All @@ -53,16 +79,16 @@ class _Published(NamedTuple):
evidence_files: tuple[tuple[str, str], ...]


def _publish(workdir: str) -> _Published:
"""Publish TechVault, export its archive, and say where both were written.
def _publish(workdir: str, pack_root: str) -> _Published:
"""Publish the pack at ``pack_root``, export its archive, and say where both were written.

The release directory, archive, and evidence names follow the pack name and
version that ``build_release`` recorded in the release metadata, so the
rehearsal tracks every ``pack.yaml`` version bump instead of a literal.
"""

release_out = os.path.join(workdir, "release")
metadata, failures = release.build_release(_TECHVAULT, release_out, publish=True)
metadata, failures = release.build_release(pack_root, release_out, publish=True)
if failures:
raise SystemExit(f"publish failed: {failures}")
pack = metadata["release"]["pack"]
Expand All @@ -71,10 +97,10 @@ def _publish(workdir: str) -> _Published:
if not os.path.isfile(os.path.join(evidence_dir, "release.yaml")):
raise SystemExit(f"publish wrote no release.yaml under {evidence_dir}")
archive = os.path.join(workdir, f"{stem}.tar.gz")
first = dist.export_pack_archive(_TECHVAULT, archive)
first = dist.export_pack_archive(pack_root, archive)
# Clean reproduction: the same source exports byte-identical bytes.
again = os.path.join(workdir, f"{stem}-again.tar.gz")
if dist.export_pack_archive(_TECHVAULT, again) != first:
if dist.export_pack_archive(pack_root, again) != first:
raise SystemExit("archive export is not reproducible")
_log(f"published {stem} + exported archive {first}")
evidence = metadata["evidence"]
Expand All @@ -87,24 +113,25 @@ def _publish(workdir: str) -> _Published:
))


def offline_rehearsal(workdir: str) -> None:
def offline_rehearsal(workdir: str, pack_root: str) -> None:
"""Publish -> export -> install -> verify -> update -> rollback, no network."""

published = _publish(workdir)
published = _publish(workdir, pack_root)
evidence_dir, archive = published.evidence_dir, published.archive
selector = dist.Selector(
repository="local", reference=archive, digest_domain=dist.DIGEST_DOMAIN_ARCHIVE)
transport = dist.ArchiveTransport()
target = os.path.join(workdir, "consumer", "techvault")
# Static validation requires an installed pack's directory to carry its name.
target = os.path.join(workdir, "consumer", published.name)

install = dist.plan_install(
_TECHVAULT, evidence_dir, target, selector=selector, transport=transport)
pack_root, evidence_dir, target, selector=selector, transport=transport)
if not install.applicable:
raise SystemExit(f"install plan not applicable: {install.render_human()}")
# The offline route carries no signature; accept unsigned explicitly. The
# authenticated path is exercised by the registry rehearsal.
receipt = dist.apply_install(
install, _TECHVAULT, evidence_dir, target,
install, pack_root, evidence_dir, target,
authorized=True,
policy=dist.PromotionPolicy(selector=selector, transport=transport,
require_signature=False))
Expand All @@ -116,10 +143,10 @@ def offline_rehearsal(workdir: str) -> None:
_log("installed pack verifies")

update = dist.plan_update(
target, _TECHVAULT, evidence_dir, selector=selector, transport=transport)
target, pack_root, evidence_dir, selector=selector, transport=transport)
_log(f"update plan changes: {[c.category for c in update.changes]}")
dist.apply_install(
update, _TECHVAULT, evidence_dir, target,
update, pack_root, evidence_dir, target,
authorized=True,
policy=dist.PromotionPolicy(selector=selector, transport=transport,
require_signature=False))
Expand All @@ -136,10 +163,10 @@ def _run(cmd: list[str], **kwargs) -> subprocess.CompletedProcess:
return subprocess.run(cmd, check=True, text=True, **kwargs)


def registry_rehearsal(registry: str, workdir: str) -> None:
def registry_rehearsal(registry: str, workdir: str, pack_root: str) -> None:
"""Push -> keyless-sign -> pull -> verify-signature -> install, over a registry."""

published = _publish(workdir)
published = _publish(workdir, pack_root)
archive = published.archive
repo = f"{registry}/{published.name}"
ref = f"{repo}:{published.version}"
Expand Down Expand Up @@ -199,7 +226,7 @@ def cosign_verifier(subject_set_digest: str, provenance: dict) -> bool:
# Stage the pulled archive into a clean, pack-named consumer directory and
# verify those staged bytes against the pulled evidence (release.yaml + SBOM +
# provenance now live under `pulled/`).
staged = os.path.join(workdir, "consumer", "techvault")
staged = os.path.join(workdir, "consumer", published.name)
os.makedirs(os.path.dirname(staged), exist_ok=True)
dist.stage_pack_archive(pulled_archive, staged)
profile, sbom_doc, prov_doc = verify.load_release_evidence(pulled)
Expand All @@ -215,7 +242,7 @@ def cosign_verifier(subject_set_digest: str, provenance: dict) -> bool:
selector = dist.Selector(
repository=repo, reference=pulled_archive, digest_domain=dist.DIGEST_DOMAIN_ARCHIVE)
transport = dist.ArchiveTransport()
target = os.path.join(workdir, "installed", "techvault")
target = os.path.join(workdir, "installed", published.name)
plan = dist.plan_install(
staged, pulled, target, selector=selector, transport=transport,
signature_verifier=cosign_verifier)
Expand All @@ -228,12 +255,14 @@ def cosign_verifier(subject_set_digest: str, provenance: dict) -> bool:


def main() -> int:
requested, pack_root = _pack_root()
_require_pack(requested, pack_root)
with tempfile.TemporaryDirectory() as workdir:
offline_rehearsal(os.path.join(workdir, "offline"))
offline_rehearsal(os.path.join(workdir, "offline"), pack_root)
registry = os.environ.get("REHEARSAL_REGISTRY")
have_tools = shutil.which("oras") and shutil.which("cosign")
if registry and have_tools:
registry_rehearsal(registry, os.path.join(workdir, "registry"))
registry_rehearsal(registry, os.path.join(workdir, "registry"), pack_root)
else:
_log("registry rehearsal skipped (set REHEARSAL_REGISTRY + install oras/cosign)")
return 0
Expand Down
Loading