"""Tests for modules/intel/credential_db — deduplication, bulk load, hashcat export. Validates credential ingestion, deduplication logic, export formats, and bulk operations without requiring network or bus events. """ import json import os import sqlite3 import time from unittest.mock import MagicMock, patch import pytest from modules.base import BaseModule from modules.intel.credential_db import CredentialDB, _TABLE_DDL, _INDEXES, _HASHCAT_MODES # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def cred_db(mock_bus, mock_state, tmp_path): """Return a started CredentialDB instance backed by a temp SQLite file.""" config = {"data_dir": str(tmp_path)} mod = CredentialDB(mock_bus, mock_state, config) mod.start() yield mod mod.stop() def _ingest(db, username="admin", service="smb", cred_type="ntlmv2", cred_value="hash123", domain="CORP", source_module="test", source_ip="10.0.0.5", target_ip="10.0.0.1", target_port=445, notes=""): """Helper to call _ingest_credential with defaults.""" return db._ingest_credential( source_module=source_module, source_ip=source_ip, target_ip=target_ip, target_port=target_port, service=service, username=username, domain=domain, cred_type=cred_type, cred_value=cred_value, notes=notes, ) # --------------------------------------------------------------------------- # Structure # --------------------------------------------------------------------------- class TestCredentialDBStructure: def test_is_base_module(self): assert issubclass(CredentialDB, BaseModule) def test_module_attributes(self): assert CredentialDB.name == "credential_db" assert CredentialDB.module_type == "intel" assert CredentialDB.requires_root is False # --------------------------------------------------------------------------- # Ingestion + Deduplication # --------------------------------------------------------------------------- class TestIngestion: def test_ingest_new_credential(self, cred_db): """A new credential returns True and increments count.""" result = _ingest(cred_db) assert result is True assert cred_db._ingest_count == 1 def test_ingest_empty_value_rejected(self, cred_db): """Empty cred_value is rejected.""" result = _ingest(cred_db, cred_value="") assert result is False def test_duplicate_returns_false(self, cred_db): """Inserting the same (service, username, cred_type, cred_value) deduplicates.""" _ingest(cred_db, username="admin", service="smb", cred_type="ntlmv2", cred_value="AAA") result = _ingest(cred_db, username="admin", service="smb", cred_type="ntlmv2", cred_value="AAA") assert result is False assert cred_db._dedup_count == 1 def test_duplicate_appends_notes(self, cred_db): """Duplicate credential appends new notes to existing record.""" _ingest(cred_db, username="user1", cred_value="hash1", notes="first seen") _ingest(cred_db, username="user1", cred_value="hash1", notes="seen again") creds = cred_db.get_credentials(username="user1") assert len(creds) == 1 assert "first seen" in creds[0]["notes"] assert "seen again" in creds[0]["notes"] def test_duplicate_appends_source_module(self, cred_db): """Duplicate from a different module appends source_module.""" _ingest(cred_db, username="user1", cred_value="hash1", source_module="sniffer") _ingest(cred_db, username="user1", cred_value="hash1", source_module="responder") creds = cred_db.get_credentials(username="user1") assert "sniffer" in creds[0]["source_module"] assert "responder" in creds[0]["source_module"] def test_different_cred_types_not_deduplicated(self, cred_db): """Same username+service but different cred_type are distinct entries.""" _ingest(cred_db, cred_type="ntlmv2", cred_value="hash_ntlm") _ingest(cred_db, cred_type="kerberos_tgs", cred_value="hash_kerb") creds = cred_db.get_credentials() assert len(creds) == 2 def test_auto_hashcat_mode_resolution(self, cred_db): """hashcat_mode is auto-resolved from cred_type when not provided.""" _ingest(cred_db, cred_type="ntlmv2", cred_value="somehash") creds = cred_db.get_credentials(cred_type="ntlmv2") assert len(creds) == 1 assert creds[0]["hashcat_mode"] == 5600 # --------------------------------------------------------------------------- # Bulk load (1000+ credentials) # --------------------------------------------------------------------------- class TestBulkLoad: def test_ingest_1000_unique_credentials(self, cred_db): """Ingest 1000 unique credentials without errors.""" for i in range(1000): result = _ingest( cred_db, username=f"user_{i}", cred_value=f"hash_{i:04d}", service="smb", cred_type="ntlmv2", ) assert result is True assert cred_db._ingest_count == 1000 s = cred_db.summary() assert s["total"] == 1000 assert s["unique_users"] == 1000 def test_bulk_deduplication_at_scale(self, cred_db): """Ingest 500 unique + 500 duplicates — exactly 500 stored.""" for i in range(500): _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") for i in range(500): _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") assert cred_db._dedup_count == 500 s = cred_db.summary() assert s["total"] == 500 def test_mixed_services_bulk(self, cred_db): """1200 credentials across multiple services.""" services = ["smb", "ldap", "http", "rdp", "ssh", "ftp"] for i in range(1200): svc = services[i % len(services)] _ingest(cred_db, username=f"user_{i}", cred_value=f"val_{i}", service=svc) s = cred_db.summary() assert s["total"] == 1200 assert len(s["by_service"]) == len(services) # --------------------------------------------------------------------------- # Hashcat export # --------------------------------------------------------------------------- class TestHashcatExport: def test_to_hashcat_filters_by_mode(self, cred_db): """to_hashcat(mode) only returns credentials with matching hashcat_mode.""" _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="ntlm_hash_1") _ingest(cred_db, username="u2", cred_type="kerberos_tgs", cred_value="kerb_hash_1") ntlm_output = cred_db.to_hashcat(mode=5600) assert "ntlm_hash_1" in ntlm_output assert "kerb_hash_1" not in ntlm_output kerb_output = cred_db.to_hashcat(mode=13100) assert "kerb_hash_1" in kerb_output assert "ntlm_hash_1" not in kerb_output def test_to_hashcat_all_hashable(self, cred_db): """to_hashcat(None) returns all credentials with a hashcat_mode.""" _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="h1") _ingest(cred_db, username="u2", cred_type="kerberos_tgs", cred_value="h2") _ingest(cred_db, username="u3", cred_type="ftp", cred_value="plaintext") # no hashcat mode output = cred_db.to_hashcat() assert "h1" in output assert "h2" in output assert "plaintext" not in output def test_to_hashcat_excludes_cracked(self, cred_db): """Already cracked credentials are excluded from hashcat export.""" _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="hash_uncracked") _ingest(cred_db, username="u2", cred_type="ntlmv2", cred_value="hash_cracked") creds = cred_db.get_credentials(username="u2") cred_db.update_crack_status(creds[0]["id"], "cracked", "password123") output = cred_db.to_hashcat(mode=5600) assert "hash_uncracked" in output assert "hash_cracked" not in output # --------------------------------------------------------------------------- # Crack status management # --------------------------------------------------------------------------- class TestCrackStatus: def test_update_crack_status(self, cred_db): """update_crack_status changes status and stores cracked value.""" _ingest(cred_db, username="target", cred_value="ntlm_hash") creds = cred_db.get_credentials(username="target") cred_id = creds[0]["id"] cred_db.update_crack_status(cred_id, "cracked", "P@ssw0rd!") updated = cred_db.get_credentials(username="target") assert updated[0]["crack_status"] == "cracked" assert updated[0]["cracked_value"] == "P@ssw0rd!" def test_invalid_status_ignored(self, cred_db): """Invalid status values are silently ignored.""" _ingest(cred_db, username="u1", cred_value="h1") creds = cred_db.get_credentials(username="u1") cred_db.update_crack_status(creds[0]["id"], "invalid_status") updated = cred_db.get_credentials(username="u1") assert updated[0]["crack_status"] == "uncracked" def test_bulk_update_cracked(self, cred_db): """bulk_update_cracked updates multiple credentials from hashcat results.""" for i in range(10): _ingest(cred_db, username=f"u{i}", cred_type="ntlmv2", cred_value=f"hash_{i}") results = {f"hash_{i}": f"pass_{i}" for i in range(5)} updated = cred_db.bulk_update_cracked(results) assert updated == 5 cracked = cred_db.get_cracked() assert len(cracked) == 5 def test_bulk_update_idempotent(self, cred_db): """Running bulk_update_cracked twice doesn't double-count.""" _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="hash_x") results = {"hash_x": "cracked_pass"} first = cred_db.bulk_update_cracked(results) second = cred_db.bulk_update_cracked(results) assert first == 1 assert second == 0 # --------------------------------------------------------------------------- # Export formats # --------------------------------------------------------------------------- class TestExports: def test_to_csv(self, cred_db): """CSV export includes header and all rows.""" for i in range(5): _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") csv_output = cred_db.to_csv() lines = csv_output.strip().split("\n") assert len(lines) == 6 # header + 5 data rows assert "username" in lines[0] def test_to_json(self, cred_db): """JSON export is valid and contains all records.""" for i in range(3): _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") json_output = cred_db.to_json() data = json.loads(json_output) assert len(data) == 3 assert all("username" in entry for entry in data) def test_summary(self, cred_db): """Summary returns correct counts by type and service.""" _ingest(cred_db, username="u1", service="smb", cred_type="ntlmv2", cred_value="h1") _ingest(cred_db, username="u2", service="ldap", cred_type="ntlmv2", cred_value="h2") _ingest(cred_db, username="u3", service="smb", cred_type="kerberos_tgs", cred_value="h3") s = cred_db.summary() assert s["total"] == 3 assert s["unique_users"] == 3 assert s["by_type"]["ntlmv2"] == 2 assert s["by_service"]["smb"] == 2 assert s["by_service"]["ldap"] == 1 # --------------------------------------------------------------------------- # Query helpers # --------------------------------------------------------------------------- class TestQueries: def test_get_credentials_filter_by_service(self, cred_db): _ingest(cred_db, username="u1", service="smb", cred_value="h1") _ingest(cred_db, username="u2", service="ldap", cred_value="h2") smb_creds = cred_db.get_credentials(service="smb") assert len(smb_creds) == 1 assert smb_creds[0]["service"] == "smb" def test_get_credentials_filter_by_cred_type(self, cred_db): _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="h1") _ingest(cred_db, username="u2", cred_type="kerberos_tgs", cred_value="h2") kerb = cred_db.get_credentials(cred_type="kerberos_tgs") assert len(kerb) == 1 def test_get_credentials_limit(self, cred_db): for i in range(20): _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") limited = cred_db.get_credentials(limit=5) assert len(limited) == 5 def test_get_cracked_returns_only_cracked(self, cred_db): _ingest(cred_db, username="u1", cred_value="h1") _ingest(cred_db, username="u2", cred_value="h2") creds = cred_db.get_credentials(username="u2") cred_db.update_crack_status(creds[0]["id"], "cracked", "pw") cracked = cred_db.get_cracked() assert len(cracked) == 1 assert cracked[0]["username"] == "u2"