Anatomi Serangan pada Ingestion Pipeline AI

Pipeline inferensi video generatif seperti model berbasis diffusion atau transformer membutuhkan alokasi VRAM besar. Penyerang mengeksploitasi karakteristik ini via tiga vektor utama: file polyglot (menyamarkan skrip executable dalam header media), decompression bomb / pixel flood (file terkompresi kecil yang mengembang masif di memory saat decode), dan exhaustion DoS (membanjiri antrean inferensi GPU).

Mengandalkan header Content-Type dari client berisiko tinggi karena header tersebut mudah dipalsukan. Sanitasi wajib dilakukan di boundary layer API sebelum payload menyentuh storage, antrean worker, atau decoding pipeline PyTorch/TorchVision.

1. Validasi MIME via Magic Byte (Binary Signatures)

Pemeriksaan signature biner memeriksa deretan byte awal file untuk memastikan integritas format tanpa membaca seluruh file ke RAM.

# signature_validator.py
from typing import Tuple

# ISO Base Media File Format (MP4) signatures pada offset 4
MP4_FTYP_SIGNATURES = {
    b"isom", b"iso2", b"mp41", b"mp42", 
    b"avc1", b"qt  ", b"MSNV", b"dash"
}

def validate_mp4_magic_bytes(header: bytes) -> Tuple[bool, str]:
    if len(header) < 12:
        return False, "Header payload terlalu pendek"
    
    # MP4 box structure: 4 bytes size + 4 bytes 'ftyp' + major brand
    if header[4:8] != b"ftyp":
        return False, "Signature ftyp box tidak ditemukan"
        
    major_brand = header[8:12]
    if major_brand not in MP4_FTYP_SIGNATURES:
        return False, f"Brand format {major_brand!r} tidak didukung"
        
    return True, "Format valid"

2. Mitigasi Decompression Bomb Tanpa Pemuatan GPU

Mendekode video utuh langsung ke tensor GPU menyebabkan out-of-memory (OOM) crash jika resolusi atau frame rate dimanipulasi secara ekstrem. Gunakan parser metadata level-container atau reader header stream minimal untuk memeriksa dimensi sebelum alokasi buffer.

# stream_guard.py
import struct
from io import BytesIO

class VideoDimensionBombError(Exception):
    pass

def inspect_mp4_dimensions(stream: BytesIO, max_width=3840, max_height=2160) -> Tuple[int, int]:
    # ponytail: traversal parser parser minimal; fallback ke ffprobe/av.open jika atom fragmented
    stream.seek(0)
    data = stream.read(8192)
    
    tkhd_idx = data.find(b"tkhd")
    if tkhd_idx == -1:
        # Menolak stream jika container tidak standar untuk mencegah blind decoding
        raise VideoDimensionBombError("Metadata tkhd atom tidak ditemukan di chunk awal")
    
    version = data[tkhd_idx + 4]
    offset = tkhd_idx + 8 + (80 if version == 1 else 72)
    
    if len(data) < offset + 8:
        raise VideoDimensionBombError("Stream terpotong saat parsing dimensi")
        
    width_fixed = struct.unpack(">I", data[offset:offset+4])[0]
    height_fixed = struct.unpack(">I", data[offset+4:offset+8])[0]
    
    width = width_fixed >> 16
    height = height_fixed >> 16
    
    if width > max_width or height > max_height:
        raise VideoDimensionBombError(f"Dimensi {width}x{height} melebihi batas {max_width}x{max_height}")
        
    return width, height

3. Token Bucket Rate Limiting per Identitas

Worker inferensi memiliki throughput tetap. Token bucket memastikan burst traffic diakomodasi tanpa menenggelamkan kapasitas compute model worker.

# rate_limiter.py
import time
from collections import defaultdict
from fastapi import HTTPException, status

class MemoryTokenBucket:
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate  # token per detik
        self.tokens = defaultdict(lambda: capacity)
        self.last_update = defaultdict(time.monotonic)

    def consume(self, key: str, tokens_needed: int = 1) -> bool:
        now = time.monotonic()
        elapsed = now - self.last_update[key]
        self.last_update[key] = now
        
        # Refill
        self.tokens[key] = min(
            self.capacity, 
            self.tokens[key] + elapsed * self.refill_rate
        )
        
        if self.tokens[key] >= tokens_needed:
            self.tokens[key] -= tokens_needed
            return True
        return False

# 10 burst capacity, 0.5 request/detik (1 req per 2 detik)
limiter = MemoryTokenBucket(capacity=10, refill_rate=0.5)

def check_rate_limit(client_id: str):
    if not limiter.consume(client_id):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded. GPU inference cluster saturated."
        )

4. Implementasi Terintegrasi pada FastAPI Endpoint

Kode endpoint berikut menyatukan rate limiting, chunked streaming ingestion, validasi magic bytes, dan pemeriksaan batas ukuran stream.

# main.py
from fastapi import FastAPI, UploadFile, File, Header, HTTPException, Depends
from io import BytesIO
from signature_validator import validate_mp4_magic_bytes
from stream_guard import inspect_mp4_dimensions, VideoDimensionBombError
from rate_limiter import check_rate_limit

app = FastAPI()

MAX_PAYLOAD_SIZE = 50 * 1024 * 1024  # 50 MB

@app.post("/v1/inference/video-upload")
async def upload_for_inference(
    file: UploadFile = File(...),
    x_api_key: str = Header(..., alias="X-API-Key")
):
    check_rate_limit(x_api_key)
    
    header_chunk = await file.read(12)
    is_valid, reason = validate_mp4_magic_bytes(header_chunk)
    if not is_valid:
        raise HTTPException(status_code=400, detail=f"Invalid format: {reason}")
        
    stream_buffer = BytesIO()
    stream_buffer.write(header_chunk)
    
    total_bytes = len(header_chunk)
    chunk_size = 64 * 1024  # 64 KB chunks
    
    while True:
        chunk = await file.read(chunk_size)
        if not chunk:
            break
        total_bytes += len(chunk)
        if total_bytes > MAX_PAYLOAD_SIZE:
            raise HTTPException(status_code=413, detail="Ukuran file melebihi batas 50MB")
        stream_buffer.write(chunk)

    try:
        width, height = inspect_mp4_dimensions(stream_buffer)
    except VideoDimensionBombError as e:
        raise HTTPException(status_code=422, detail=str(e))

    # stream_buffer kini aman diteruskan ke object storage atau Celery worker
    return {
        "status": "queued",
        "filename": file.filename,
        "dimensions": {"width": width, "height": height},
        "size_bytes": total_bytes
    }

5. Unit Test: Verifikasi Payload Malicious

Uji implementasi menggunakan test case untuk file rusak dan polyglot exploit.

# test_upload_security.py
import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_reject_fake_mime_type():
    # File teks biasa disamarkan sebagai video/mp4 via header HTTP
    payload = b"MALICIOUS_PAYLOAD_SHELLCODE_HERE"
    files = {"file": ("exploit.mp4", payload, "video/mp4")}
    headers = {"X-API-Key": "tenant-prod-01"}
    
    response = client.post("/v1/inference/video-upload", files=files, headers=headers)
    assert response.status_code == 400
    assert "Invalid format" in response.json()["detail"]

def test_reject_pixel_flood_metadata():
    # ftyp box valid, diikuti track header (tkhd) dengan dimensi 65535x65535
    fake_header = b"\x00\x00\x00\x14ftypmp42" + b"\x00" * 8
    fake_tkhd = (
        b"tkhd" + 
        b"\x00" +              # version 0
        b"\x00" * 79 +         # flags & other properties
        b"\xFF\xFF\x00\x00" + # Width: 65535
        b"\xFF\xFF\x00\x00"   # Height: 65535
    )
    payload = fake_header + fake_tkhd
    
    files = {"file": ("bomb.mp4", payload, "video/mp4")}
    headers = {"X-API-Key": "tenant-prod-01"}
    
    response = client.post("/v1/inference/video-upload", files=files, headers=headers)
    assert response.status_code == 422
    assert "melebihi batas" in response.json()["detail"]

def test_rate_limit_saturation():
    headers = {"X-API-Key": "tenant-abuse-01"}
    files = {"file": ("test.mp4", b"\x00\x00\x00\x14ftypisom" + b"\x00"*20, "video/mp4")}
    
    # Menghabiskan 10 token burst
    for _ in range(10):
        client.post("/v1/inference/video-upload", files=files, headers=headers)
        
    # Request ke-11 wajib ditolak
    res = client.post("/v1/inference/video-upload", files=files, headers=headers)
    assert res.status_code == 429

Trade-offs dan Batasan Desain

  • In-Memory Token Bucket vs Redis: Implementasi MemoryTokenBucket di atas terbatas pada instance tunggal. Untuk multi-worker deployment (Gunicorn/Uvicorn workers > 1), gantikan dictionary dengan Redis instance menggunakan atomic script LUA.
  • Container Parsing vs Deep Frame Inspection: Pembacaan byte atomik tkhd cepat (<1ms), namun hanya memeriksa header container. Jika penyerang menyuntikkan corrupted NAL units pada stream H.264/H.265 di tengah file, library decoder GPU tetap dapat mengalami segmentasi fault. Wajib sandbox worker decoder menggunakan proses terisolasi tanpa hak akses root.