199 lines
9.8 KiB
Python
199 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only LUKS2/AES-XTS/ext4 wrapper extraction, without kernel mounts."""
|
|
import ctypes as C
|
|
import ctypes.util
|
|
from collections import deque
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
import json
|
|
from multiprocessing import get_context
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
from progress import report
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
|
|
|
|
def volume_key(image, credential):
|
|
lib = C.CDLL(ctypes.util.find_library('cryptsetup') or 'libcryptsetup.so.12')
|
|
lib.crypt_init.argtypes = [C.POINTER(C.c_void_p), C.c_char_p]
|
|
lib.crypt_load.argtypes = [C.c_void_p, C.c_char_p, C.c_void_p]
|
|
lib.crypt_volume_key_get.argtypes = [C.c_void_p, C.c_int, C.c_void_p, C.POINTER(C.c_size_t), C.c_char_p, C.c_size_t]
|
|
lib.crypt_free.argtypes = [C.c_void_p]
|
|
lib.crypt_free.restype = None
|
|
device = C.c_void_p()
|
|
buffer = C.create_string_buffer(64)
|
|
length = C.c_size_t(64)
|
|
try:
|
|
if lib.crypt_init(C.byref(device), os.fsencode(image)) < 0 or lib.crypt_load(device, b'LUKS2', None) < 0:
|
|
raise ValueError('cannot load LUKS2 metadata')
|
|
if lib.crypt_volume_key_get(device, -1, buffer, C.byref(length), credential, len(credential)) < 0:
|
|
raise ValueError('credential did not unlock this container')
|
|
if length.value not in (32, 64):
|
|
raise ValueError('unsupported volume key size')
|
|
return buffer.raw[:length.value]
|
|
finally:
|
|
C.memset(buffer, 0, len(buffer))
|
|
if device:
|
|
lib.crypt_free(device)
|
|
|
|
|
|
def decrypt_block(block, position, key, sector, iv):
|
|
plain = bytearray()
|
|
for start in range(0, len(block), sector):
|
|
# plain64 uses a 512-byte sector number, even for 4K data units.
|
|
tweak = (iv + (position + start) // 512).to_bytes(16, 'little')
|
|
context = Cipher(algorithms.AES(key), modes.XTS(tweak)).decryptor()
|
|
plain.extend(context.update(block[start:start + sector]) + context.finalize())
|
|
return plain
|
|
|
|
|
|
def decrypt_blocks(source, key, sector, iv, workers):
|
|
chunk = 4 * 1024 * 1024
|
|
if workers == 1:
|
|
position = 0
|
|
while block := source.read(chunk):
|
|
yield decrypt_block(block, position, key, sector, iv)
|
|
position += len(block)
|
|
return
|
|
# Bound both queued ciphertext and completed plaintext. Credentials stay
|
|
# in process memory; workers never write files or report key material.
|
|
with ProcessPoolExecutor(max_workers=workers, mp_context=get_context('spawn')) as pool:
|
|
pending = deque();position = 0
|
|
while True:
|
|
while len(pending) < workers * 2:
|
|
block = source.read(chunk)
|
|
if not block:break
|
|
pending.append(pool.submit(decrypt_block, block, position, key, sector, iv))
|
|
position += len(block)
|
|
if not pending:break
|
|
yield pending.popleft().result()
|
|
|
|
|
|
def decrypt(image, output, key, metadata, workers=1):
|
|
if not isinstance(workers, int) or not 1 <= workers <= 16:
|
|
raise ValueError('invalid decryption worker count')
|
|
if list(metadata['segments']) != ['0']:
|
|
raise ValueError('multiple/re-encrypting segments are unsupported')
|
|
segment = metadata['segments']['0']
|
|
if segment['type'] != 'crypt' or segment['encryption'] != 'aes-xts-plain64' or segment.get('flags'):
|
|
raise ValueError('unsupported encryption segment')
|
|
sector = segment['sector_size']
|
|
if sector not in (512, 4096):
|
|
raise ValueError('unsupported sector size')
|
|
offset = int(segment['offset'])
|
|
length = image.stat().st_size - offset
|
|
if length <= 0 or length % sector or offset % sector:
|
|
raise ValueError('truncated or misaligned container')
|
|
if segment['size'] != 'dynamic' and int(segment['size']) != length:
|
|
raise ValueError('unsupported fixed segment length')
|
|
iv = int(segment['iv_tweak'])
|
|
with image.open('rb') as source, output.open('wb') as dest:
|
|
source.seek(offset)
|
|
position = 0
|
|
for plain in decrypt_blocks(source, key, sector, iv, workers):
|
|
if plain.count(0)==len(plain):dest.seek(len(plain),1)
|
|
else:dest.write(plain)
|
|
position += len(plain)
|
|
report('Decrypting filesystem',position,length,'bytes',image.name)
|
|
dest.truncate(position)
|
|
with output.open('rb') as stream:
|
|
stream.seek(1024 + 56)
|
|
if stream.read(2) != b'\x53\xef':
|
|
raise ValueError('decryption did not produce an ext filesystem')
|
|
|
|
|
|
def assemble(path, destination, budget):
|
|
"""Strict complete split ordering; ZIP entry names never become output paths."""
|
|
if zipfile.is_zipfile(path):
|
|
with zipfile.ZipFile(path) as archive:
|
|
entries = [e for e in archive.infolist() if not e.is_dir()]
|
|
names = [e.filename for e in entries]
|
|
if len(names) != len(set(names)):
|
|
raise ValueError('duplicate ZIP entry')
|
|
matches = [re.fullmatch(r'(.+\.spk)\.(\d{3})\.(\d{3})', name, re.I) for name in names]
|
|
if not entries or not all(matches):
|
|
raise ValueError('expected only split SPK entries')
|
|
count = int(matches[0][2])
|
|
if len(entries) != count or len({(m[1],m[2]) for m in matches}) != 1 or sorted(int(m[3]) for m in matches) != list(range(count)):
|
|
raise ValueError('missing or inconsistent split parts')
|
|
total = sum(e.file_size for e in entries)
|
|
if total * 4 > budget:
|
|
raise ValueError('insufficient declared workspace for extraction')
|
|
with destination.open('wb') as out:
|
|
for entry in sorted(entries, key=lambda e: int(e.filename.rsplit('.',1)[1])):
|
|
with archive.open(entry) as stream:
|
|
shutil.copyfileobj(stream, out, 1024 * 1024)
|
|
return total
|
|
size = path.stat().st_size
|
|
if size * 4 > budget:
|
|
raise ValueError('insufficient declared workspace for extraction')
|
|
shutil.copyfile(path, destination)
|
|
return size
|
|
|
|
|
|
def run(request):
|
|
if request['protocol'] != 1:
|
|
raise ValueError('unsupported protocol')
|
|
settings = request['settings']
|
|
source = Path(request['input_dir'])
|
|
path = source / settings['input_path']
|
|
if not path.resolve().is_relative_to(source.resolve()) or not path.is_file():
|
|
raise ValueError('input_path must select a file within the input snapshot')
|
|
if bool(settings.get('key_file')) == bool(settings.get('key_env')):
|
|
raise ValueError('configure exactly one credential reference')
|
|
if settings.get('key_file'):
|
|
with Path(settings['key_file']).open('rb') as f:
|
|
credential = f.read(65537)
|
|
else:
|
|
credential = os.environ[settings['key_env']].encode()
|
|
if not credential or len(credential) > 65536:
|
|
raise ValueError('credential size unsupported')
|
|
output = Path(request['output_dir'])
|
|
with tempfile.TemporaryDirectory(prefix='luks-', dir=output.parent) as tmp:
|
|
work = Path(tmp)
|
|
encrypted, plain = work / 'container.luks', work / 'filesystem.ext4'
|
|
size = assemble(path, encrypted, request['workspace_bytes'])
|
|
metadata_process = subprocess.run([settings.get('cryptsetup','/usr/sbin/cryptsetup'),'luksDump','--dump-json-metadata',str(encrypted)],capture_output=True,check=True)
|
|
metadata = json.loads(metadata_process.stdout)
|
|
key = volume_key(encrypted, credential)
|
|
del credential
|
|
decrypt(encrypted, plain, key, metadata)
|
|
del key
|
|
destination = output / 'filesystem'
|
|
destination.mkdir()
|
|
if any(c in str(destination) for c in ['"','\\','\n','\r']):
|
|
raise ValueError('workspace path cannot be represented to debugfs')
|
|
process = subprocess.run([settings.get('debugfs','/usr/sbin/debugfs'),'-R',f'rdump / "{destination}"',str(plain)],capture_output=True)
|
|
if process.returncode != 0 or any(s in process.stderr.lower() for s in [b'error',b'failed',b'cannot',b'not found',b'short read']):
|
|
raise ValueError('ext4 extraction reported an error')
|
|
files = [p for p in destination.rglob('*') if p.is_file() and not p.is_symlink()]
|
|
if not any(p.suffix.lower()=='.spk' for p in files):
|
|
raise ValueError('no inner SPK was recovered')
|
|
links=[]
|
|
for p in destination.rglob('*'):
|
|
if p.is_symlink():
|
|
links.append({'path':p.relative_to(output).as_posix(),'target':str(p.readlink())});p.unlink()
|
|
evidence={'schema':1,'input_path':settings['input_path'],'container_bytes':size,'unlock_verified':True,
|
|
'credential_reference':settings.get('key_file') or settings.get('key_env'),
|
|
'encryption':metadata['segments']['0']['encryption'],'sector_size':metadata['segments']['0']['sector_size'],
|
|
'symlinks':links,'filesystem':'ext4','method':'libcryptsetup + userspace AES-XTS + read-only debugfs'}
|
|
(output/'wrapper-evidence.json').write_text(json.dumps(evidence,indent=2))
|
|
return {'protocol':1,'layer':'extracted','coverage':'partial','warnings':['Encrypted original retained. Inner SPK still requires extraction; filesystem metadata is not promised byte-for-byte restoration.'],
|
|
'files':[p.relative_to(output).as_posix() for p in sorted(output.rglob('*')) if p.is_file()]}
|
|
|
|
if __name__ == '__main__':
|
|
request=json.loads(Path(sys.argv[1]).read_text())
|
|
try:
|
|
result=run(request)
|
|
except Exception as error:
|
|
# Fixed messages from this adapter only; never serialize raw tool output.
|
|
message=str(error) if isinstance(error,ValueError) else 'LUKS extraction failed; check dependencies, source completeness, and workspace capacity.'
|
|
Path(request['result_file']).write_text(json.dumps({'protocol':1,'error':message}))
|
|
sys.exit(1)
|
|
Path(request['result_file']).write_text(json.dumps(result))
|