ctf/2026/srdnlen_quals/forensic/chapter3/parse_fs_events.py

#!/usr/bin/env python3
"""
parse_fsevents.py — Parse macOS FSEvents logs and look for 64-char hex strings (potential keys).

Usage:
    python3 parse_fsevents.py /path/to/.fseventsd/

The .fseventsd directory is inside the mounted DMG. Each file in it is a
gzip-compressed binary log. This script will:
  1. Parse every log file
  2. Extract all file paths
  3. Flag any path component that looks like a 64-char hex string (a potential key)
"""

import gzip
import os
import re
import struct
import sys

# Regex to match 64-char hex string anywhere in a path
HEX64 = re.compile(r'\b([0-9a-fA-F]{64})\b')


def parse_fsevent_record(data: bytes, offset: int):
    """Parse a single FSEvent record. Returns (path, flags, event_id, next_offset)."""
    # Format: null-terminated path string, then 4-byte flags, then 8-byte event_id
    end = data.index(b'\x00', offset)
    path = data[offset:end].decode('utf-8', errors='replace')
    offset = end + 1
    if offset + 12 > len(data):
        return path, 0, 0, len(data)
    flags = struct.unpack_from('<I', data, offset)[0]
    event_id = struct.unpack_from('<Q', data, offset + 4)[0]
    return path, flags, event_id, offset + 12


def parse_fsevent_file(filepath: str):
    """Parse one FSEvents log file, yield (path, flags, event_id) tuples."""
    with open(filepath, 'rb') as f:
        raw = f.read()

    # FSEvents files start with a 'DLS' magic, then gzip chunks
    # Each chunk: 4-byte magic (1SLD), 4-byte size, gzip data
    # OR the whole file might be one gzip stream (older format)
    
    results = []
    
    # Try multi-chunk format (macOS 10.5+)
    # Magic: b'\x31\x53\x4c\x44' = "1SLD" or b'1SLD'
    if raw[:4] in (b'1SLD', b'2SLD'):
        offset = 0
        while offset < len(raw):
            if offset + 8 > len(raw):
                break
            magic = raw[offset:offset+4]
            if magic not in (b'1SLD', b'2SLD'):
                break
            size = struct.unpack_from('<I', raw, offset + 4)[0]
            chunk_data = raw[offset + 8: offset + 8 + size]
            offset += 8 + size
            try:
                decompressed = gzip.decompress(chunk_data)
                pos = 0
                while pos < len(decompressed):
                    try:
                        path, flags, event_id, pos = parse_fsevent_record(decompressed, pos)
                        results.append((path, flags, event_id))
                    except (ValueError, struct.error):
                        break
            except Exception:
                pass
    else:
        # Try raw gzip
        try:
            decompressed = gzip.decompress(raw)
            pos = 0
            while pos < len(decompressed):
                try:
                    path, flags, event_id, pos = parse_fsevent_record(decompressed, pos)
                    results.append((path, flags, event_id))
                except (ValueError, struct.error):
                    break
        except Exception:
            pass

    return results


def flag_description(flags: int) -> str:
    """Human-readable flags."""
    names = []
    flag_map = {
        0x00000001: 'Created',
        0x00000002: 'Removed',
        0x00000004: 'InodeMetaMod',
        0x00000008: 'Renamed',
        0x00000010: 'Modified',
        0x00000020: 'Exchange',
        0x00000040: 'FinderInfoMod',
        0x00000080: 'FolderCreated',
        0x00000100: 'PermissionChanged',
        0x00000200: 'XattrModified',
        0x00000400: 'XattrRemoved',
        0x00001000: 'DocumentRevision',
        0x00400000: 'ItemIsFile',
        0x00800000: 'ItemIsDir',
        0x01000000: 'ItemIsSymlink',
        0x02000000: 'OwnedByActiveUSSD',
        0x04000000: 'ItemIsHardlink',
        0x10000000: 'ItemIsLastHardlink',
    }
    for bit, name in flag_map.items():
        if flags & bit:
            names.append(name)
    return '|'.join(names) if names else f'0x{flags:08x}'


def main():
    if len(sys.argv) < 2:
        print("Usage: python3 parse_fsevents.py /path/to/.fseventsd/")
        sys.exit(1)

    fseventsd_dir = sys.argv[1]
    
    all_paths = []
    hex_key_hits = []

    log_files = sorted([
        f for f in os.listdir(fseventsd_dir)
        if not f.startswith('.') and f != 'no_log' and f != 'fseventsd-uuid'
    ])

    print(f"[*] Found {len(log_files)} FSEvents log file(s) in {fseventsd_dir}")
    print()

    for fname in log_files:
        fpath = os.path.join(fseventsd_dir, fname)
        print(f"[+] Parsing: {fname}")
        records = parse_fsevent_file(fpath)
        print(f"    {len(records)} records found")
        
        for path, flags, event_id in records:
            all_paths.append(path)
            # Look for 64-char hex strings in the path
            m = HEX64.search(path)
            if m:
                hex_key_hits.append({
                    'key': m.group(1).lower(),
                    'path': path,
                    'flags': flags,
                    'flags_str': flag_description(flags),
                    'event_id': event_id,
                    'source_file': fname,
                })

    print()
    print("=" * 70)
    print(f"[*] Total paths parsed: {len(all_paths)}")
    print(f"[*] Paths containing 64-char hex strings: {len(hex_key_hits)}")
    print("=" * 70)

    if hex_key_hits:
        print("\n[!] POTENTIAL KEYS FOUND IN FSEVENTS:\n")
        for hit in hex_key_hits:
            print(f"  Key:      {hit['key']}")
            print(f"  Path:     {hit['path']}")
            print(f"  Flags:    {hit['flags_str']}")
            print(f"  EventID:  {hit['event_id']}")
            print(f"  Source:   {hit['source_file']}")
            print()
    else:
        print("\n[*] No 64-char hex strings found in paths.")
        print("[*] Showing all unique paths for manual analysis:\n")
        for p in sorted(set(all_paths)):
            print(f"  {p}")

    # Also dump full path list to file
    out_path = "fsevents_all_paths.txt"
    with open(out_path, 'w') as f:
        for path, flags, event_id in []:
            pass
        # Write unique paths
        for p in sorted(set(all_paths)):
            f.write(p + '\n')
    print(f"\n[*] All unique paths written to: {out_path}")


if __name__ == '__main__':
    main()