import gzip import hashlib import io import tarfile import struct from typing import List, Tuple class SourceArtifactError(ValueError): """Custom exception for source artifact processing errors.""" pass MAX_SOURCE_ARCHIVE_MEMBERS = 10_000 MAX_NORMALIZED_SOURCE_BYTES = 209_715_200 _SAFE_MESSAGES = { "INVALID": "Invalid source archive.", "LIMITS": "Source archive exceeds allowed limits.", "UNSUPPORTED": "Source archive contains unsupported entries.", "UNSAFE_PATHS": "Source archive has unsafe paths.", "DUPLICATE_PATHS": "Source archive has duplicate paths.", "MISSING_FILES": "Source archive is missing required build files.", } def _validate_archive_member_path( name: str, *, is_directory: bool, ) -> str: """ Validates and normalizes a tar member path. """ if not isinstance(name, str) or not name: raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) if is_directory and name.endswith('/'): name = name[:-1] if ( not name or '\\' in name or '\0' in name or name.startswith('/') or '//' in name ): raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) parts = name.split('/') if any(p in ('.', '..') for p in parts) or not all(parts): raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) return name def _validate_ustar_output_path(path: str) -> None: """ Validates that a normalized path is representable in USTAR format. """ _validate_archive_member_path(path, is_directory=False) try: path_bytes = path.encode('utf-8') except UnicodeEncodeError: raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) from None if len(path_bytes) > 255: raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) if b'/' in path_bytes: prefix_bytes, name_bytes = path_bytes.rsplit(b'/', 1) else: prefix_bytes, name_bytes = b"", path_bytes if not name_bytes: raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) if len(name_bytes) > 100 or len(prefix_bytes) > 155: raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) def normalize_gitea_archive( archive_bytes: bytes, *, required_paths: tuple[str, ...], ) -> tuple[bytes, dict]: """ Normalizes a Gitea source archive for deterministic builds. """ if not isinstance(archive_bytes, bytes) or not archive_bytes: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) if not isinstance(required_paths, tuple) or not required_paths: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) validated_req_paths = [] try: for p in required_paths: if not isinstance(p, str) or not p: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) validated_req_paths.append(_validate_archive_member_path(p, is_directory=False)) except SourceArtifactError: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) from None if len(validated_req_paths) != len(set(validated_req_paths)): raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) final_files: List[Tuple[tarfile.TarInfo, str]] = [] try: with gzip.GzipFile(fileobj=io.BytesIO(archive_bytes), mode="rb") as gzip_file: with tarfile.open(fileobj=gzip_file, mode="r:") as tar: members = tar.getmembers() if len(members) > MAX_SOURCE_ARCHIVE_MEMBERS: raise SourceArtifactError(_SAFE_MESSAGES["LIMITS"]) validated_members: List[Tuple[tarfile.TarInfo, str]] = [] total_size = 0 for member in members: is_dir = member.isdir() if member.isreg() or is_dir: normalized_path = _validate_archive_member_path(member.name, is_directory=is_dir) if member.isreg(): if member.size < 0: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) if total_size + member.size > MAX_NORMALIZED_SOURCE_BYTES: raise SourceArtifactError(_SAFE_MESSAGES["LIMITS"]) total_size += member.size validated_members.append((member, normalized_path)) elif member.issym() or member.islnk() or member.ischr() or member.isblk() or member.isfifo(): raise SourceArtifactError(_SAFE_MESSAGES["UNSUPPORTED"]) else: raise SourceArtifactError(_SAFE_MESSAGES["UNSUPPORTED"]) regular_files = [(m, p) for m, p in validated_members if m.isreg()] if not regular_files: raise SourceArtifactError(_SAFE_MESSAGES["MISSING_FILES"]) first_segments = {path.split('/')[0] for _, path in regular_files if '/' in path} wrapper_dir_stripped = False if len(first_segments) == 1 and all('/' in p for _, p in regular_files): wrapper_dir = first_segments.pop() temp_files: List[Tuple[tarfile.TarInfo, str]] = [] for member, path in regular_files: new_path = path.partition(f"{wrapper_dir}/")[2] final_path = _validate_archive_member_path(new_path, is_directory=False) temp_files.append((member, final_path)) final_files = temp_files wrapper_dir_stripped = True else: final_files = regular_files final_paths = [path for _, path in final_files] if len(final_paths) != len(set(final_paths)): raise SourceArtifactError(_SAFE_MESSAGES["DUPLICATE_PATHS"]) if not set(validated_req_paths).issubset(set(final_paths)): raise SourceArtifactError(_SAFE_MESSAGES["MISSING_FILES"]) out_buffer = io.BytesIO() with gzip.GzipFile(fileobj=out_buffer, mode='wb', mtime=0) as gz: with tarfile.open(fileobj=gz, mode='w:', format=tarfile.USTAR_FORMAT) as out_tar: for member, path in sorted(final_files, key=lambda item: item[1]): _validate_ustar_output_path(path) content_file = tar.extractfile(member) if content_file is None: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) content_bytes = content_file.read() if len(content_bytes) != member.size: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) info = tarfile.TarInfo(name=path) info.size = member.size info.mtime = 0 info.uid = 0 info.gid = 0 info.uname = "" info.gname = "" info.mode = 0o644 out_tar.addfile(info, io.BytesIO(content_bytes)) except SourceArtifactError: raise except (gzip.BadGzipFile, tarfile.TarError, EOFError, OSError, struct.error, ValueError): raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) from None except Exception: raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) from None normalized_bytes = out_buffer.getvalue() sha256_hash = hashlib.sha256(normalized_bytes).hexdigest() manifest = { "sha256": sha256_hash, "source_bytes": len(normalized_bytes), "required_paths": list(validated_req_paths), "wrapper_directory_stripped": wrapper_dir_stripped, } return normalized_bytes, manifest