ingress

azure_bootstrap.ingress

Tier 2 attachment / file-upload hardening.

Four gates, fixed order: extension → MIME → size → magic-byte. Use :class:AttachmentClassifier to run the whole pipeline; each gate is also independently usable for projects that need finer control.

Modules:

Name	Description
classifier	Attachment classifier — runs the four gates in fixed order.
extensions	Extension allowlist gate (cheapest, runs first in the classifier pipeline).
magic_bytes	Magic-byte classifier (authoritative — final gate in the pipeline).
mime	MIME allowlist gate (advisory — magic-byte gate is the real authority).
size	Size-cap gate. Raises OversizedAttachmentError over the cap.
zip_safety	Zip-bomb defense. Inspect infolist metadata BEFORE reading any entry.

Functions:

Name	Description
classify_bytes	Match the leading bytes against the signature table.
extension_matches_kind	Audit helper — is the filename's extension consistent with the bytes?
enforce_size_cap	Raise :class:OversizedAttachmentError when size_bytes > cap_bytes.
enforce_zip_safety_limits	Raise :class:ZipBombError when the archive metadata exceeds limits.

classify_bytes

classify_bytes(content: bytes, *, allowed: tuple[ClassifiedKind, ...] = ('pdf', 'zip')) -> ClassifiedKind

Match the leading bytes against the signature table.

Returns the first matched kind that's in allowed; otherwise "reject".

Source code in azure_bootstrap/ingress/magic_bytes.py

def classify_bytes(
    content: bytes,
    *,
    allowed: tuple[ClassifiedKind, ...] = ("pdf", "zip"),
) -> ClassifiedKind:
    """Match the leading bytes against the signature table.

    Returns the first matched kind that's in ``allowed``; otherwise ``"reject"``.
    """
    if not isinstance(content, (bytes, bytearray)):
        return "reject"
    head = bytes(content[:16])
    for kind, sigs in _SIGNATURES.items():
        if kind not in allowed:
            continue
        for sig in sigs:
            if head.startswith(sig):
                return kind
    return "reject"

extension_matches_kind

extension_matches_kind(filename: str, kind: ClassifiedKind) -> bool

Audit helper — is the filename's extension consistent with the bytes?

Source code in azure_bootstrap/ingress/magic_bytes.py

def extension_matches_kind(filename: str, kind: ClassifiedKind) -> bool:
    """Audit helper — is the filename's extension consistent with the bytes?"""
    if not filename or kind == "reject":
        return False
    suffix = PurePosixPath(filename).suffix.lower()
    expected = _EXTENSION_TO_KIND.get(suffix)
    return expected == kind

enforce_size_cap

enforce_size_cap(*, size_bytes: int, cap_bytes: int, filename: str, counter_name: str | None = None) -> None

Raise :class:OversizedAttachmentError when size_bytes > cap_bytes.

Source code in azure_bootstrap/ingress/size.py

def enforce_size_cap(
    *,
    size_bytes: int,
    cap_bytes: int,
    filename: str,
    counter_name: str | None = None,
) -> None:
    """Raise :class:`OversizedAttachmentError` when ``size_bytes > cap_bytes``."""
    if size_bytes > cap_bytes:
        if counter_name:
            bump_counter(counter_name)
        raise OversizedAttachmentError(
            f"attachment {filename!r} is {size_bytes} bytes; " f"exceeds cap of {cap_bytes} bytes"
        )

enforce_zip_safety_limits

enforce_zip_safety_limits(zf: ZipFile, *, filename: str, max_entries: int = MAX_ZIP_ENTRIES, max_uncompressed_bytes: int = MAX_ZIP_UNCOMPRESSED_BYTES, counter_name: str | None = None) -> None

Raise :class:ZipBombError when the archive metadata exceeds limits.

Inspects zf.infolist() only — does NOT call zf.read() (which would allocate the uncompressed bytes). The whole point is to gate on the declared metadata BEFORE any expansion.

Source code in azure_bootstrap/ingress/zip_safety.py

def enforce_zip_safety_limits(
    zf: zipfile.ZipFile,
    *,
    filename: str,
    max_entries: int = MAX_ZIP_ENTRIES,
    max_uncompressed_bytes: int = MAX_ZIP_UNCOMPRESSED_BYTES,
    counter_name: str | None = None,
) -> None:
    """Raise :class:`ZipBombError` when the archive metadata exceeds limits.

    Inspects ``zf.infolist()`` only — does NOT call ``zf.read()`` (which
    would allocate the uncompressed bytes). The whole point is to gate on
    the declared metadata BEFORE any expansion.
    """
    infos = zf.infolist()
    if len(infos) > max_entries:
        if counter_name:
            bump_counter(counter_name)
        raise ZipBombError(
            f"archive {filename!r} has {len(infos)} entries; " f"exceeds max_entries={max_entries}"
        )
    total = sum(info.file_size for info in infos)
    if total > max_uncompressed_bytes:
        if counter_name:
            bump_counter(counter_name)
        raise ZipBombError(
            f"archive {filename!r} declares {total} uncompressed bytes; "
            f"exceeds max_uncompressed_bytes={max_uncompressed_bytes}"
        )