atlas-iac/dockerfiles/hermes-kaniko-heredoc-runner.py

80 lines
2.8 KiB
Python

#!/usr/bin/env python3
"""Replay the reviewed Hermes Dockerfile heredocs for pinned Kaniko.
Kaniko v1.24 parses shell heredoc bodies into ``RunCommand.Files`` but its RUN
implementation executes only ``CmdLine``. The interpreters therefore receive
empty stdin and exit successfully. Docker and BuildKit execute these blocks
normally, so this compatibility runner is enabled only by the Kaniko pipeline.
"""
from __future__ import annotations
import argparse
import subprocess
from pathlib import Path
MAX_DOCKERFILE_BYTES = 2_000_000
BLOCKS = (
("RUN node <<'NODE'", "NODE", ("node",)),
("RUN python - <<'PY'", "PY", ("python", "-")),
)
EXPECTED_COMMANDS = ("node", *("python",) * 7, "node")
def extract_blocks(source: str) -> list[tuple[tuple[str, ...], str]]:
"""Return only exact supported RUN heredocs in their reviewed order."""
markers = {start: (end, command) for start, end, command in BLOCKS}
lines = source.splitlines()
blocks: list[tuple[tuple[str, ...], str]] = []
index = 0
while index < len(lines):
marker = markers.get(lines[index])
if marker is None:
index += 1
continue
end, command = marker
body_start = index + 1
index = body_start
while index < len(lines) and lines[index] != end:
index += 1
if index == len(lines):
raise ValueError(f"unterminated {command[0]} heredoc")
body = "\n".join(lines[body_start:index]) + "\n"
if not body.strip():
raise ValueError(f"empty {command[0]} heredoc")
blocks.append((command, body))
index += 1
commands = tuple(command[0] for command, _body in blocks)
if commands != EXPECTED_COMMANDS:
raise ValueError(
"Hermes Dockerfile heredoc contract changed: "
f"expected {EXPECTED_COMMANDS!r}, received {commands!r}"
)
return blocks
def replay(dockerfile: Path) -> None:
"""Execute each exact heredoc in an isolated interpreter process."""
size = dockerfile.stat().st_size
if size < 1 or size > MAX_DOCKERFILE_BYTES:
raise ValueError("Hermes Dockerfile size is outside the reviewed boundary")
source = dockerfile.read_text(encoding="utf-8")
for index, (command, body) in enumerate(extract_blocks(source), start=1):
print(f"replaying reviewed Dockerfile heredoc {index}/{len(EXPECTED_COMMANDS)}")
subprocess.run(command, input=body, text=True, check=True)
def main() -> int:
"""Parse the one explicit Dockerfile path and replay its reviewed patches."""
parser = argparse.ArgumentParser()
parser.add_argument("--dockerfile", required=True, type=Path)
args = parser.parse_args()
replay(args.dockerfile)
return 0
if __name__ == "__main__":
raise SystemExit(main())