import json
import hashlib
import shutil
import contextlib
import tempfile
import os
from test import test_volatility, WindowsSamples


class TestWindowsVolshell:
    def test_windows_volshell(self, image, volatility, python):
        out = test_volatility.basic_volshell_test(
            image, volatility, python, volshellargs=("-w",)
        )
        assert out.count(b"<EPROCESS") > 40


class TestWindowsPslist:
    def test_windows_generic_pslist(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.pslist.PsList",
            image,
            volatility,
            python,
            # Notice that this is needed to hit lru_cache when "specific" will run
            globalargs=("-r", "json"),
        )
        assert rc == 0
        out = out.lower()
        assert out.find(b"system") != -1
        assert out.find(b"csrss.exe") != -1
        assert out.find(b"svchost.exe") != -1
        assert out.count(b"\n") > 10

    def test_windows_specific_pslist(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.pslist.PsList",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        expected_row = {
            "CreateTime": None,
            "ExitTime": None,
            "File output": "Disabled",
            "Handles": 1140,
            "ImageFileName": "System",
            "Offset(V)": 2185004992,
            "PID": 4,
            "PPID": 0,
            "SessionId": None,
            "Threads": 61,
            "Wow64": False,
            "__children": [],
        }
        assert test_volatility.match_output_row(expected_row, json.loads(out))


class TestWindowsPsscan:
    def test_windows_generic_psscan(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.psscan.PsScan", image, volatility, python
        )
        assert rc == 0
        out = out.lower()
        assert out.find(b"system") != -1
        assert out.find(b"csrss.exe") != -1
        assert out.find(b"svchost.exe") != -1
        assert out.count(b"\n") > 10


class TestWindowsDlllist:
    def test_windows_generic_dlllist(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.dlllist.DllList",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 2000
        expected_rows = [
            {
                "Path": "C:\\Windows\\SYSTEM32\\kernel32.dll",
                "Process": "csrss.exe",
            },
            {
                "Path": "C:\\Windows\\system32\\USER32.dll",
                "Process": "csrss.exe",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(
                expected_row, json_out, case_sensitive=False
            )


class TestWindowsModules:
    def test_windows_specific_modules(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.modules.Modules",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 110
        expected_rows = [
            {
                "Name": "ntoskrnl.exe",
                "Offset": 2185216944,
                "Path": "\\WINDOWS\\system32\\ntoskrnl.exe",
                "Size": 2179328,
            },
            {
                "Name": "hal.dll",
                "Offset": 2185216840,
                "Path": "\\WINDOWS\\system32\\hal.dll",
                "Size": 81280,
            },
            {
                "Name": "netbios.sys",
                "Offset": 2182050616,
                "Path": "\\SystemRoot\\System32\\DRIVERS\\netbios.sys",
                "Size": 36864,
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsDumpfiles:
    def test_windows_specific_dumpfiles(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        with open("./test/known_files.json") as json_file:
            known_files = json.load(json_file)

        failed_chksms = 0
        file_name = os.path.basename(image)

        try:
            for addr in known_files["windows_dumpfiles"][file_name]:
                path = tempfile.mkdtemp()

                rc, _out, _err = test_volatility.runvol_plugin(
                    "windows.dumpfiles.DumpFiles",
                    image,
                    volatility,
                    python,
                    globalargs=("-o", path),
                    pluginargs=("--virtaddr", addr),
                )

                for file in os.listdir(path):
                    with open(os.path.join(path, file), "rb") as fp:
                        if (
                            hashlib.md5(fp.read()).hexdigest()
                            not in known_files["windows_dumpfiles"][file_name][addr]
                        ):
                            failed_chksms += 1

                shutil.rmtree(path)
            json_file.close()

            assert failed_chksms == 0
            assert rc == 0
        except Exception as e:
            json_file.close()
            print("Key Error raised on " + str(e))
            assert False


class TestWindowsHandles:
    def test_windows_generic_handles(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.handles.Handles",
            image,
            volatility,
            python,
            pluginargs=("--pid", "4"),
        )
        assert rc == 0
        assert out.find(b"System Pid 4") != -1
        assert (
            out.find(
                b"MACHINE\\SYSTEM\\CONTROLSET001\\CONTROL\\SESSION MANAGER\\MEMORY MANAGEMENT\\PREFETCHPARAMETERS"
            )
            != -1
        )
        assert out.find(b"MACHINE\\SYSTEM\\SETUP") != -1
        assert out.count(b"\n") > 500


class TestWindowsSvcList:
    def test_windows_generic_svclist(self, volatility, python, image):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.svclist.SvcList",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert len(json_out) > 250
        expected_row = {
            "Binary": "\\Driver\\ACPI",
            "Display": "ACPI",
            "Name": "ACPI",
            "Start": "SERVICE_BOOT_START",
            "State": "SERVICE_RUNNING",
            "Type": "SERVICE_KERNEL_DRIVER",
        }
        assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsSvcScan:
    def test_windows_generic_svcscan(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.svcscan.SvcScan",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert len(json_out) > 250
        expected_rows = [
            {"Name": "ACPI", "Type": "SERVICE_KERNEL_DRIVER"},
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsThrdscan:
    def test_windows_generic_thrdscan(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.thrdscan.ThrdScan", image, volatility, python
        )
        assert rc == 0
        assert out.find(b"\t4\t8") != -1
        assert out.find(b"\t4\t12") != -1
        assert out.find(b"\t4\t16") != -1


class TestWindowsPrivileges:
    def test_windows_generic_privileges(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.privileges.Privs",
            image,
            volatility,
            python,
            pluginargs=("--pid", "4"),
        )
        assert rc == 0
        assert out.find(b"SeCreateTokenPrivilege") != -1
        assert out.find(b"SeCreateGlobalPrivilege") != -1
        assert out.find(b"SeAssignPrimaryTokenPrivilege") != -1
        assert out.count(b"\n") > 20


class TestWindowsGetSIDs:
    def test_windows_generic_getsids(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.getsids.GetSIDs",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 400
        expected_rows = [
            {
                "Name": "Local System",
                "Process": "csrss.exe",
                "SID": "S-1-5-18",
            },
            {
                "Name": "Administrators",
                "Process": "csrss.exe",
                "SID": "S-1-5-32-544",
            },
            {
                "Name": "Everyone",
                "Process": "csrss.exe",
                "SID": "S-1-1-0",
            },
            {
                "Name": "Authenticated Users",
                "Process": "csrss.exe",
                "SID": "S-1-5-11",
            },
            {
                "Name": "System Mandatory Level",
                "Process": "csrss.exe",
                "SID": "S-1-16-16384",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsEnvars:
    def test_windows_generic_envars(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.envars.Envars", image, volatility, python
        )
        assert rc == 0
        assert out.find(b"PATH") != -1
        assert out.find(b"PROCESSOR_ARCHITECTURE") != -1
        assert out.find(b"USERNAME") != -1
        assert out.find(b"SystemRoot") != -1
        assert out.find(b"CommonProgramFiles") != -1
        assert out.count(b"\n") > 500


class TestWindowsCallbacks:
    def test_windows_generic_callbacks(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.callbacks.Callbacks", image, volatility, python
        )
        assert rc == 0
        assert out.find(b"PspCreateProcessNotifyRoutine") != -1
        assert out.find(b"KeBugCheckCallbackListHead") != -1
        assert out.find(b"KeBugCheckReasonCallbackListHead") != -1
        assert out.count(b"KeBugCheckReasonCallbackListHead	") > 5


class TestWindowsVadwalk:
    def test_windows_specific_vadwalk(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.vadwalk.VadWalk",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
            pluginargs=("--pid", "4"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = test_volatility.load_test_data(
            "windows.vadwalk.VadWalk", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsDevicetree:
    def test_windows_generic_devicetree(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.devicetree.DeviceTree", image, volatility, python
        )
        assert rc == 0
        assert out.find(b"DEV") != -1
        assert out.find(b"DRV") != -1
        assert out.find(b"ATT") != -1
        assert out.find(b"FILE_DEVICE_CONTROLLER") != -1
        assert out.find(b"FILE_DEVICE_DISK") != -1
        assert out.find(b"FILE_DEVICE_DISK_FILE_SYSTEM") != -1


class TestWindowsVadyarascan:
    def test_windows_specific_vadyarascan_yara_rule(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        yara_rule_01 = r"""
            rule fullvadyarascan
            {
                strings:
                    $s1 = "!This program cannot be run in DOS mode."
                    $s2 = "Qw))Pw"
                    $s3 = "W_wD)Pw"
                    $s4 = "1Xw+2Xw"
                    $s5 = "xd`wh``w"
                    $s6 = "0g`w0g`w8g`w8g`w@g`w@g`wHg`wHg`wPg`wPg`wXg`wXg`w`g`w`g`whg`whg`wpg`wpg`wxg`wxg`w"
                condition:
                    all of them
            }
        """
        fd, filename = tempfile.mkstemp(suffix=".yar")
        try:
            with os.fdopen(fd, "w") as f:
                f.write(yara_rule_01)
            rc, out, _err = test_volatility.runvol_plugin(
                "windows.vadyarascan.VadYaraScan",
                image,
                volatility,
                python,
                pluginargs=("--pid", "4012", "--yara-file", filename),
            )
        finally:
            with contextlib.suppress(FileNotFoundError):
                os.remove(filename)
        assert rc == 0
        assert out.count(b"\n") > 4

    def test_windows_specific_vadyarascan_yara_string(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.vadyarascan.VadYaraScan",
            image,
            volatility,
            python,
            pluginargs=("--pid", "4012", "--yara-string", "MZ"),
        )
        assert rc == 0
        assert out.count(b"\n") > 10


class TestWindowsAmcache:
    def test_windows_generic_amcache(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.amcache.Amcache",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 100
        # Win10+ expected package names
        expected_rows = [
            {
                "Path": "C:\\Windows\\SystemApps\\Microsoft.Windows.StartMenuExperienceHost_cw5n1h2txyewy",
                "ProductName": "Microsoft.Windows.StartMenuExperienceHost",
            },
            {
                "Path": "C:\\Windows\\SystemApps\\Microsoft.Windows.FileExplorer_cw5n1h2txyewy",
                "ProductName": "c5e2524a-ea46-4f67-841f-6a9465d9d515",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsBigPools:
    def test_windows_generic_bigpools(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.bigpools.BigPools",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 2000
        expected_rows = [
            {
                "PoolType": "PagedPool",
            },
            {
                "PoolType": "PagedPoolCacheAligned",
            },
            {
                "PoolType": "NonPagedPoolNx",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


# FIXME: Empty on WIN10 and XP samples
# class TestWindowsCachedump:
#     def test_windows_generic_cachedump(self, volatility, python, image):
#         rc, out, _err = test_volatility.runvol_plugin(
#             "windows.cachedump.Cachedump",
#             image,
#             volatility,
#             python,
#             globalargs=("-r", "json"),
#         )
#         assert rc == 0
#         json_out = json.loads(out)


class TestWindowsCmdLine:
    def test_windows_generic_cmdline(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.cmdline.CmdLine",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 20
        out = out.lower()
        assert (
            out.find(b"C:\\Windows\\system32\\svchost.exe -k DcomLaunch -p".lower())
            != -1
        )
        assert (
            out.count(
                b"C:\\Windows\\system32\\svchost.exe -k LocalServiceNetworkRestricted -p".lower()
            )
            > 3
        )


class TestWindowsCmdScan:
    def test_windows_specific_cmdscan(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.cmdscan.CmdScan",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        expected_row = {
            "Process": "conhost.exe",
            "Property": "_COMMAND_HISTORY",
        }
        assert test_volatility.match_output_row(expected_row, json.loads(out))


class TestWindowsConsoles:
    def test_windows_specific_consoles(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.consoles.Consoles",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        expected_row = {
            "Process": "conhost.exe",
            "Property": "_CONSOLE_INFORMATION",
        }
        assert test_volatility.match_output_row(expected_row, json.loads(out))


class TestWindowsCrashinfo:
    def test_windows_specific_crashinfo(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.crashinfo.Crashinfo",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        expected_row = {
            "BitmapHeaderSize": 176128,
            "BitmapPages": 511191,
            "BitmapSize": 1310720,
            "Comment": "PAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGEPAGE",
            "DirectoryTableBase": 4610162688,
            "DumpType": "Bitmap Dump (0x5)",
            "MachineImageType": 34404,
            "MajorVersion": 15,
            "MinorVersion": 19041,
            "NumberProcessors": 1,
            "Signature": "PAGE",
            "SystemTime": "2025-03-06T17:59:20+00:00",
            "SystemUpTime": "0:11:23.199374",
            "__children": [],
        }
        assert test_volatility.match_output_row(expected_row, json.loads(out))


class TestWindowsDriverIrp:
    def test_windows_generic_driverirp(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.driverirp.DriverIrp",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 2000
        assert out.count(b"ntoskrnl") > 400
        for irp in test_volatility.load_test_data(
            "windows.driverirp.DriverIrp", "GENERIC"
        ):
            assert out.find(irp.encode()) != -1


class TestWindowsDriverScan:
    def test_windows_generic_driverscan(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.driverscan.DriverScan",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 50
        expected_rows = [
            {
                "Name": "\\Driver\\ACPI_HAL",
                "Service Key": "\\Driver\\ACPI_HAL",
            },
            {
                "Name": "\\Driver\\Tcpip",
                "Service Key": "Tcpip",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsGetServiceSIDs:
    def test_windows_generic_getservicesids(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.getservicesids.GetServiceSIDs",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"S-1-5-80-") > 90


class TestWindowsIAT:
    def test_windows_generic_iat(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.iat.IAT",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 2000
        expected_rows = [
            {
                "Function": "NtTerminateProcess",
                "Library": "ntdll.dll",
                "Name": "csrss.exe",
            },
            {
                "Function": "RtlSetHeapInformation",
                "Library": "ntdll.dll",
                "Name": "csrss.exe",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsInfo:
    def test_windows_specific_info(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.info.Info",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = test_volatility.load_test_data(
            "windows.info.Info", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsJobLinks:
    def test_windows_specific_joblinks(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.joblinks.JobLinks",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 30
        expected_row = {
            "Active": 1,
            "JobLink": None,
            "JobSess": 2,
            "Name": "taskhostw.exe",
            "Offset(V)": 145201782567040,
            "PID": 4304,
            "PPID": 1008,
            "Process": "(Original Process)",
            "Sess": 2,
            "Term": 0,
            "Total": 1,
            "Wow64": False,
            "__children": [
                {
                    "Active": 0,
                    "JobLink": "Yes",
                    "JobSess": 0,
                    "Name": "taskhostw.exe",
                    "Offset(V)": 145201782567040,
                    "PID": 4304,
                    "PPID": 1008,
                    "Process": "C:\\Windows\\system32\\taskhostw.exe",
                    "Sess": 2,
                    "Term": 0,
                    "Total": 0,
                    "Wow64": False,
                    "__children": [],
                }
            ],
        }

        assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsKPCRs:
    def test_windows_generic_kpcrs(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.kpcrs.KPCRs",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        assert test_volatility.count_entries_flat(json.loads(out)) > 0


class TestWindowsLdrModules:
    def test_windows_generic_ldrmodules(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.ldrmodules.LdrModules",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 800
        out = out.lower()
        assert out.find(b"\\Windows\\System32\\ntdll.dll".lower()) > 10


class TestWindowsLsadump:
    def test_windows_specific_lsadump(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.lsadump.Lsadump",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 5
        expected_row = {
            "Hex": "01 00 00 00 2b 2b f1 09 a3 b3 4b af 02 19 5a 61 2f 09 3a 88 03 52 51 64 8a 6c d2 a8 34 07 cb 61 41 ca a4 5d f1 fb 4c e0 41 72 69 32",
            "Key": "DPAPI_SYSTEM",
        }

        assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsMBRScan:
    def test_windows_specific_mbrscan(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.mbrscan.MBRScan",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 4300
        expected_rows = [
            {
                "Bootcode MD5": "dbcef88b4d770658b0050bf20b2d3061",
                "Disk Signature": "82-78-77-32",
                "Full MBR MD5": "8eea93bb1c63863f6e7f95b084411672",
                "Potential MBR at Physical Offset": 154029739,
            },
            {
                "Bootcode MD5": "591213a9dfef595735e419eff6eeb39d",
                "Disk Signature": "7a-74-60-53",
                "Full MBR MD5": "4e00711a5014941f5ad8b3a4cde69c9c",
                "Potential MBR at Physical Offset": 437808348,
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsMemmap:
    def test_windows_specific_memmap(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.memmap.Memmap",
            image,
            volatility,
            python,
            pluginargs=("--pid", "504"),
        )
        assert rc == 0
        assert out.count(b"\n") > 12000


class TestWindowsMFTscan:
    def test_windows_specific_mftscan_ads_xp(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.mftscan.ADS",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = [
            {
                "ADS Filename": "Zone.Identifier",
                "Filename": "libby_hoeler_part1.wmv",
                "Hexdump": '"\n5b 5a 6f 6e 65 54 72 61 6e 73 66 65 72 5d 0d 0a [ZoneTransfer]..\n5a 6f 6e 65 49 64 3d 33 0d 0a                   ZoneId=3..      "',
                "MFT Type": "DATA",
                "Offset": 55926304,
                "Record Number": 323,
                "Record Type": "FILE",
            },
            {
                "ADS Filename": "Zone.Identifier",
                "Filename": "NetZeroQuickHelpLite.exe",
                "Hexdump": '"\n5b 5a 6f 6e 65 54 72 61 6e 73 66 65 72 5d 0d 0a [ZoneTransfer]..\n5a 6f 6e 65 49 64 3d 33 0d 0a                   ZoneId=3..      "',
                "MFT Type": "DATA",
                "Offset": 56102400,
                "Record Number": 347,
                "Record Type": "FILE",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)

    def test_windows_specific_mftscan_ads_win10(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.mftscan.ADS",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = [
            {
                "ADS Filename": "$Max",
                "Filename": "$UsnJrnl",
                "Hexdump": '"\n00 00 00 02 00 00 00 00 00 00 80 00 00 00 00 00 ................\nb9 dd f0 cc df 73 db 01 00 00 00 00 00 00 00 00 .....s.........."',
                "MFT Type": "DATA",
                "Offset": 1058018088,
                "Record Number": 107240,
                "Record Type": "FILE",
            },
            {
                "ADS Filename": "$Config",
                "Filename": "$Repair",
                "Hexdump": '"\n01 00 00 00 03 00 00 00                         ........        "',
                "MFT Type": "DATA",
                "Offset": 5009678688,
                "Record Number": 28,
                "Record Type": "FILE",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)

    def test_windows_specific_mftscan_mftscan(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.mftscan.MFTScan",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 15000
        assert out.count(b"STANDARD_INFORMATION") > 5000
        assert out.count(b"FILE_NAME") > 11000

    def test_windows_specific_mftscan_residentdata_win10(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.mftscan.ResidentData",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 850
        expected_rows = [
            {
                "Filename": "index",
                "Hexdump": '"\n30 5c 72 a7 1b 6d fb fc 09 00 00 00 00 00 00 00 0\\r..m..........\n00 00 00 00 00 00 00 00                         ........        "',
                "MFT Type": "DATA",
                "Offset": 4961536280,
                "Record Number": 116474,
                "Record Type": "FILE",
            },
            {
                "Filename": "0.2.filtertrie.intermediate.txt",
                "Hexdump": '"\n30 09 32 0d 0a                                  0.2..           "',
                "MFT Type": "DATA",
                "Offset": 619242944,
                "Record Number": 113013,
                "Record Type": "FILE",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsModScan:
    def test_windows_generic_modscan(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.modscan.ModScan",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 90
        expected_rows = [
            {
                "Name": "ntoskrnl.exe",
                "Offset": 37733296,
                "Path": "\\WINDOWS\\system32\\ntoskrnl.exe",
                "Size": 2179328,
            },
            {
                "Name": "hal.dll",
                "Offset": 37733192,
                "Path": "\\WINDOWS\\system32\\hal.dll",
                "Size": 81280,
            },
            {
                "Name": "netbios.sys",
                "Offset": 34566968,
                "Path": "\\SystemRoot\\System32\\DRIVERS\\netbios.sys",
                "Size": 36864,
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsMutantScan:
    def test_windows_specific_mutantscan(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.mutantscan.MutantScan",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 350


class TestWindowsNetScan:
    def test_windows_specific_netscan(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.netscan.NetScan",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 100
        expected_rows = [
            {
                "Created": "2025-03-06T17:56:53+00:00",
                "ForeignAddr": "13.107.246.254",
                "ForeignPort": 443,
                "LocalAddr": "10.0.0.4",
                "LocalPort": 49929,
                "Offset": 145201667934000,
                "Owner": "SearchApp.exe",
                "PID": 5644,
                "Proto": "TCPv4",
                "State": "CLOSE_WAIT",
            },
            {
                "Created": "2025-03-06T17:50:02+00:00",
                "ForeignAddr": "168.63.129.16",
                "ForeignPort": 80,
                "LocalAddr": "10.0.0.4",
                "LocalPort": 49689,
                "Offset": 145201778694688,
                "Owner": "WindowsAzureGu",
                "PID": 1944,
                "Proto": "TCPv4",
                "State": "CLOSED",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsNetStat:
    def test_windows_specific_netstat(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.netstat.NetStat",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 70
        expected_rows = [
            {
                "Created": "2025-03-06T17:56:53+00:00",
                "ForeignAddr": "13.107.246.254",
                "ForeignPort": 443,
                "LocalAddr": "10.0.0.4",
                "LocalPort": 49929,
                "Offset": 145201667934000,
                "Owner": "SearchApp.exe",
                "PID": 5644,
                "Proto": "TCPv4",
                "State": "CLOSE_WAIT",
            },
            {
                "Created": "2025-03-06T17:50:02+00:00",
                "ForeignAddr": "168.63.129.16",
                "ForeignPort": 80,
                "LocalAddr": "10.0.0.4",
                "LocalPort": 49688,
                "Offset": 145201778506032,
                "Owner": "WindowsAzureGu",
                "PID": 1944,
                "Proto": "TCPv4",
                "State": "ESTABLISHED",
            },
        ]
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsPESymbols:
    def test_windows_specific_pe_symbols_processes(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.pe_symbols.PESymbols",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
            pluginargs=(
                "--source",
                "processes",
                "--module",
                "ntdll.dll",
                "--symbol",
                "NtProtectVirtualMemory",
            ),
        )
        assert rc == 0
        expected_row = {
            "Address": 2089868982,
            "Module": "ntdll.dll",
            "Symbol": "NtProtectVirtualMemory",
        }

        assert test_volatility.match_output_row(expected_row, json.loads(out))

    def test_windows_specific_pe_symbols_kernel(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.pe_symbols.PESymbols",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
            pluginargs=(
                "--source",
                "kernel",
                "--module",
                "ntoskrnl.exe",
                "--symbol",
                "ZwOpenThread",
            ),
        )
        assert rc == 0
        expected_row = {
            "Address": 2152583356,
            "Module": "ntoskrnl.exe",
            "Symbol": "ZwOpenThread",
        }

        assert test_volatility.match_output_row(expected_row, json.loads(out))


class TestWindowsPoolScanner:
    def test_windows_specific_poolscanner(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.poolscanner.PoolScanner",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 4800
        assert out.find(b"_FILE_OBJECT") != -1
        assert out.find(b"_ETHREAD") != -1
        assert out.find(b"_RTL_ATOM_TABLE") != -1
        assert out.find(b"_KMUTANT") != -1


class TestWindowsPsTree:
    def test_windows_specific_pstree(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.pstree.PsTree",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 110
        expected_row = test_volatility.load_test_data(
            "windows.pstree.PsTree", "WINDOWS10_GENERIC"
        )

        assert test_volatility.match_output_row(
            expected_row, json_out, children_recursive=True
        )


class TestWindowsRegistry:
    def test_windows_specific_registry_certificates(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.registry.certificates.Certificates",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 30
        expected_row = {
            "Certificate ID": "ProtectedRoots",
            "Certificate path": "Software\\Microsoft\\SystemCertificates",
            "Certificate section": "Root",
        }
        assert test_volatility.match_output_row(expected_row, json_out)

    def test_windows_generic_registry_hivelist(self, volatility, python, image):
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.registry.hivelist.HiveList", image, volatility, python
        )
        assert rc == 0
        out = out.lower()

        not_xp = out.find(b"\\systemroot\\system32\\config\\software")
        if not_xp == -1:
            assert (
                out.find(
                    b"\\device\\harddiskvolume1\\windows\\system32\\config\\software"
                )
                != -1
            )
        assert out.count(b"\n") > 10

    def test_windows_specific_registry_hivescan(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.registry.hivescan.HiveScan",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = test_volatility.load_test_data(
            "windows.registry.hivescan.HiveScan", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)

    def test_windows_specific_registry_printkey(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.registry.printkey.PrintKey",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 450
        expected_rows = test_volatility.load_test_data(
            "windows.registry.printkey.PrintKey", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)

    def test_windows_specific_registry_userassist(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.registry.userassist.UserAssist",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 35
        expected_row = test_volatility.load_test_data(
            "windows.registry.userassist.UserAssist", "WINDOWS10_GENERIC"
        )
        assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsSessions:
    def test_windows_specific_sessions(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.sessions.Sessions",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 115
        expected_rows = test_volatility.load_test_data(
            "windows.sessions.Sessions", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsShimcacheMem:
    def test_windows_specific_shimcachemem(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.shimcachemem.ShimcacheMem",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = test_volatility.load_test_data(
            "windows.shimcachemem.ShimcacheMem", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsSSDT:
    def test_windows_specific_ssdt(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.ssdt.SSDT",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 770
        assert out.find(b"ntoskrnl") != -1
        assert out.find(b"Nt") != -1
        assert out.find(b"xHal") != -1


class TestWindowsThreads:
    def test_windows_specific_threads(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.threads.Threads",
            image,
            volatility,
            python,
        )
        assert rc == 0
        assert out.count(b"\n") > 1730


class TestWindowsTimers:
    def test_windows_specific_timers(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.timers.Timers",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = test_volatility.load_test_data(
            "windows.timers.Timers", "WINDOWSXP_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsVadInfo:
    def test_windows_specific_vadinfo(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.vadinfo.VadInfo",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
            pluginargs=("--pid", "4"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = test_volatility.load_test_data(
            "windows.vadinfo.VadInfo", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsVerInfo:
    def test_windows_specific_verinfo(self, volatility, python):
        image = WindowsSamples.WINDOWSXP_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.verinfo.VerInfo",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        assert test_volatility.count_entries_flat(json_out) > 125
        expected_row = {
            "Base": 2152558592,
            "Build": 2622,
            "Major": 5,
            "Minor": 1,
            "Name": "ntoskrnl.exe",
            "Product": 2600,
            "__children": [],
        }
        assert test_volatility.match_output_row(expected_row, json_out)


class TestWindowsVirtMap:
    def test_windows_specific_virtmap(self, volatility, python):
        image = WindowsSamples.WINDOWS10_GENERIC.value.path
        rc, out, _err = test_volatility.runvol_plugin(
            "windows.virtmap.VirtMap",
            image,
            volatility,
            python,
            globalargs=("-r", "json"),
        )
        assert rc == 0
        json_out = json.loads(out)
        expected_rows = test_volatility.load_test_data(
            "windows.virtmap.VirtMap", "WINDOWS10_GENERIC"
        )
        for expected_row in expected_rows:
            assert test_volatility.match_output_row(expected_row, json_out)