SpecterOps-Nemesis/libs/nemesis_dpapi/examples/example.py

"""Example showing DPAPI usage with eventing.

This example demonstrates the complete DPAPI workflow including:

1. Event monitoring setup
   - Creates a custom DpapiObserver to monitor library events
   - Subscribes to events for encrypted masterkeys, domain backup keys, and plaintext masterkeys
   - Shows how to track DPAPI operations in real-time

2. Masterkey management
   - Loads masterkey files and domain backup key
   - Adds multiple encrypted masterkeys to the DpapiManager
   - Demonstrates both fake and real backup key scenarios

3. Domain backup key operations
   - First adds a fake/invalid backup key (shows failed decryption)
   - Then adds the domain backup key
   - Shows automatic decryption of masterkeys when valid backup key is added

4. DPAPI blob decryption
   - Loads an encrypted DPAPI blob
   - Parses the blob to extract its masterkey GUID
   - Decrypts the blob using the previously loaded masterkeys
   - Displays the decrypted plaintext content

"""

import asyncio
import base64
import json
from pathlib import Path
from uuid import UUID

from nemesis_dpapi import Blob, DomainBackupKey, DpapiManager, MasterKey, MasterKeyFile, MasterKeyFilter
from nemesis_dpapi.eventing import (
    DpapiEvent,
    DpapiObserver,
    NewDomainBackupKeyEvent,
    NewEncryptedMasterKeyEvent,
    NewPlaintextMasterKeyEvent,
)


class MyDpapiEventMonitor(DpapiObserver):
    """Class that observes DPAPI events."""

    async def update(self, event: DpapiEvent) -> None:
        """Handler for different types of DPAPI events."""

        if isinstance(event, NewEncryptedMasterKeyEvent):
            print(f"New encrypted masterkey added: {event.masterkey_guid}")
        elif isinstance(event, NewDomainBackupKeyEvent):
            print(f"New domain backup key added: {event.backup_key_guid}")
        elif isinstance(event, NewPlaintextMasterKeyEvent):
            print(f"New plaintext masterkey added: {event.masterkey_guid}")
        else:
            print(f"Event received: {type(event).__name__}")


async def main() -> None:
    """Demonstrate DPAPI library usage with eventing system."""

    # Real/valid DPAPI test data
    fixtures_path = Path(__file__).parent.parent / "tests" / "fixtures"
    backup_key_path = fixtures_path / "backupkey.json"
    masterkey_file_path = fixtures_path / "masterkey_domain.bin"
    blob_path = fixtures_path / "blob_without_entropy.bin"

    # Load and add real masterkey
    with open(backup_key_path) as f:
        backup_key_data = json.load(f)

    masterkey_file = MasterKeyFile.parse(masterkey_file_path)
    backup_key_bytes = base64.b64decode(backup_key_data["key"])

    # Create a fake backup key by corrupting one byte
    pvk_header_size = 20  # PVK_FILE_HDR size
    fake_backup_key_bytes = (
        backup_key_bytes[: pvk_header_size + 32] + b"\xff" + backup_key_bytes[pvk_header_size + 33 :]
    )

    if not masterkey_file or not masterkey_file.master_key or not masterkey_file.domain_backup_key:
        raise ValueError("Invalid masterkey file")

    print("\n=== Adding Real DPAPI Data ===")

    print("=== DPAPI Library Usage with Events ===")
    async with DpapiManager(storage_backend="memory") as manager:
        # Register custom observer
        monitor = MyDpapiEventMonitor()
        await manager.subscribe(monitor)

        # Add masterkeys
        domain_mk_guid = UUID("12345678-1234-5678-9abc-123456789abc")
        await manager.upsert_masterkey(
            MasterKey(
                guid=domain_mk_guid,
                encrypted_key_usercred=masterkey_file.master_key[:-1] + b"\x00",
                encrypted_key_backup=masterkey_file.domain_backup_key[:-1] + b"\x00",
            )
        )

        cred_mk_guid1 = UUID("87654321-4321-8765-cba9-987654321cba")
        await manager.upsert_masterkey(
            MasterKey(
                guid=cred_mk_guid1,
                encrypted_key_usercred=masterkey_file.master_key[:-1] + b"\x00",
                encrypted_key_backup=masterkey_file.domain_backup_key[:-1] + b"\x00",
            )
        )

        # Create a fake masterkey that won't decrypt
        cred_mk_guid2 = UUID("11111111-2222-3333-4444-555555555555")
        await manager.upsert_masterkey(
            MasterKey(
                guid=cred_mk_guid2,
                encrypted_key_usercred=masterkey_file.master_key[:50] + b"\xff" * 20 + masterkey_file.master_key[70:],
                encrypted_key_backup=masterkey_file.domain_backup_key[:50]
                + b"\xff" * 30
                + masterkey_file.domain_backup_key[80:],
            )
        )

        print("Added 3 masterkeys")

        # Add domain backup key (fake one first)
        backup_key = DomainBackupKey(
            guid=UUID("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"),
            key_data=fake_backup_key_bytes,
        )
        await manager.upsert_domain_backup_key(backup_key)

        # Check results after fake backup key
        all_keys = await manager.get_all_masterkeys()
        decrypted_keys = await manager.get_all_masterkeys(filter_by=MasterKeyFilter.DECRYPTED_ONLY)
        print(f"\nAfter fake backup key - Total masterkeys: {len(all_keys)}, Decrypted: {len(decrypted_keys)}")

        await manager.upsert_masterkey(
            MasterKey(
                guid=masterkey_file.masterkey_guid,
                encrypted_key_usercred=masterkey_file.master_key,
                encrypted_key_backup=masterkey_file.domain_backup_key,
            )
        )

        # Add real backup key
        real_backup_key = DomainBackupKey(
            guid=UUID(backup_key_data["backup_key_guid"]),
            key_data=base64.b64decode(backup_key_data["key"]),
            domain_controller=backup_key_data["dc"],
        )
        await manager.upsert_domain_backup_key(real_backup_key)

        # Give auto-decryption time to work
        await asyncio.sleep(1)

        # Check final results
        all_keys_final = await manager.get_all_masterkeys()
        decrypted_keys_final = await manager.get_all_masterkeys(filter_by=MasterKeyFilter.DECRYPTED_ONLY)
        print(
            f"After real backup key - Total masterkeys: {len(all_keys_final)}, Decrypted: {len(decrypted_keys_final)}"
        )

        # Print the decrypted masterkeys  in the form of {GUID}:SHA1
        for key in decrypted_keys_final:
            print(f"{key.guid}:{key.plaintext_key_sha1.hex()}")  # type: ignore

        # Demonstrate blob decryption with the blob_without_entropy.bin fixture
        print("\n=== Decrypting DPAPI Blob ===")

        # Load and parse the blob
        with open(blob_path, "rb") as f:
            blob_data = f.read()

        # Parse blob to get its structure and masterkey GUID

        blob = Blob.parse(blob_data)
        print(f"Blob masterkey GUID: {blob.masterkey_guid}")

        # Decrypt the blob using the DPAPI manager
        decrypted_blob_data = await manager.decrypt_blob(blob)
        print(f"Decrypted blob data: {decrypted_blob_data.decode('utf-8')}")

        # Update domain_mk_guid to use the real one for blob decryption test
        if decrypted_keys_final:
            domain_mk_guid = masterkey_file.masterkey_guid


if __name__ == "__main__":
    asyncio.run(main())