Optional entropy / Vault
Drop or inspect an artifact to see its parsed structure.
Decrypted results appear here after Decrypt.
#!/usr/bin/env python3 """Web front end for the offline DPAPI toolkit (standard library only). Serves a single drag-and-drop page on loopback only. Dropped files are sent to the server as base64 JSON (no framework, no multipart), written to a short-lived temp directory, and fed to the same ``dpapi_toolkit`` core the CLI uses. python3 dpapi_web.py # http://127.0.0.1:8765/ python3 dpapi_web.py --port 9000 --no-open Security notes: - The server binds to 127.0.0.1 only. It cannot be bound to a LAN or public interface; each analyst runs a separate copy on their own machine. - Host and Origin values are checked against loopback, and a per-run request token rejects cross-origin mutations. This is browser-request hardening, not user authentication and not an isolation boundary against another process running as the same OS user. - Decrypted secrets pass through this process. Stop it when finished. """ from __future__ import annotations import argparse import base64 import binascii import contextlib import html import io import ipaddress import json import secrets import shutil import tempfile import threading import time import webbrowser import zipfile from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlsplit import dpapi_toolkit as dt import dpapi_plugins # A web request must never block on an interactive password prompt. def _no_getpass(*_args, **_kwargs): raise ValueError( "a password is required: fill the Password field (or tick 'empty " "password'), or supply other key material" ) dt.getpass.getpass = _no_getpass MAX_BODY = 96 * 1024 * 1024 MAX_UPLOAD_FILE = 64 * 1024 * 1024 MAX_UPLOAD_TOTAL = 64 * 1024 * 1024 MAX_UPLOAD_FILES = 4096 MAX_DOWNLOAD_BYTES = 128 * 1024 * 1024 MAX_DOWNLOADS = 64 MAX_PREVIEW_BYTES = 256 * 1024 DOWNLOAD_TTL_SECONDS = 10 * 60 REQUEST_TIMEOUT_SECONDS = 20 LOCAL_HOSTS = frozenset(("127.0.0.1", "localhost", "::1")) PAGE_PATH = "/" MEMORY_TEMP_ROOT = Path("/dev/shm") # Field names that carry a dropped single file (value -> temp path). FILE_FIELDS = ( "input", "masterkey", "real_masterkey", "dpapi_system", "domain_backup_key", "system_masterkey", "entropy_file", "vault_policy", "certificate", "pfx_password_file", "pvk_password_file", "dpapi_ng_root_key", "vault_key", "security_hive", "sam_hive", ) # Free-text fields passed straight to the config (CLI dest names). TEXT_FIELDS = ( "sid", "machine_sid", "password", "nt_hash", "sha1_hash", "prekey", "credkey", "entropy", "pfx_password", "pvk_password", "key_password", "pin", ) SELECT_FIELDS = ("type", "plugin", "output_format", "hashcat_context") MATERIAL_TEXT_FIELDS = frozenset(("nt_hash", "sha1_hash", "prekey", "credkey")) SERVER_STORAGE_FIELDS = frozenset(("autosave", "out_dir", "out_file")) @dataclass class UploadBudget: files: int = 0 total: int = 0 def decode(self, payload: dict, label: str) -> bytes: if not isinstance(payload, dict) or not isinstance(payload.get("b64"), str): raise ValueError(f"{label} is not a valid uploaded file") encoded = payload["b64"] if len(encoded) > ((MAX_UPLOAD_FILE + 2) // 3) * 4 + 4: raise ValueError(f"{label} exceeds the {MAX_UPLOAD_FILE // 1048576} MiB file limit") try: data = base64.b64decode(encoded, validate=True) except (binascii.Error, ValueError): raise ValueError(f"{label} contains invalid Base64") from None if len(data) > MAX_UPLOAD_FILE: raise ValueError(f"{label} exceeds the {MAX_UPLOAD_FILE // 1048576} MiB file limit") self.files += 1 self.total += len(data) if self.files > MAX_UPLOAD_FILES: raise ValueError(f"upload contains more than {MAX_UPLOAD_FILES} files") if self.total > MAX_UPLOAD_TOTAL: raise ValueError(f"decoded uploads exceed {MAX_UPLOAD_TOTAL // 1048576} MiB") return data @dataclass class DownloadEntry: name: str data: bytes expires_at: float # Short-lived, single-use decrypted outputs awaiting download. _DOWNLOADS: dict[str, DownloadEntry] = {} _DOWNLOADS_LOCK = threading.Lock() _WORK_SLOTS = threading.BoundedSemaphore(2) def _safe_upload_path(root: Path, relpath: str) -> Path: if not isinstance(relpath, str) or not relpath or "\x00" in relpath: raise ValueError("uploaded path is empty or invalid") rel = Path(relpath) if ( rel.is_absolute() or rel.anchor or rel.drive or any(part in ("", ".", "..") for part in rel.parts) ): raise ValueError(f"unsafe uploaded path: {relpath!r}") resolved_root = root.resolve() target = (resolved_root / rel).resolve() try: target.relative_to(resolved_root) except ValueError: raise ValueError(f"uploaded path escapes its temporary directory: {relpath!r}") from None return target def _safe_download_name(name: str) -> str: cleaned = "".join( character if character.isascii() and (character.isalnum() or character in "._-") else "_" for character in name ).strip("._") return cleaned[:180] or "dpapi-output.bin" def _prune_downloads_locked(now: float | None = None) -> None: current = time.monotonic() if now is None else now for key in [key for key, entry in _DOWNLOADS.items() if entry.expires_at <= current]: del _DOWNLOADS[key] def _expire_download(token: str) -> None: """Remove an unused decrypted download from process memory immediately.""" with _DOWNLOADS_LOCK: _DOWNLOADS.pop(token, None) def _download_reaper(stop: threading.Event) -> None: """Bound expired decrypted data in memory even while the server is idle.""" interval = 1 while not stop.wait(interval): with _DOWNLOADS_LOCK: _prune_downloads_locked() def _register_download(name: str, data: bytes) -> str: if len(data) > MAX_DOWNLOAD_BYTES: raise ValueError(f"output exceeds the {MAX_DOWNLOAD_BYTES // 1048576} MiB download limit") token = secrets.token_urlsafe(24) with _DOWNLOADS_LOCK: _prune_downloads_locked() while _DOWNLOADS and ( len(_DOWNLOADS) >= MAX_DOWNLOADS or sum(len(entry.data) for entry in _DOWNLOADS.values()) + len(data) > MAX_DOWNLOAD_BYTES ): oldest = min(_DOWNLOADS, key=lambda key: _DOWNLOADS[key].expires_at) del _DOWNLOADS[oldest] _DOWNLOADS[token] = DownloadEntry( _safe_download_name(name), data, time.monotonic() + DOWNLOAD_TTL_SECONDS ) return token def _take_download(token: str) -> DownloadEntry | None: with _DOWNLOADS_LOCK: _prune_downloads_locked() return _DOWNLOADS.pop(token, None) def _clear_downloads() -> None: with _DOWNLOADS_LOCK: _DOWNLOADS.clear() def _write_upload(tmp: Path, field: str, payload: dict, budget: UploadBudget) -> str: """Write one uploaded {name,b64} file into tmp and return its path.""" supplied_name = payload.get("name", field) if not isinstance(supplied_name, str) or "\x00" in supplied_name: raise ValueError(f"invalid filename for {field}") name = Path(supplied_name).name or field data = budget.decode(payload, field) target = tmp / field / name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(data) target.chmod(0o600) return str(target) def _write_text_literal(tmp: Path, field: str, value: str) -> str: """Store browser text as literal data, never as a hosting-server path.""" target = tmp / "literals" / f"{field}.txt" target.parent.mkdir(parents=True, exist_ok=True) target.write_text(value, encoding="utf-8") target.chmod(0o600) return str(target) def _write_tree(tmp: Path, field: str, entries: list, budget: UploadBudget) -> str: """Rebuild an uploaded directory tree under tmp and return its root path.""" root = tmp / field root.mkdir(parents=True, exist_ok=True) if not isinstance(entries, list): raise ValueError(f"{field} directory upload is invalid") seen: set[str] = set() for index, entry in enumerate(entries): if not isinstance(entry, dict): raise ValueError(f"{field} directory entry {index} is invalid") safe = _safe_upload_path(root, entry.get("relpath")) relative_key = safe.relative_to(root.resolve()).as_posix().casefold() if relative_key in seen: raise ValueError(f"duplicate uploaded path: {entry.get('relpath')!r}") seen.add(relative_key) data = budget.decode(entry, f"{field}/{entry.get('relpath', index)}") safe.parent.mkdir(parents=True, exist_ok=True) safe.write_bytes(data) safe.chmod(0o600) return str(root) def _new_request_tempdir() -> Path: """Prefer volatile memory storage; fall back to a private removed directory.""" if MEMORY_TEMP_ROOT.is_dir(): try: return Path(tempfile.mkdtemp(prefix="dpapi_web_", dir=MEMORY_TEMP_ROOT)) except OSError: pass return Path(tempfile.mkdtemp(prefix="dpapi_web_")) def _build_config(request: dict, tmp: Path): if not isinstance(request, dict): raise ValueError("request must be a JSON object") files = request.get("files", {}) trees = request.get("trees", {}) fields = request.get("fields", {}) if not all(isinstance(item, dict) for item in (files, trees, fields)): raise ValueError("files, trees, and fields must be JSON objects") requested_storage = sorted( name for name in SERVER_STORAGE_FIELDS if fields.get(name) not in (None, "", False) ) if requested_storage: raise ValueError( "the web interface never writes persistent server-side output: " + ", ".join(requested_storage) ) budget = UploadBudget() input_text = fields.get("input", "") if input_text is not None and not isinstance(input_text, str): raise ValueError("input artifact text must be a string") # Plugin artifacts have their own tab/drop target but intentionally enter # the same core input pipeline after upload validation. input_field = ( files.get("plugin_input") or files.get("input") or (input_text or "").strip() ) if "plugin_input" in trees: # a folder of keys dropped for a plugin input_value = _write_tree(tmp, "plugin_input", trees["plugin_input"], budget) elif isinstance(input_field, dict): input_value = _write_upload(tmp, "input", input_field, budget) elif "input" in trees: # batch directory dropped as the artifact input_value = _write_tree(tmp, "input", trees["input"], budget) elif isinstance(input_field, str) and input_field: input_value = _write_text_literal(tmp, "input", input_field) elif input_field: raise ValueError("input artifact text must be a string") else: raise ValueError("drop or choose an input artifact first") pfx_password_sources = sum(( bool(fields.get("pfx_password")), bool(fields.get("empty_pfx_password")), isinstance(files.get("pfx_password_file"), dict), )) if pfx_password_sources > 1: raise ValueError( "choose exactly one PFX password source: entered password, password file, " "or intentionally unencrypted" ) overrides = {} for field in SELECT_FIELDS: if fields.get(field): overrides[field] = fields[field] # Only operation booleans go through; persistent web output is forbidden above. for field in ("hashcat", "cachedata_hashcat", "batch"): overrides[field] = bool(fields.get(field)) for field in FILE_FIELDS: if field == "input": continue raw_value = fields.get(field, "") if raw_value is not None and not isinstance(raw_value, str): raise ValueError(f"{field} must be uploaded or supplied as text") if isinstance(files.get(field), dict): overrides[field] = _write_upload(tmp, field, files[field], budget) elif (raw_value or "").strip(): overrides[field] = _write_text_literal(tmp, field, raw_value.strip()) if "masterkey_dir" in trees: overrides["masterkey_dir"] = _write_tree( tmp, "masterkey_dir", trees["masterkey_dir"], budget ) elif fields.get("masterkey_dir") not in (None, ""): raise ValueError("masterkey directories must be uploaded; server paths are disabled") # The certificate/PFX plugin accepts a dropped folder of certificates and # matches by public key, so a certificate tree overrides a single cert file. if "certificate" in trees: overrides["certificate"] = _write_tree( tmp, "certificate", trees["certificate"], budget ) for field in TEXT_FIELDS: raw_value = fields.get(field) if raw_value is not None and not isinstance(raw_value, str): raise ValueError(f"{field} must be text") value = ( (raw_value or "") if field in ("password", "entropy", "pfx_password", "pvk_password", "key_password", "pin") else (raw_value or "").strip() ) if value: overrides[field] = ( _write_text_literal(tmp, field, value) if field in MATERIAL_TEXT_FIELDS else value ) # Password: blank -> None (keeps DPAPI_SYSTEM auto-try working); explicit # empty password only when ticked. if bool(fields.get("empty_password")): overrides["password"] = "" if bool(fields.get("empty_pfx_password")): overrides["pfx_password"] = "" overrides["show"] = False return dt.make_config(input_value, **overrides) def _run(request: dict, inspect_only: bool) -> dict: tmp = _new_request_tempdir() log: list[str] = [] emit = log.append results = [] structure: list = [] raw_hex = "" detected_type = None note = None ok = False try: cfg = _build_config(request, tmp) dt.validate_config(cfg, emit=emit) # Structure/raw preview for any single (non-batch) artifact, so the # Structure and Raw tabs fill whether the user inspects or decrypts. # A plugin folder input (e.g. a keys directory) has no single artifact # to preview, so skip it there. info = {"kind": "none"} if not cfg.batch and not Path(cfg.input).is_dir(): data, _, _ = dt.read_data(cfg.input, "input") raw_hex = data[:65536].hex() info = dt.describe_input(data, cfg.type) structure = info["structure"] detected_type = info["detected_type"] note = info["note"] if cfg.batch and cfg.hashcat and not inspect_only: # Export a $DPAPImk$ record for every master key in the dropped folder, # one download per Hashcat mode. Each key is cracked independently. for item in dt.run_batch_hashcat(cfg, emit=emit): results.append(_store_output(item)) elif cfg.batch and not inspect_only: # Batch output is always short-lived and returned to the browser as # one in-memory zip. The web API never accepts a hosting-server path. cfg.out_dir = str(tmp / "batch_out") buffer = io.StringIO() with contextlib.redirect_stdout(buffer): dt.run_batch(cfg) log.extend(buffer.getvalue().splitlines()) zip_result = _zip_batch_output(Path(cfg.out_dir), log) if zip_result: results.append(zip_result) elif inspect_only: if cfg.batch: emit("[i] batch mode: Decrypt processes all artifacts; Masterkey -> Hashcat exports every master key's hash") elif info["kind"] == "masterkey": # An encrypted master key is not a classic blob; don't run the # blob inspector (which would report "no classic DPAPI blob"). emit("[+] input type: encrypted master key") for field in structure[0]["sections"][0]["fields"]: if field["name"] == "Master-key GUID": emit(f"[+] master key {field['value']}") if note: emit(f"[i] {note}") elif info["kind"] == "cert": emit("[+] input type: public certificate") if note: emit(f"[i] {note}") else: for attr in ("masterkey", "real_masterkey", "masterkey_dir"): setattr(cfg, attr, None) dt.run_single(cfg, emit=emit) else: outputs = dt.run_single(cfg, emit=emit) if outputs: emit( f"[+] operation completed successfully: " f"{len(outputs)} output item(s) ready" ) for item in outputs: results.append(_store_output(item)) ok = True except (OSError, ValueError) as error: log.append(f"error: {error}") ok = False finally: try: shutil.rmtree(tmp) except OSError as error: log.append(f"error: could not remove temporary server data: {error}") ok = False return {"ok": ok, "log": log, "results": results, "required": [] if results else dt.required_guids(log), "structure": structure, "raw": raw_hex, "detected_type": detected_type, "note": note} def _zip_batch_output(out_root: Path, log: list) -> dict | None: """Zip a batch run's output folder and register it as a single download.""" buffer = io.BytesIO() files = sorted(p for p in out_root.rglob("*") if p.is_file()) if out_root.exists() else [] if not files: return None with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: for path in files: archive.write(path, path.relative_to(out_root)) data = buffer.getvalue() token = _register_download("batch_results.zip", data) log.append(f"[+] packaged {len(files)} file(s) -> download batch_results.zip") return { "label": "batch results", "extension": ".zip", "size": len(data), "preview": f"{len(files)} file(s) - download the zip to extract", "is_text": False, "download": token, "name": "batch_results.zip", } def _store_output(item) -> dict: preview_data = item.data[:MAX_PREVIEW_BYTES] truncated = len(item.data) > len(preview_data) try: preview = preview_data.decode("utf-8") is_text = True except UnicodeDecodeError: preview = preview_data.hex() is_text = False name = f"{item.label.replace(' ', '_')}{item.extension}" token = _register_download(name, item.data) if truncated: preview += f"\n… preview truncated; download contains all {len(item.data)} bytes" return { "label": item.label, "extension": item.extension, "size": len(item.data), "preview": preview, "is_text": is_text, "download": token, "name": name, } def build_page(token: str, nonce: str) -> bytes: plugin_options = '' plugin_panels = "" for manifest in dpapi_plugins.discover_plugins().values(): plugin_options += ( f'' ) controls = [] for field in manifest.web_fields: name = html.escape(field.name, quote=True) label = html.escape(field.label) help_text = html.escape(field.help, quote=True) title = f' title="{help_text}"' if help_text else "" optional = ' data-optional="true"' if field.optional else "" if field.kind == "file": controls.append( f'
Drop or inspect an artifact to see its parsed structure.
Decrypted results appear here after Decrypt.