98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Replay the reviewed Hermes Dockerfile heredocs for pinned Kaniko v1.23.2.
|
|
|
|
Kaniko v1.23.2 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 _continued_run_header(lines: list[str], index: int) -> tuple[str, int]:
|
|
"""Return one logical RUN header, including backslash continuations."""
|
|
header = lines[index]
|
|
while header.rstrip().endswith("\\") and index + 1 < len(lines):
|
|
index += 1
|
|
header = f"{header}\n{lines[index]}"
|
|
return header, index
|
|
|
|
|
|
def extract_blocks(source: str) -> list[tuple[tuple[str, ...], str]]:
|
|
"""Inventory every RUN heredoc, then return only the reviewed contract."""
|
|
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:
|
|
stripped = lines[index].lstrip()
|
|
if stripped[:3].upper() == "RUN" and (
|
|
len(stripped) == 3 or stripped[3].isspace()
|
|
):
|
|
header, index = _continued_run_header(lines, index)
|
|
if "<<" in header:
|
|
raise ValueError(
|
|
"unsupported RUN heredoc outside the exact reviewed contract"
|
|
)
|
|
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())
|