validation.py

"""Data validation and integrity checking for memory forensics artifacts"""
from typing import Dict, List, Any
from database import ForensicsDatabase


class DataValidator:
    """Validates consistency between Volatility results and database storage"""

    def __init__(self, db: ForensicsDatabase):
        self.db = db

    async def validate_dump_integrity(self, dump_id: str) -> Dict[str, Any]:
        """
        Check if dump data is complete and consistent

        Args:
            dump_id: Dump identifier

        Returns:
            Dict with 'valid' (bool) and 'issues' (list of strings)
        """
        issues = []
        warnings = []

        # Check if dump exists
        dump = await self.db.get_dump(dump_id)
        if not dump:
            issues.append(f"Dump '{dump_id}' not found in database")
            return {"valid": False, "issues": issues, "warnings": warnings}

        # Check basic data exists
        processes = await self.db.get_processes(dump_id)
        if len(processes) == 0:
            issues.append("No processes found - dump may not have been processed")

        # Check command log vs database consistency
        commands = await self.db.get_command_history(dump_id, limit=1000)

        # Validate network data
        netscan_cmds = [c for c in commands if 'netscan' in c.get('plugin_name', '').lower()]
        if netscan_cmds:
            netscan_rows = netscan_cmds[0].get('row_count', 0)
            connections = await self.db.get_network_connections(dump_id)
            db_rows = len(connections)

            if netscan_rows > 0 and db_rows == 0:
                issues.append(
                    f"Network data mismatch: Volatility netscan found {netscan_rows} "
                    f"connections but database has 0 - possible parsing error"
                )
            elif netscan_rows != db_rows:
                warnings.append(
                    f"Network data count mismatch: Volatility={netscan_rows}, Database={db_rows}"
                )

        # Validate process data
        pslist_cmds = [c for c in commands if 'pslist' in c.get('plugin_name', '').lower()]
        if pslist_cmds:
            pslist_rows = pslist_cmds[0].get('row_count', 0)
            db_rows = len(processes)

            if pslist_rows != db_rows:
                warnings.append(
                    f"Process count mismatch: Volatility={pslist_rows}, Database={db_rows}"
                )

        # Check for failed commands
        failed_cmds = [c for c in commands if not c.get('success', True)]
        if failed_cmds:
            for cmd in failed_cmds:
                error = cmd.get('error_message', 'Unknown error')
                warnings.append(
                    f"Command '{cmd.get('plugin_name')}' failed: {error}"
                )

        return {
            "valid": len(issues) == 0,
            "issues": issues,
            "warnings": warnings,
            "stats": {
                "total_commands": len(commands),
                "failed_commands": len(failed_cmds),
                "process_count": len(processes),
                "network_count": len(await self.db.get_network_connections(dump_id))
            }
        }

    async def compare_volatility_to_database(
        self,
        dump_id: str,
        volatility_results: Dict[str, int],
        database_results: Dict[str, int]
    ) -> List[str]:
        """
        Compare Volatility command results with database storage

        Args:
            dump_id: Dump identifier
            volatility_results: Dict mapping data type to Volatility row count
            database_results: Dict mapping data type to database row count

        Returns:
            List of warning messages
        """
        warnings = []

        for data_type, vol_count in volatility_results.items():
            db_count = database_results.get(data_type, 0)

            if vol_count > 0 and db_count == 0:
                warnings.append(
                    f"{data_type}: Volatility returned {vol_count} rows but 0 "
                    f"were stored in database - check for parsing errors"
                )
            elif vol_count != db_count:
                diff = vol_count - db_count
                warnings.append(
                    f"{data_type}: Count mismatch (Volatility={vol_count}, "
                    f"Database={db_count}, Difference={diff})"
                )

        return warnings

    def validate_plugin_output(
        self,
        plugin_name: str,
        results: List[Dict[str, Any]]
    ) -> Dict[str, Any]:
        """
        Validate that plugin output has expected columns

        Args:
            plugin_name: Name of the Volatility plugin
            results: List of result dictionaries

        Returns:
            Dict with 'valid' (bool) and 'missing_columns' (list)
        """
        expected_columns = self._get_expected_columns(plugin_name)

        if not results or not expected_columns:
            return {"valid": True, "missing_columns": []}

        actual_columns = set(results[0].keys())
        expected_set = set(expected_columns)
        missing = expected_set - actual_columns

        return {
            "valid": len(missing) == 0,
            "missing_columns": list(missing),
            "unexpected_columns": list(actual_columns - expected_set)
        }

    def _get_expected_columns(self, plugin_name: str) -> List[str]:
        """Get expected column names for a Volatility plugin"""
        # Map plugin names to expected columns
        column_map = {
            'NetScan': ['Offset', 'Proto', 'LocalAddr', 'LocalPort',
                       'ForeignAddr', 'ForeignPort', 'State', 'PID', 'Owner', 'Created'],
            'PsList': ['PID', 'PPID', 'ImageFileName', 'Offset', 'Threads',
                      'Handles', 'SessionId', 'Wow64', 'CreateTime', 'ExitTime'],
            'CmdLine': ['PID', 'Process', 'Args'],
            'DllList': ['PID', 'Process', 'Base', 'Size', 'Name', 'Path'],
            'Malfind': ['PID', 'Process', 'Start', 'End', 'Tag', 'Protection', 'CommitCharge', 'PrivateMemory', 'Hexdump', 'Disasm'],
        }

        # Extract plugin class name from full module path
        for key in column_map:
            if key.lower() in plugin_name.lower():
                return column_map[key]

        return []