50 lines
2.9 KiB
Python
50 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Bounded structural probe; records missing credentials without reading key values."""
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import zipfile
|
|
|
|
def classify(header):
|
|
if header.startswith(b"LUKS\xba\xbe\x00\x02"):
|
|
uuid = header[168:208].split(b"\0")[0].decode("ascii", errors="replace")
|
|
return {"format": "luks2", "generation": "unknown", "uuid": uuid,
|
|
"required_secret": "stern.spike3.luks", "next_step": "Provide a LUKS unlocking keyfile; encrypted-container decoding is a separate pending adapter."}
|
|
if header.startswith(b"hsqs"):
|
|
return {"format": "squashfs", "generation": "unknown", "required_secret": None, "next_step": "Assemble all split parts and extract the inner SPK."}
|
|
if header.startswith(b"SPKS"):
|
|
return {"format": "spk", "generation": "unknown", "required_secret": None, "next_step": "Run spike-extract."}
|
|
if header.startswith(b"\x1f\x8b"):
|
|
return {"format": "gzip", "generation": "unknown", "required_secret": None, "next_step": "Decode the legacy wrapper; do not pass it directly to the SPKS parser."}
|
|
return {"format": "unknown", "generation": "unknown", "required_secret": None, "next_step": "Preserve as opaque content."}
|
|
|
|
def probe(path):
|
|
if zipfile.is_zipfile(path):
|
|
rows = []
|
|
with zipfile.ZipFile(path) as z:
|
|
for entry in z.infolist():
|
|
if entry.is_dir() or not (entry.filename.lower().endswith(".spk") or entry.filename.endswith(".000")):
|
|
continue
|
|
if entry.flag_bits & 1:
|
|
rows.append({"entry": entry.filename, "format": "encrypted_zip", "required_secret": "zip.password"})
|
|
else:
|
|
with z.open(entry) as stream:
|
|
rows.append({"entry": entry.filename, **classify(stream.read(4096))})
|
|
return {"format": "zip", "entries": rows}
|
|
with path.open("rb") as stream:
|
|
return classify(stream.read(4096))
|
|
|
|
def run(request):
|
|
if request["protocol"] != 1:
|
|
raise ValueError("unsupported protocol")
|
|
source, output = Path(request["input_dir"]), Path(request["output_dir"])
|
|
rows = [{"path": p.relative_to(source).as_posix(), **probe(p)} for p in sorted(source.rglob("*")) if p.is_file() and not p.is_symlink()]
|
|
(output / "format-report.json").write_text(json.dumps({"schema": 1, "inputs": rows}, indent=2))
|
|
missing = [r["path"] for r in rows if r.get("required_secret") or any(e.get("required_secret") for e in r.get("entries", []))]
|
|
return {"protocol": 1, "layer": "derived", "coverage": "partial", "files": ["format-report.json"],
|
|
"warnings": ["Required credentials: " + ", ".join(missing)] if missing else ["Structural probe only; generation is not inferred from container format."]}
|
|
|
|
if __name__ == "__main__":
|
|
request = json.loads(Path(sys.argv[1]).read_text())
|
|
Path(request["result_file"]).write_text(json.dumps(run(request)))
|