angee.workflows_integrate.archives

Bounded ZIP inspection and safe staging for archive extractor clients.

The workflows-integrate bridge owns archive mechanics shared by vendor extractors: one aggregate byte budget during recognition, normalized unique member names, subtree selection, and deterministic extraction that rejects traversal and symbolic links. Vendor addons remain responsible for recognizing their own archive vocabulary and translating staged data into domain ingest.

EXTRACT_DECLARED_LIMIT

Aggregate declared uncompressed bytes accepted for one staged subtree.

ArchiveError

class ArchiveError(Exception)

An archive cannot be inspected or staged within the bridge contract.

BoundedReader

class BoundedReader()

Seekable binary-stream proxy enforcing one aggregate read budget.

ZIP recognition runs once per registered extractor, so every probe must use an explicit byte budget and must never issue an unbounded read. Callers may reset :attr:remaining between independent candidates inside one archive.

read

def read(size: int = -1) -> bytes

Read at most the remaining budget, rejecting unbounded requests.

seek

def seek(offset: int, whence: int = 0) -> int

Seek without spending the byte-read budget.

tell

def tell() -> int

Return the wrapped stream position.

close

def close() -> None

Close the wrapped stream.

readable

def readable() -> bool

Return whether the wrapped stream can be read.

seekable

def seekable() -> bool

Return whether the wrapped stream supports ZIP random access.

archive_entries

def archive_entries(archive: zipfile.ZipFile) -> dict[str, zipfile.ZipInfo]

Return safe, unique archive members keyed by normalized POSIX path.

safe_member_name

def safe_member_name(value: str) -> str

Return a normalized relative member path or reject root traversal.

subtree_entries

def subtree_entries(entries: dict[str, zipfile.ZipInfo],
                    parent: PurePosixPath) -> dict[str, zipfile.ZipInfo]

Return only archive members inside parent.

extract_archive

def extract_archive(archive: zipfile.ZipFile, root: Path, *,
                    entries: dict[str, zipfile.ZipInfo]) -> None

Extract normalized files deterministically without following links.

stage_subtree

@contextmanager
def stage_subtree(archive: zipfile.ZipFile,
                  parent: PurePosixPath) -> Iterator[Path]

Safely stage one archive subtree under the bridge's declared-size cap.

The bridge owns the complete generic staging lifecycle: normalized member inventory, subtree selection, aggregate declared-size enforcement, deterministic extraction, and temporary-directory cleanup. The yielded path is the selected subtree root, ready for a vendor-specific delegate.