226 lines
8.7 KiB
Python
226 lines
8.7 KiB
Python
"""Deterministic Candidate artifact and package-closure construction."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import stat
|
|
import tarfile
|
|
from typing import Mapping
|
|
|
|
from .errors import EvidenceIncomplete, ReleaseBlocked
|
|
from .models import is_git_oid
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ArtifactFile:
|
|
path: str
|
|
content: bytes
|
|
mode: int = 0o644
|
|
|
|
def __post_init__(self) -> None:
|
|
parsed = PurePosixPath(self.path)
|
|
if (
|
|
parsed.is_absolute()
|
|
or ".." in parsed.parts
|
|
or not self.path
|
|
or "\\" in self.path
|
|
or self.path != self.path.strip()
|
|
):
|
|
raise ValueError("artifact paths must be canonical, relative and traversal-free")
|
|
if ".git" in parsed.parts:
|
|
raise ValueError(".git content is forbidden")
|
|
if self.mode not in (0o644, 0o755):
|
|
raise ValueError("artifact mode must be 0644 or 0755")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ArtifactManifest:
|
|
source_oid: str
|
|
tree_oid: str
|
|
files: Mapping[str, str]
|
|
artifact_sha256: str
|
|
package_closure_sha256: str
|
|
format_version: int = 1
|
|
secrets_included: bool = False
|
|
|
|
def as_json_bytes(self) -> bytes:
|
|
payload = {
|
|
"format_version": self.format_version,
|
|
"source_oid": self.source_oid,
|
|
"tree_oid": self.tree_oid,
|
|
"files": dict(sorted(self.files.items())),
|
|
"artifact_sha256": self.artifact_sha256,
|
|
"package_closure_sha256": self.package_closure_sha256,
|
|
"secrets_included": self.secrets_included,
|
|
}
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
|
@property
|
|
def manifest_sha256(self) -> str:
|
|
return sha256_bytes(self.as_json_bytes())
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BuiltArtifact:
|
|
content: bytes
|
|
manifest: ArtifactManifest
|
|
|
|
|
|
class ArtifactBuilder:
|
|
MAX_FILE_BYTES = 8 * 1024 * 1024
|
|
MAX_ARTIFACT_BYTES = 64 * 1024 * 1024
|
|
FORBIDDEN_MARKERS = (
|
|
b"begin private " + b"key",
|
|
b"authorization:" + b" bearer ",
|
|
b"password" + b"=",
|
|
b"token" + b"=",
|
|
b"secret" + b"=",
|
|
b"api_key" + b"=",
|
|
)
|
|
|
|
@staticmethod
|
|
def _validate_oid(name: str, value: str) -> None:
|
|
if not is_git_oid(value):
|
|
raise ValueError(f"{name} must be a full lowercase Git OID")
|
|
|
|
def build_from_directory(
|
|
self, source_oid: str, tree_oid: str, source: str | Path
|
|
) -> BuiltArtifact:
|
|
root = Path(source)
|
|
if root.is_symlink() or not root.is_dir():
|
|
raise ReleaseBlocked(
|
|
"ARTIFACT_SOURCE_LINK_REJECTED", "artifact source must be a real directory"
|
|
)
|
|
files: list[ArtifactFile] = []
|
|
for directory, dirnames, filenames in os.walk(root, followlinks=False):
|
|
current = Path(directory)
|
|
for name in tuple(dirnames):
|
|
item = current / name
|
|
if item.is_symlink() or name == ".git":
|
|
raise ReleaseBlocked(
|
|
"ARTIFACT_SOURCE_LINK_REJECTED",
|
|
"artifact source contains a linked or repository directory",
|
|
)
|
|
for name in filenames:
|
|
item = current / name
|
|
info = item.lstat()
|
|
if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
|
|
raise ReleaseBlocked(
|
|
"ARTIFACT_SOURCE_LINK_REJECTED",
|
|
"artifact source contains a link or special file",
|
|
)
|
|
if info.st_size > self.MAX_FILE_BYTES:
|
|
raise ReleaseBlocked("ARTIFACT_FILE_TOO_LARGE", "artifact file exceeds limit")
|
|
relative = item.relative_to(root).as_posix()
|
|
mode = 0o755 if info.st_mode & 0o111 else 0o644
|
|
files.append(ArtifactFile(relative, item.read_bytes(), mode))
|
|
return self.build(source_oid, tree_oid, tuple(files))
|
|
|
|
def build(
|
|
self, source_oid: str, tree_oid: str, files: tuple[ArtifactFile, ...]
|
|
) -> BuiltArtifact:
|
|
self._validate_oid("source_oid", source_oid)
|
|
self._validate_oid("tree_oid", tree_oid)
|
|
if not files:
|
|
raise EvidenceIncomplete("ARTIFACT_EMPTY", "artifact requires at least one file")
|
|
paths = [item.path for item in files]
|
|
if len(paths) != len(set(paths)):
|
|
raise ReleaseBlocked("ARTIFACT_DUPLICATE_PATH", "artifact contains duplicate paths")
|
|
total = 0
|
|
for item in files:
|
|
total += len(item.content)
|
|
if len(item.content) > self.MAX_FILE_BYTES or total > self.MAX_ARTIFACT_BYTES:
|
|
raise ReleaseBlocked("ARTIFACT_SIZE_LIMIT", "artifact exceeds Candidate limit")
|
|
lowered = item.content.lower()
|
|
if any(marker in lowered for marker in self.FORBIDDEN_MARKERS):
|
|
raise ReleaseBlocked(
|
|
"ARTIFACT_SECRET_MARKER",
|
|
"artifact contains a forbidden secret marker",
|
|
{"path": item.path},
|
|
)
|
|
output = io.BytesIO()
|
|
with tarfile.open(fileobj=output, mode="w", format=tarfile.PAX_FORMAT) as archive:
|
|
for item in sorted(files, key=lambda value: value.path):
|
|
info = tarfile.TarInfo(item.path)
|
|
info.size = len(item.content)
|
|
info.mode = item.mode
|
|
info.mtime = 0
|
|
info.uid = 0
|
|
info.gid = 0
|
|
info.uname = "root"
|
|
info.gname = "root"
|
|
archive.addfile(info, io.BytesIO(item.content))
|
|
content = output.getvalue()
|
|
file_hashes = {item.path: sha256_bytes(item.content) for item in files}
|
|
closure_payload = json.dumps(
|
|
{
|
|
"source_oid": source_oid,
|
|
"tree_oid": tree_oid,
|
|
"files": dict(sorted(file_hashes.items())),
|
|
"secrets_included": False,
|
|
},
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
manifest = ArtifactManifest(
|
|
source_oid=source_oid,
|
|
tree_oid=tree_oid,
|
|
files=file_hashes,
|
|
artifact_sha256=sha256_bytes(content),
|
|
package_closure_sha256=sha256_bytes(closure_payload),
|
|
secrets_included=False,
|
|
)
|
|
return BuiltArtifact(content=content, manifest=manifest)
|
|
|
|
def verify(self, built: BuiltArtifact) -> None:
|
|
if built.manifest.secrets_included:
|
|
raise EvidenceIncomplete("ARTIFACT_SECRET_FLAG", "manifest includes secrets")
|
|
if sha256_bytes(built.content) != built.manifest.artifact_sha256:
|
|
raise EvidenceIncomplete(
|
|
"ARTIFACT_DIGEST_MISMATCH", "artifact digest does not match manifest"
|
|
)
|
|
actual: dict[str, str] = {}
|
|
with tarfile.open(fileobj=io.BytesIO(built.content), mode="r:") as archive:
|
|
for member in archive.getmembers():
|
|
try:
|
|
ArtifactFile(member.name, b"", member.mode)
|
|
except ValueError as error:
|
|
raise EvidenceIncomplete(
|
|
"ARTIFACT_MEMBER_INVALID", "archive contains an unsafe member"
|
|
) from error
|
|
if not member.isfile() or member.name in actual:
|
|
raise EvidenceIncomplete(
|
|
"ARTIFACT_MEMBER_INVALID", "archive contains a link, special or duplicate member"
|
|
)
|
|
stream = archive.extractfile(member)
|
|
if stream is None:
|
|
raise EvidenceIncomplete("ARTIFACT_MEMBER_INVALID", "archive member is unreadable")
|
|
actual[member.name] = sha256_bytes(stream.read())
|
|
if actual != dict(built.manifest.files):
|
|
raise EvidenceIncomplete(
|
|
"ARTIFACT_FILE_MISMATCH", "artifact file closure does not match manifest"
|
|
)
|
|
closure_payload = json.dumps(
|
|
{
|
|
"source_oid": built.manifest.source_oid,
|
|
"tree_oid": built.manifest.tree_oid,
|
|
"files": dict(sorted(actual.items())),
|
|
"secrets_included": False,
|
|
},
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
if sha256_bytes(closure_payload) != built.manifest.package_closure_sha256:
|
|
raise EvidenceIncomplete(
|
|
"PACKAGE_CLOSURE_MISMATCH", "package closure digest does not match"
|
|
)
|