forensic_diff_app.py

import os
import io
import sys
import json
import sqlite3
import plistlib
import traceback
import unittest
import tempfile
from unittest.mock import patch, MagicMock, mock_open
import tkinter as tk
from tkinter import filedialog, ttk, messagebox
from pathlib import Path
from deepdiff import DeepDiff

# ---------------------------------------------------------------------------
# External Forensic Parsers (cclgroupltd & Protobuf)
# ---------------------------------------------------------------------------
try:
    import ccl_bplist
except ImportError:
    ccl_bplist = None

try:
    import ccl_leveldb
except ImportError:
    ccl_leveldb = None

try:
    import ccl_segdb
except ImportError:
    ccl_segdb = None

try:
    import blackboxprotobuf
except ImportError:
    blackboxprotobuf = None


# ---------------------------------------------------------------------------
# Core Parsing Engine
# ---------------------------------------------------------------------------
class ForensicParser:
    """Parses various forensic artifacts and handles deeply nested structures."""

    @classmethod
    def parse_file(cls, filepath):
        filepath = Path(filepath)
        if not filepath.exists() or not filepath.is_file():
            return None

        try:
            with open(filepath, 'rb') as f:
                header = f.read(512)
        except Exception:
            return None

        parsed_data = None
        ext = filepath.suffix.lower()

        try:
            if header.startswith(b'SQLite format 3'):
                parsed_data = cls.parse_sqlite(filepath)
            elif header.startswith(b'bplist00'):
                parsed_data = cls.parse_bplist(filepath)
            elif header.startswith(b'<?xml') and b'<plist' in header:
                parsed_data = cls.parse_xml_plist(filepath)
            elif ext == '.ldb' or filepath.name == 'CURRENT':
                parsed_data = cls.parse_leveldb(filepath.parent)
            elif ext == '.segdb':
                parsed_data = cls.parse_segdb(filepath)
            else:
                with open(filepath, 'rb') as f:
                    parsed_data = cls.recursive_inspect(f.read())
        except Exception as e:
            parsed_data = {"_error": f"Failed to parse {filepath.name}: {str(e)}"}

        return parsed_data

    @classmethod
    def parse_sqlite(cls, filepath):
        db_data = {}
        try:
            conn = sqlite3.connect(f"file:{filepath}?mode=ro", uri=True)
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()

            cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
            tables = [row[0] for row in cursor.fetchall()]

            for table in tables:
                cursor.execute(f'SELECT * FROM "{table}"')
                rows = [dict(row) for row in cursor.fetchall()]
                db_data[table] = cls.recursive_inspect(rows)

            conn.close()
        except Exception as e:
            db_data["_sqlite_error"] = str(e)
        return db_data

    @classmethod
    def parse_leveldb(cls, dirpath):
        if not globals().get('ccl_leveldb'):
            return {"_error": "ccl_leveldb not installed."}

        db_data = {}
        try:
            db = globals()['ccl_leveldb'].RawLevelDb(str(dirpath))
            for record in db.iterate_records_raw():
                key = cls.recursive_inspect(record.key)
                val = cls.recursive_inspect(record.value)

                if isinstance(key, (dict, list)):
                    key = str(key)

                db_data[key] = val
            db.close()
        except Exception as e:
            db_data["_leveldb_error"] = str(e)
        return db_data

    @classmethod
    def parse_segdb(cls, filepath):
        if not globals().get('ccl_segdb'):
            return {"_error": "ccl_segdb not installed."}

        db_data = []
        try:
            with open(filepath, 'rb') as f:
                db = globals()['ccl_segdb'].SegDb(f)
                for record in db.records():
                    db_data.append(cls.recursive_inspect(record.data))
        except Exception as e:
            db_data = {"_segdb_error": str(e)}
        return db_data

    @classmethod
    def parse_bplist(cls, filepath):
        if not globals().get('ccl_bplist'):
            return {"_error": "ccl_bplist not installed."}
        with open(filepath, 'rb') as f:
            parsed = globals()['ccl_bplist'].load(f)
        return cls.recursive_inspect(parsed)

    @classmethod
    def parse_xml_plist(cls, filepath):
        with open(filepath, 'rb') as f:
            parsed = plistlib.load(f)
        return cls.recursive_inspect(parsed)

    @classmethod
    def recursive_inspect(cls, data):
        if isinstance(data, dict):
            return {k: cls.recursive_inspect(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [cls.recursive_inspect(item) for item in data]
        elif isinstance(data, tuple):
            return tuple(cls.recursive_inspect(item) for item in data)
        elif isinstance(data, bytes):
            if not data:
                return data

            ccl_bp = globals().get('ccl_bplist')
            if data.startswith(b'bplist00') and ccl_bp:
                try:
                    parsed = ccl_bp.load(io.BytesIO(data))
                    return cls.recursive_inspect(parsed)
                except Exception:
                    pass

            if data.startswith(b'<?xml') and b'<plist' in data:
                try:
                    parsed = plistlib.loads(data)
                    return cls.recursive_inspect(parsed)
                except Exception:
                    pass

            bb_proto = globals().get('blackboxprotobuf')
            if bb_proto:
                try:
                    msg, _ = bb_proto.decode_message(data)
                    if isinstance(msg, dict) and len(msg) > 0:
                        return cls.recursive_inspect(msg)
                except Exception:
                    pass

            try:
                return data.decode('utf-8')
            except UnicodeDecodeError:
                return data.hex()

        return data


# ---------------------------------------------------------------------------
# GUI Application
# ---------------------------------------------------------------------------
class App(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("Forensic Deep Diff Viewer")
        self.geometry("1000x700")

        self.folder1_path = tk.StringVar()
        self.folder2_path = tk.StringVar()

        self.diff_results = {}
        self.setup_ui()

    def setup_ui(self):
        top_frame = ttk.Frame(self, padding=10)
        top_frame.pack(side=tk.TOP, fill=tk.X)

        ttk.Label(top_frame, text="Folder A (Baseline):").grid(row=0, column=0, sticky=tk.W)
        ttk.Entry(top_frame, textvariable=self.folder1_path, width=60).grid(row=0, column=1, padx=5, pady=5)
        ttk.Button(top_frame, text="Browse...", command=lambda: self.browse_folder(self.folder1_path)).grid(row=0,
                                                                                                            column=2)

        ttk.Label(top_frame, text="Folder B (Changed):").grid(row=1, column=0, sticky=tk.W)
        ttk.Entry(top_frame, textvariable=self.folder2_path, width=60).grid(row=1, column=1, padx=5, pady=5)
        ttk.Button(top_frame, text="Browse...", command=lambda: self.browse_folder(self.folder2_path)).grid(row=1,
                                                                                                            column=2)

        compare_btn = ttk.Button(top_frame, text="Compare Folders", command=self.run_comparison)
        compare_btn.grid(row=2, column=1, pady=10)

        paned = ttk.PanedWindow(self, orient=tk.HORIZONTAL)
        paned.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)

        left_frame = ttk.Frame(paned)
        paned.add(left_frame, weight=1)

        ttk.Label(left_frame, text="Modified Files").pack(anchor=tk.W)
        self.file_listbox = tk.Listbox(left_frame)
        self.file_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
        self.file_listbox.bind('<<ListboxSelect>>', self.on_file_select)

        scrollbar = ttk.Scrollbar(left_frame, command=self.file_listbox.yview)
        scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
        self.file_listbox.config(yscrollcommand=scrollbar.set)

        right_frame = ttk.Frame(paned)
        paned.add(right_frame, weight=3)

        top_right_frame = ttk.Frame(right_frame)
        top_right_frame.pack(side=tk.TOP, fill=tk.X)
        ttk.Label(top_right_frame, text="Diff Details").pack(side=tk.LEFT)
        ttk.Button(top_right_frame, text="Export Diff to JSON", command=self.export_diff).pack(side=tk.RIGHT)

        self.diff_text = tk.Text(right_frame, wrap=tk.NONE)
        self.diff_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

        text_scroll_y = ttk.Scrollbar(right_frame, orient=tk.VERTICAL, command=self.diff_text.yview)
        text_scroll_y.pack(side=tk.RIGHT, fill=tk.Y)
        text_scroll_x = ttk.Scrollbar(self.diff_text, orient=tk.HORIZONTAL, command=self.diff_text.xview)
        text_scroll_x.pack(side=tk.BOTTOM, fill=tk.X)
        self.diff_text.config(yscrollcommand=text_scroll_y.set, xscrollcommand=text_scroll_x.set)

    def browse_folder(self, string_var):
        folder = filedialog.askdirectory()
        if folder:
            string_var.set(folder)

    @staticmethod
    def get_relative_files(folder_path):
        files = set()
        for root, _, filenames in os.walk(folder_path):
            for filename in filenames:
                full_path = os.path.join(root, filename)
                rel_path = os.path.relpath(full_path, folder_path)
                files.add(rel_path)
        return files

    def run_comparison(self):
        folder1 = self.folder1_path.get()
        folder2 = self.folder2_path.get()

        if not folder1 or not folder2:
            messagebox.showwarning("Warning", "Please select both folders.")
            return

        self.file_listbox.delete(0, tk.END)
        self.diff_text.delete(1.0, tk.END)
        self.diff_results.clear()

        files1 = self.get_relative_files(folder1)
        files2 = self.get_relative_files(folder2)

        common_files = files1.intersection(files2)

        for rel_path in sorted(common_files):
            self.file_listbox.insert(tk.END, rel_path)

        messagebox.showinfo("Status", f"Found {len(common_files)} common files. Select a file to view its changes.")
        self.common_files_set = common_files

    def on_file_select(self, event):
        selection = self.file_listbox.curselection()
        if not selection:
            return

        rel_path = self.file_listbox.get(selection[0])

        folder1 = Path(self.folder1_path.get())
        folder2 = Path(self.folder2_path.get())

        file1_path = folder1 / rel_path
        file2_path = folder2 / rel_path

        self.diff_text.delete(1.0, tk.END)
        self.diff_text.insert(tk.END, f"Parsing {rel_path}...\nThis may take a moment depending on file size/nesting.")
        self.update()

        try:
            data1 = ForensicParser.parse_file(file1_path)
            data2 = ForensicParser.parse_file(file2_path)

            if data1 is None and data2 is None:
                self.diff_text.delete(1.0, tk.END)
                self.diff_text.insert(tk.END, "Unsupported file format or unreadable files.")
                return

            diff = DeepDiff(data1, data2, ignore_order=True, report_repetition=True)

            self.diff_results[rel_path] = {
                "file1": str(file1_path),
                "file2": str(file2_path),
                "diff": json.loads(diff.to_json())
            }

            self.diff_text.delete(1.0, tk.END)

            if not diff:
                self.diff_text.insert(tk.END, "No changes detected between the files.")
            else:
                formatted_diff = json.dumps(json.loads(diff.to_json()), indent=4)
                self.diff_text.insert(tk.END, formatted_diff)

        except Exception as e:
            self.diff_text.delete(1.0, tk.END)
            self.diff_text.insert(tk.END, f"Error computing diff:\n\n{traceback.format_exc()}")

    def export_diff(self):
        if not self.diff_results:
            messagebox.showinfo("Export", "No diffs computed yet. Please select a file to compute its diff first.")
            return

        save_path = filedialog.asksaveasfilename(
            defaultextension=".json",
            filetypes=[("JSON files", "*.json")],
            title="Save Diff Results"
        )
        if save_path:
            try:
                with open(save_path, 'w', encoding='utf-8') as f:
                    json.dump(self.diff_results, f, indent=4)
                messagebox.showinfo("Success", "Diff results exported successfully.")
            except Exception as e:
                messagebox.showerror("Export Error", str(e))


# ---------------------------------------------------------------------------
# Unit Tests
# ---------------------------------------------------------------------------
class MockLevelDBRecord:
    def __init__(self, key, value):
        self.key = key
        self.value = value


class MockSegDBRecord:
    def __init__(self, data):
        self.data = data


class TestForensicParser(unittest.TestCase):

    def setUp(self):
        self.orig_bplist = globals().get('ccl_bplist')
        self.orig_leveldb = globals().get('ccl_leveldb')
        self.orig_segdb = globals().get('ccl_segdb')
        self.orig_bbpb = globals().get('blackboxprotobuf')

        globals()['ccl_leveldb'] = MagicMock()
        globals()['ccl_segdb'] = MagicMock()
        globals()['blackboxprotobuf'] = MagicMock()

        mock_ccl = MagicMock()
        mock_ccl.load.side_effect = lambda f: plistlib.load(f)
        globals()['ccl_bplist'] = mock_ccl

    def tearDown(self):
        globals()['ccl_bplist'] = self.orig_bplist
        globals()['ccl_leveldb'] = self.orig_leveldb
        globals()['ccl_segdb'] = self.orig_segdb
        globals()['blackboxprotobuf'] = self.orig_bbpb

    def test_recursive_inspect_real_deep_nesting(self):
        inner_dict = {"Target": "Achieved"}
        bplist_bytes = plistlib.dumps(inner_dict, fmt=plistlib.FMT_BINARY)

        xml_dict = {"NestedBinary": bplist_bytes}
        xml_bytes = plistlib.dumps(xml_dict, fmt=plistlib.FMT_XML)

        complex_data = {
            "Level1": [
                {"Level2": xml_bytes},
                b"Just a normal string",
                b"\x00\xFF\x00"
            ]
        }

        result = ForensicParser.recursive_inspect(complex_data)

        expected = {
            "Level1": [
                {"Level2": {"NestedBinary": {"Target": "Achieved"}}},
                "Just a normal string",
                "00ff00"
            ]
        }
        self.assertEqual(result, expected)

    def test_real_sqlite_with_nested_xml_plist(self):
        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
            tmp_name = tmp.name

        try:
            conn = sqlite3.connect(tmp_name)
            cursor = conn.cursor()
            cursor.execute("CREATE TABLE AppData (id INTEGER, payload BLOB)")

            plist_payload = {"UserPreferences": {"DarkMode": True, "FontSize": 14}}
            plist_bytes = plistlib.dumps(plist_payload, fmt=plistlib.FMT_XML)

            cursor.execute("INSERT INTO AppData (id, payload) VALUES (?, ?)", (1, plist_bytes))
            cursor.execute("INSERT INTO AppData (id, payload) VALUES (?, ?)", (2, b"Plain text BLOB"))

            conn.commit()
            conn.close()

            result = ForensicParser.parse_sqlite(tmp_name)

            expected = {
                "AppData": [
                    {"id": 1, "payload": {"UserPreferences": {"DarkMode": True, "FontSize": 14}}},
                    {"id": 2, "payload": "Plain text BLOB"}
                ]
            }
            self.assertEqual(result, expected)
        finally:
            if os.path.exists(tmp_name):
                os.remove(tmp_name)

    def test_parse_leveldb_with_real_nested_bplist(self):
        bplist_payload = {"Highscore": 9999, "Username": "Player1"}
        real_bplist_bytes = plistlib.dumps(bplist_payload, fmt=plistlib.FMT_BINARY)

        mock_db = MagicMock()
        mock_db.iterate_records_raw.return_value = [
            MockLevelDBRecord(key=b"game_save_slot_1", value=real_bplist_bytes)
        ]
        globals()['ccl_leveldb'].RawLevelDb.return_value = mock_db

        result = ForensicParser.parse_leveldb("dummy_dir")

        expected = {"game_save_slot_1": {"Highscore": 9999, "Username": "Player1"}}
        self.assertEqual(result, expected)

    def test_protobuf_fallback_mocked(self):
        raw_proto_bytes = b'\x08\x96\x01'
        globals()['blackboxprotobuf'].decode_message.return_value = ({"1": 150}, {})

        data = {"ProtoData": raw_proto_bytes}
        result = ForensicParser.recursive_inspect(data)

        self.assertEqual(result, {"ProtoData": {"1": 150}})

    @patch('pathlib.Path.exists', return_value=True)
    @patch('pathlib.Path.is_file', return_value=True)
    def test_parse_file_router(self, mock_is_file, mock_exists):
        with patch('builtins.open', mock_open(read_data=b'SQLite format 3\x00')) as mocked_file:
            with patch.object(ForensicParser, 'parse_sqlite', return_value={"mock": "sqlite"}) as mock_sqlite:
                result = ForensicParser.parse_file("any_file.xyz")
                mock_sqlite.assert_called_once()
                self.assertEqual(result, {"mock": "sqlite"})

        xml_header_mock = b'<?xml version="1.0" encoding="UTF-8"?>\n<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">\n<plist version="1.0">'
        with patch('builtins.open', mock_open(read_data=xml_header_mock)) as mocked_file:
            with patch.object(ForensicParser, 'parse_xml_plist', return_value={"mock": "xml"}) as mock_xml:
                result = ForensicParser.parse_file("any_file.plist")
                mock_xml.assert_called_once()
                self.assertEqual(result, {"mock": "xml"})


class TestAppUtils(unittest.TestCase):

    @patch('os.walk')
    def test_get_relative_files(self, mock_walk):
        mock_walk.return_value = [
            ('/base/folder', ['subfolder'], ['file1.txt', 'file2.txt']),
            ('/base/folder/subfolder', [], ['file3.db'])
        ]

        files = App.get_relative_files('/base/folder')

        expected = {'file1.txt', 'file2.txt', os.path.join('subfolder', 'file3.db')}
        self.assertEqual(files, expected)


# ---------------------------------------------------------------------------
# Execution Entry Point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--test":
        sys.argv.pop(1)
        unittest.main()
    else:
        app = App()
        app.mainloop()