110 lines
5.4 KiB
Python
110 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Check supplied LUKS credentials without mounting, decrypting, or logging secrets."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
|
|
HEADER_LIMIT = 16 * 1024 * 1024
|
|
|
|
def credentials(settings):
|
|
if bool(settings.get('key_file')) == bool(settings.get('key_env')):
|
|
raise ValueError('configure exactly one key_file or key_env reference')
|
|
if settings.get('key_file'):
|
|
with Path(settings['key_file']).open('rb') as stream:
|
|
raw = stream.read(65537)
|
|
else:
|
|
raw = os.environ[settings['key_env']].encode()
|
|
if not raw or len(raw) > 65536:
|
|
raise ValueError('credential must contain 1 to 65536 bytes')
|
|
variants = {'raw': raw, 'text-stripped': raw.strip()}
|
|
tokens = raw.split()
|
|
if tokens and all(re.fullmatch(rb'0x[0-9a-fA-F]{8}', t) for t in tokens):
|
|
words = [int(t, 16) for t in tokens]
|
|
variants.update({'u32le': struct.pack('<' + 'I' * len(words), *words),
|
|
'u32be': struct.pack('>' + 'I' * len(words), *words),
|
|
'hex-text': ''.join(f'{w:08x}' for w in words).encode()})
|
|
# The signed SPIKE 3 SD boot init reads eight OTP words with
|
|
# vcmailbox tag 0x00030021. A saved response includes seven leading
|
|
# protocol words and a terminator, which are not part of the key.
|
|
if (len(words)==16 and words[:7]==[64,0x80000000,0x00030021,40,0x80000028,0,8]
|
|
and words[15]==0):
|
|
variants['vcmailbox']=struct.pack('>8I',*words[7:15])
|
|
encoding = settings.get('key_encoding', 'raw')
|
|
if encoding == 'auto':
|
|
return variants
|
|
if encoding not in variants:
|
|
raise ValueError('unsupported credential encoding for supplied file')
|
|
return {encoding: variants[encoding]}
|
|
|
|
def check_stream(stream, settings, keys, work):
|
|
header = stream.read(HEADER_LIMIT)
|
|
if not header.startswith(b'LUKS\xba\xbe\x00\x02'):
|
|
return None
|
|
uuid = header[168:208].split(b'\0')[0].decode('ascii', errors='replace')
|
|
path = work / 'header.luks'
|
|
path.write_bytes(header)
|
|
attempts = {}
|
|
for encoding, secret in keys.items():
|
|
# Credentials use stdin, never argv or a workspace file. Tool output is
|
|
# captured and reduced to fixed statuses; it cannot become provenance.
|
|
result = subprocess.run([settings.get('cryptsetup', '/usr/sbin/cryptsetup'),
|
|
'open', '--test-passphrase', '--key-file', '-', str(path)],
|
|
input=secret, capture_output=True, timeout=120)
|
|
attempts[encoding] = ('accepted' if result.returncode == 0 else
|
|
'rejected' if b'No key available with this passphrase' in result.stderr else 'tool_error')
|
|
if result.returncode == 0:
|
|
break
|
|
return {'uuid': uuid, 'unlock_verified': 'accepted' in attempts.values(), 'attempts': attempts}
|
|
|
|
def run(request):
|
|
if request['protocol'] != 1:
|
|
raise ValueError('unsupported protocol')
|
|
settings = request['settings']
|
|
keys = credentials(settings)
|
|
rows = []
|
|
with tempfile.TemporaryDirectory(prefix='luks-header-', dir=Path(request['output_dir']).parent) as directory:
|
|
work = Path(directory)
|
|
for path in sorted(Path(request['input_dir']).rglob('*')):
|
|
if not path.is_file() or path.is_symlink():
|
|
continue
|
|
rel = path.relative_to(request['input_dir']).as_posix()
|
|
if zipfile.is_zipfile(path):
|
|
with zipfile.ZipFile(path) as archive:
|
|
for entry in archive.infolist():
|
|
if entry.is_dir() or not (entry.filename.endswith('.000') or entry.filename.lower().endswith('.spk')):
|
|
continue
|
|
with archive.open(entry) as stream:
|
|
report = check_stream(stream, settings, keys, work)
|
|
if report:
|
|
rows.append({'path': rel, 'entry': entry.filename, **report})
|
|
else:
|
|
with path.open('rb') as stream:
|
|
report = check_stream(stream, settings, keys, work)
|
|
if report:
|
|
rows.append({'path': rel, **report})
|
|
if not rows:
|
|
raise ValueError('no supported LUKS2 header found')
|
|
record = {'schema': 1, 'operation': 'credential-validation', 'inputs': rows,
|
|
'content_extracted': False, 'credential_reference': settings.get('key_file') or settings.get('key_env')}
|
|
Path(request['output_dir'], 'luks-key-check.json').write_text(json.dumps(record, indent=2))
|
|
warnings = ['Credential validation only. No decrypted content has been extracted.']
|
|
if any(not row['unlock_verified'] for row in rows):
|
|
warnings.append('One or more containers did not unlock with the supplied credential. Inspect attempts; tool_error is distinct from key rejection.')
|
|
return {'protocol': 1, 'layer': 'derived', 'coverage': 'partial', 'files': ['luks-key-check.json'], 'warnings': warnings}
|
|
|
|
if __name__ == '__main__':
|
|
request = json.loads(Path(sys.argv[1]).read_text())
|
|
try:
|
|
result = run(request)
|
|
except Exception:
|
|
# Avoid serializing exceptions that could contain credentials or tool output.
|
|
Path(request['result_file']).write_text(json.dumps({'protocol': 1, 'error': 'LUKS credential check failed; check key reference, dependencies, and supported encoding.'}))
|
|
sys.exit(1)
|
|
Path(request['result_file']).write_text(json.dumps(result))
|