import asyncio import json import sqlite3 import threading from datetime import UTC, datetime, timedelta, timezone from uuid import UUID import pytest from theHarvester.lib import database as database_module from theHarvester.lib.active_evidence import ActionExecution, ActiveEvidence, ArtifactReference from theHarvester.lib.asn_attribution import AsnAttributionObservation from theHarvester.lib.completed_result import CompletedResult, ResultObservation, SourceExecution from theHarvester.lib.database import ( DuplicateRunError, ResultStore, ResultStoreError, _sqlite_has_wal_reset_fix, dispose_sqlite_databases, ) from theHarvester.lib.network_evidence import ( BgpRouteObservation, PrefixOriginObservation, RpkiValidationObservation, network_observation_details, ) from theHarvester.lib.virtual_host import VirtualHostObservation RELEASED_COMPLETED_SCHEMA = """ CREATE TABLE completed_results ( run_id TEXT PRIMARY KEY, target TEXT NOT NULL, started_at TEXT NOT NULL, completed_at TEXT NOT NULL ); CREATE TABLE completed_result_items ( run_id TEXT NOT NULL REFERENCES completed_results(run_id) ON DELETE CASCADE, position INTEGER NOT NULL, kind TEXT NOT NULL, value TEXT NOT NULL, PRIMARY KEY (run_id, position), UNIQUE (run_id, kind, value) ); """ RELEASED_RESULTS_SCHEMA = """ CREATE TABLE results ( domain TEXT, resource TEXT, type TEXT, find_date DATE, source TEXT ); """ SCHEMA_V1_DISCOVERY_OBSERVATIONS = """ CREATE TABLE discovery_observations ( id INTEGER NOT NULL PRIMARY KEY, domain TEXT NOT NULL, resource TEXT NOT NULL, kind TEXT NOT NULL, discovered_on DATE NOT NULL, source TEXT NOT NULL ); PRAGMA user_version = 1; """ SCHEMA_V2_RUN_PROVENANCE = """ CREATE TABLE runs ( run_id TEXT PRIMARY KEY, target TEXT NOT NULL, started_at TEXT NOT NULL, completed_at TEXT NOT NULL ); CREATE TABLE results ( run_id TEXT NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE, position INTEGER NOT NULL, kind TEXT NOT NULL, value TEXT NOT NULL, PRIMARY KEY (run_id, position), UNIQUE (run_id, kind, value) ); CREATE TABLE executions ( run_id TEXT NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE, position INTEGER NOT NULL, producer_kind TEXT NOT NULL, name TEXT NOT NULL, status TEXT NOT NULL, duration_ms REAL NOT NULL, result_count INTEGER NOT NULL, error_type TEXT, stop_reason TEXT, PRIMARY KEY (run_id, position), UNIQUE (run_id, producer_kind, name) ); CREATE TABLE result_origins ( run_id TEXT NOT NULL, result_position INTEGER NOT NULL, execution_position INTEGER NOT NULL, PRIMARY KEY (run_id, result_position, execution_position), FOREIGN KEY (run_id, result_position) REFERENCES results(run_id, position) ON DELETE CASCADE, FOREIGN KEY (run_id, execution_position) REFERENCES executions(run_id, position) ON DELETE CASCADE ); CREATE TABLE legacy_observations ( id INTEGER PRIMARY KEY, domain TEXT, resource TEXT, kind TEXT, discovered_on DATE, source TEXT ); PRAGMA user_version = 2; """ SCHEMA_V4_URL_KINDS = ( SCHEMA_V2_RUN_PROVENANCE + """ CREATE TABLE artifacts ( run_id TEXT NOT NULL, position INTEGER NOT NULL, result_position INTEGER NOT NULL, execution_position INTEGER NOT NULL, kind TEXT NOT NULL, path TEXT NOT NULL, media_type TEXT NOT NULL, size_bytes INTEGER NOT NULL, sha256 TEXT NOT NULL, created_at TEXT NOT NULL, PRIMARY KEY (run_id, position), FOREIGN KEY (run_id, result_position) REFERENCES results(run_id, position) ON DELETE CASCADE, FOREIGN KEY (run_id, execution_position) REFERENCES executions(run_id, position) ON DELETE CASCADE ); PRAGMA user_version = 4; """ ) SCHEMA_V5_RESULT_KINDS = SCHEMA_V4_URL_KINDS + 'PRAGMA user_version = 5;' SCHEMA_V6_RESULT_KINDS = SCHEMA_V5_RESULT_KINDS + 'PRAGMA user_version = 6;' SCHEMA_V7_EVIDENCE_STATUS = ( SCHEMA_V6_RESULT_KINDS + """ ALTER TABLE runs ADD COLUMN evidence_status TEXT; PRAGMA user_version = 7; """ ) def completed_result(run_id: str = 'f047261c-0afb-4e18-89d5-28a7d977f51f') -> CompletedResult: return CompletedResult.finish( run_id=UUID(run_id), target='example.com', started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC), groups={ 'breach': ['ExampleBreach'], 'dns-recursive-finding': ['{"addresses":["192.0.2.2"],"hostname":"dev.api.example.com","parent":"api.example.com"}'], 'dns-recursive-classification': [ '{"addressability":"not-currently-addressable","addresses":[],"cnames":["missing.vendor.test"],"hostname":"unused.api.example.com","parent":"api.example.com"}' ], 'dns-recursive-summary': ['{"depth_reached":1,"query_count":24,"stop_reason":"depth-limit","zero_yield_batches":0}'], 'hostname': ['api.example.com'], 'ip': ['192.0.2.1'], 'person': ['{"firstname":"Ada","lastname":"Lovelace"}'], }, ) def screenshot_execution(completed_at: datetime) -> ActionExecution: return ActionExecution.finish( action='screenshot', status='completed', duration_ms=4.0, groups={}, artifacts=( ArtifactReference( kind='screenshot', subject_kind='hostname', subject_value='api.example.com', path='screenshots/api.example.com.png', media_type='image/png', size_bytes=3, sha256='0' * 64, created_at=completed_at, ), ), ) def vhost_observation(endpoint: str, *, status: int, control_status: int) -> VirtualHostObservation: return VirtualHostObservation.from_record( { 'type': 'vhost', 'endpoint': endpoint, 'hostname': 'admin.example.com', 'http_host': 'admin.example.com', 'tls_server_name': None, 'classification': 'distinct', 'phase': 'body', 'status': status, 'location': None, 'body_sha256': 'a' * 64, 'body_size': 5, 'body_truncated': False, 'context_phase': 'body', 'context_status': control_status, 'context_location': None, 'context_body_sha256': 'a' * 64, 'context_body_size': 5, 'context_body_truncated': False, 'control_phase': 'body', 'control_status': control_status, 'control_location': None, 'control_body_sha256': 'a' * 64, 'control_body_size': 5, 'control_body_truncated': False, 'confirmation_body_sha256': None, 'tls_verified': None, 'distinct_signals': ['status'], 'reflection_normalized': False, } ) @pytest.mark.asyncio async def test_initialization_enables_wal_when_sqlite_contains_the_reset_fix(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() with sqlite3.connect(database) as db: journal_mode = db.execute('PRAGMA journal_mode').fetchone()[0] expected_mode = 'wal' if _sqlite_has_wal_reset_fix(sqlite3.sqlite_version_info) else 'delete' assert journal_mode == expected_mode @pytest.mark.parametrize( ('version', 'expected'), [ ((3, 44, 5), False), ((3, 44, 6), True), ((3, 50, 6), False), ((3, 50, 7), True), ((3, 51, 2), False), ((3, 51, 3), True), ], ) def test_wal_reset_fix_version_boundaries(version: tuple[int, int, int], expected: bool) -> None: assert _sqlite_has_wal_reset_fix(version) is expected @pytest.mark.asyncio async def test_initialization_fails_when_runtime_connections_do_not_enforce_foreign_keys( tmp_path, monkeypatch: pytest.MonkeyPatch ) -> None: def configure_without_foreign_keys(dbapi_connection: object, _connection_record: object) -> None: dbapi_connection.isolation_level = None # type: ignore[attr-defined] monkeypatch.setattr(database_module, '_configure_sqlite_connection', configure_without_foreign_keys) store = ResultStore(tmp_path / 'stash.sqlite') with pytest.raises(RuntimeError, match='foreign-key enforcement'): await store.initialize() @pytest.mark.asyncio async def test_newer_schema_is_rejected_without_changing_journal_mode(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) with sqlite3.connect(database) as db: db.execute('PRAGMA user_version = 9') original_journal_mode = db.execute('PRAGMA journal_mode').fetchone()[0] with pytest.raises(RuntimeError, match='schema version 9 is newer than supported version 8'): await store.initialize() with sqlite3.connect(database) as db: assert db.execute('PRAGMA journal_mode').fetchone()[0] == original_journal_mode @pytest.mark.asyncio async def test_locked_database_write_does_not_block_the_event_loop(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() blocker = sqlite3.connect(database, check_same_thread=False) blocker.execute('BEGIN EXCLUSIVE') heartbeat_ran = threading.Event() heartbeat_seen_before_unlock: list[bool] = [] def unlock_database() -> None: heartbeat_seen_before_unlock.append(heartbeat_ran.is_set()) blocker.rollback() unlock_timer = threading.Timer(0.2, unlock_database) unlock_timer.start() try: write = asyncio.create_task(store.record_observations('example.com', ['api.example.com'], 'hostname', 'crtsh')) heartbeat = asyncio.create_task(asyncio.sleep(0, result=None)) heartbeat.add_done_callback(lambda _task: heartbeat_ran.set()) await asyncio.gather(write, heartbeat) finally: unlock_timer.join() blocker.close() assert heartbeat_seen_before_unlock == [True] @pytest.mark.asyncio async def test_schema_enforces_foreign_keys_and_cascades_results(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() result = completed_result() await store.save_run(result) with sqlite3.connect(database) as db: db.execute('PRAGMA foreign_keys = ON') assert db.execute('PRAGMA foreign_keys').fetchone()[0] == 1 db.execute('DELETE FROM runs WHERE run_id = ?', (str(result.run_id),)) db.commit() assert db.execute('SELECT COUNT(*) FROM results').fetchone()[0] == 0 @pytest.mark.asyncio async def test_completed_result_round_trip_preserves_legacy_observations(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() await store.record_observations('example.com', ['legacy.example.com'], 'hostname', 'legacy-source') result = completed_result() await store.save_run(result) assert await store.load_run(result.run_id) == result with sqlite3.connect(database) as db: stored = db.execute('SELECT domain, resource, kind, source FROM legacy_observations').fetchall() stored_items = set(db.execute('SELECT kind, value FROM results').fetchall()) run_types = {row[1]: row[2] for row in db.execute('PRAGMA table_info(runs)')} result_types = {row[1]: row[2] for row in db.execute('PRAGMA table_info(results)')} jsonl_items = {(record['type'], record['value']) for line in result.jsonl().splitlines()[1:] if (record := json.loads(line))} assert stored == [('example.com', 'legacy.example.com', 'hostname', 'legacy-source')] assert stored_items == jsonl_items assert run_types == { 'run_id': 'TEXT', 'target': 'TEXT', 'started_at': 'TEXT', 'completed_at': 'TEXT', 'evidence_status': 'TEXT', } assert result_types == { 'run_id': 'TEXT', 'position': 'INTEGER', 'kind': 'TEXT', 'value': 'TEXT', 'details_json': 'TEXT', } @pytest.mark.asyncio async def test_structured_vhost_evidence_round_trips_in_the_results_table(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() completed_at = datetime(2026, 8, 9, 12, 1, tzinfo=UTC) first = vhost_observation('http://192.0.2.10', status=200, control_status=404) second = vhost_observation('http://192.0.2.11', status=201, control_status=404) result = CompletedResult.finish( run_id=UUID('3ac7bba2-45da-4cdd-96a4-019a4d42bca4'), target='example.com', started_at=completed_at, completed_at=completed_at, groups={}, source_executions=(SourceExecution('crtsh', 'completed', 4.0, 1),), observations=(ResultObservation('crtsh', 'hostname', 'admin.example.com'),), active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='vhost', status='completed', duration_ms=12.5, groups={'hostname': ['admin.example.com']}, ), ) ), virtual_hosts=(second, first), ) await store.save_run(result) assert await store.load_run(result.run_id) == result with sqlite3.connect(database) as db: schema_version = db.execute('PRAGMA user_version').fetchone()[0] stored_result = db.execute( 'SELECT kind, value, details_json FROM results WHERE run_id = ?', (str(result.run_id),), ).fetchone() executions = db.execute( 'SELECT producer_kind, name, result_count FROM executions WHERE run_id = ? ORDER BY producer_kind', (str(result.run_id),), ).fetchall() origin_count = db.execute( 'SELECT COUNT(*) FROM result_origins WHERE run_id = ?', (str(result.run_id),), ).fetchone()[0] assert schema_version == 8 assert stored_result[:2] == ('hostname', 'admin.example.com') assert json.loads(stored_result[2]) == [ {key: value for key, value in observation.to_record().items() if key not in {'type', 'hostname'}} for observation in (first, second) ] assert executions == [('action', 'vhost', 1), ('source', 'crtsh', 1)] assert origin_count == 2 @pytest.mark.asyncio async def test_structured_network_evidence_round_trips_in_the_results_table(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() collected_at = datetime(2026, 8, 11, 12, 1, tzinfo=UTC) observed_at = datetime(2026, 8, 11, 12, 0, tzinfo=UTC) network_observations = ( PrefixOriginObservation('routeviews', '198.51.100.7/24', 64500, collected_at), BgpRouteObservation( 'routeviews', '198.51.100.0/24', 'AS64500', 'route-views.test', 64496, '192.0.2.7', '64496 64500', '', observed_at, collected_at, ), RpkiValidationObservation( 'routeviews', '198.51.100.0/24', 64500, 'not-found', observed_at, collected_at, ), ) result = CompletedResult.finish( run_id=UUID('6db84c57-e459-4a5a-95c5-1c231a160ba6'), target='example.com', started_at=observed_at, completed_at=collected_at, groups={'asn': ['AS64500']}, active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='routeviews', status='completed', duration_ms=15, groups={'prefix': ['198.51.100.0/24']}, ), ) ), network_observations=network_observations, ) await store.save_run(result) assert await store.load_run(result.run_id) == result with sqlite3.connect(database) as db: schema_version = db.execute('PRAGMA user_version').fetchone()[0] stored_details = db.execute( "SELECT details_json FROM results WHERE run_id = ? AND kind = 'prefix'", (str(result.run_id),), ).fetchone()[0] assert schema_version == 8 assert json.loads(stored_details) == network_observation_details(network_observations) @pytest.mark.asyncio async def test_shodan_host_evidence_round_trips_without_a_json_string_value(tmp_path) -> None: from theHarvester.lib.shodan_evidence import ShodanHostObservation database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() collected_at = datetime(2026, 8, 14, 12, 2, tzinfo=UTC) shodan_host = ShodanHostObservation.from_record( '192.0.2.10', { 'asn': 'AS64496', 'organization': 'Example Transit', 'services': [ {'port': 53, 'transport': 'udp', 'product': 'dnsmasq'}, {'port': 443, 'transport': 'tcp', 'product': 'nginx'}, ], }, ) result = CompletedResult.finish( target='example.com', started_at=collected_at, completed_at=collected_at, groups={}, source_executions=(SourceExecution('shodan', 'completed', 1, 1),), observations=(ResultObservation('shodan', 'shodan-host', '192.0.2.10'),), active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='shodan', status='completed', duration_ms=1, groups={'shodan-host': ['192.0.2.10']}, ), ) ), shodan_hosts=(shodan_host,), ) await store.save_run(result) assert await store.load_run(result.run_id) == result with sqlite3.connect(database) as db: stored = db.execute( 'SELECT kind, value, details_json FROM results WHERE run_id = ?', (str(result.run_id),), ).fetchone() origins = db.execute( 'SELECT COUNT(*) FROM result_origins WHERE run_id = ?', (str(result.run_id),), ).fetchone()[0] assert stored[:2] == ('shodan-host', '192.0.2.10') assert json.loads(stored[2]) == shodan_host.to_details() assert origins == 2 @pytest.mark.asyncio async def test_takeover_evidence_round_trips_with_structured_details(tmp_path) -> None: from theHarvester.lib.takeover_evidence import TakeoverCandidateOutcome database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() collected_at = datetime(2026, 8, 15, 12, 2, tzinfo=UTC) outcome = TakeoverCandidateOutcome.from_record( 'bucket.example.test', { 'status': 'indicator', 'dns': [ { 'resolver': '1.1.1.1', 'cname_chain': ['missing-bucket.s3.amazonaws.com'], 'terminal_rcode': 'NOERROR', } ], 'wildcard_dns': [ { 'resolver': '1.1.1.1', 'cname_chain': [], 'terminal_rcode': 'NXDOMAIN', } ], 'http': [{'scheme': 'https', 'status': 404}], 'indicators': [ { 'classification': 'vulnerable-indicator', 'service': 'AWS/S3', 'rule_id': 'aws-s3', 'rule_revision': 'takeover-rules-v1', 'scheme': 'https', 'matched': ['body:BucketName', 'body:The specified bucket does not exist'], } ], 'error_types': [], }, ) no_indicator = TakeoverCandidateOutcome.from_record( 'live.example.test', { 'status': 'no-indicator', 'dns': [ { 'resolver': '1.1.1.1', 'cname_chain': [], 'terminal_rcode': 'NOERROR', } ], 'wildcard_dns': [], 'http': [], 'indicators': [], 'error_types': [], }, ) inconclusive = TakeoverCandidateOutcome.from_record( 'uncertain.example.test', { 'status': 'inconclusive', 'dns': [ { 'resolver': '1.1.1.1', 'cname_chain': [], 'terminal_rcode': 'ERROR', 'error_type': 'DNSTimeoutError', } ], 'wildcard_dns': [], 'http': [], 'indicators': [], 'error_types': ['DNSTimeoutError'], }, ) outcomes = (outcome, no_indicator, inconclusive) result = CompletedResult.finish( target='example.test', started_at=collected_at, completed_at=collected_at, groups={}, active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='takeover', status='completed', duration_ms=1, groups={'takeover': [item.hostname for item in outcomes]}, ), ) ), takeover_outcomes=outcomes, ) await store.save_run(result) assert await store.load_run(result.run_id) == result with sqlite3.connect(database) as db: stored = db.execute( 'SELECT kind, value, details_json FROM results WHERE run_id = ? ORDER BY value', (str(result.run_id),), ).fetchall() assert stored == [ ('takeover', item.hostname, json.dumps(item.to_details(), separators=(',', ':'), sort_keys=True)) for item in outcomes ] @pytest.mark.asyncio async def test_asn_organization_attribution_round_trips_in_a_normalized_table(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() collected_at = datetime(2026, 8, 12, 12, 1, tzinfo=UTC) attribution = AsnAttributionObservation( 'source', 'urlscan', 'AS64500', 'Example Network', 'ip', '192.0.2.10', collected_at, ) result = CompletedResult.finish( target='example.com', started_at=collected_at, completed_at=collected_at, groups={'asn': ['AS64500'], 'ip': ['192.0.2.10']}, source_executions=(SourceExecution('urlscan', 'completed', 1, 2),), observations=( ResultObservation('urlscan', 'asn', 'AS64500'), ResultObservation('urlscan', 'ip', '192.0.2.10'), ), asn_attributions=(attribution,), ) await store.save_run(result) assert await store.load_run(result.run_id) == result with sqlite3.connect(database) as db: row = db.execute( 'SELECT organization_label, collected_at, asn_result_position, subject_result_position, ' 'execution_position FROM asn_attributions WHERE run_id = ?', (str(result.run_id),), ).fetchone() result_rows = db.execute( 'SELECT position, kind, value FROM results WHERE run_id = ? ORDER BY position', (str(result.run_id),), ).fetchall() assert result_rows == [(0, 'asn', 'AS64500'), (1, 'ip', '192.0.2.10')] assert row == ('Example Network', '2026-08-12T12:01:00+00:00', 0, 1, 0) @pytest.mark.asyncio @pytest.mark.parametrize('corruption', ['non-asn-result', 'duplicate-row']) async def test_loading_corrupt_asn_attribution_fails_closed(tmp_path, corruption: str) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() collected_at = datetime(2026, 8, 12, 12, 1, tzinfo=UTC) attribution = AsnAttributionObservation('source', 'urlscan', 'AS64500', 'Example Network', 'ip', '192.0.2.10', collected_at) result = CompletedResult.finish( target='example.com', started_at=collected_at, completed_at=collected_at, groups={'asn': ['AS64500'], 'ip': ['192.0.2.10']}, source_executions=(SourceExecution('urlscan', 'completed', 1, 2),), observations=( ResultObservation('urlscan', 'asn', 'AS64500'), ResultObservation('urlscan', 'ip', '192.0.2.10'), ), asn_attributions=(attribution,), ) await store.save_run(result) with sqlite3.connect(database) as db: if corruption == 'non-asn-result': db.execute( 'UPDATE asn_attributions SET asn_result_position = subject_result_position WHERE run_id = ?', (str(result.run_id),), ) else: db.execute( 'INSERT INTO asn_attributions ' '(run_id, position, asn_result_position, subject_result_position, execution_position, ' 'organization_label, collected_at) ' 'SELECT run_id, position + 1, asn_result_position, subject_result_position, execution_position, ' 'organization_label, collected_at FROM asn_attributions WHERE run_id = ?', (str(result.run_id),), ) db.commit() with pytest.raises(ResultStoreError, match='Persisted ASN attribution is invalid'): await store.load_run(result.run_id) @pytest.mark.asyncio async def test_loading_prefix_details_with_vhost_provenance_fails_closed(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() collected_at = datetime(2026, 8, 11, 12, 1, tzinfo=UTC) result = CompletedResult.finish( target='example.com', started_at=collected_at, completed_at=collected_at, groups={'asn': ['AS64500']}, active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='routeviews', status='completed', duration_ms=1, groups={'prefix': ['198.51.100.0/24']}, ), ) ), network_observations=(PrefixOriginObservation('routeviews', '198.51.100.0/24', 'AS64500', collected_at),), ) await store.save_run(result) with sqlite3.connect(database) as db: db.execute( "UPDATE executions SET name = 'vhost' WHERE run_id = ? AND name = 'routeviews'", (str(result.run_id),), ) db.commit() with pytest.raises(ResultStoreError, match=r'Persisted virtual-host details are missing: 198\.51\.100\.0/24'): await store.load_run(result.run_id) @pytest.mark.asyncio async def test_schema_v7_migrates_vhost_collision_without_losing_references(tmp_path) -> None: database = tmp_path / 'stash.sqlite' run_id = UUID('750ab571-778d-490e-b760-70394d936eb4') with sqlite3.connect(database) as db: db.executescript(SCHEMA_V7_EVIDENCE_STATUS) db.execute( 'INSERT INTO runs (run_id, target, started_at, completed_at) VALUES (?, ?, ?, ?)', (str(run_id), 'example.com', '2026-08-09T12:00:00+00:00', '2026-08-09T12:01:00+00:00'), ) db.execute( 'INSERT INTO results (run_id, position, kind, value) VALUES (?, ?, ?, ?)', (str(run_id), 0, 'hostname', 'admin.example.com'), ) db.execute( 'INSERT INTO results (run_id, position, kind, value) VALUES (?, ?, ?, ?)', (str(run_id), 1, 'vhost', 'admin.example.com'), ) db.execute( 'INSERT INTO executions ' '(run_id, position, producer_kind, name, status, duration_ms, result_count) ' 'VALUES (?, ?, ?, ?, ?, ?, ?)', (str(run_id), 0, 'action', 'vhost', 'completed', 1.0, 2), ) db.executemany( 'INSERT INTO result_origins (run_id, result_position, execution_position) VALUES (?, ?, ?)', [(str(run_id), position, 0) for position in (0, 1)], ) db.execute( 'INSERT INTO artifacts ' '(run_id, position, result_position, execution_position, kind, path, media_type, size_bytes, sha256, created_at) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', ( str(run_id), 0, 1, 0, 'screenshot', 'screenshots/admin.example.com.png', 'image/png', 3, '0' * 64, '2026-08-09T12:01:00+00:00', ), ) store = ResultStore(database) await store.initialize() await store.dispose() store = ResultStore(database) await store.initialize() with pytest.raises(ResultStoreError, match='virtual-host details'): await store.load_run(run_id) with sqlite3.connect(database) as db: result_columns = [row[1] for row in db.execute('PRAGMA table_info(results)')] stored = db.execute('SELECT kind, value, details_json FROM results').fetchall() execution = db.execute('SELECT name, result_count FROM executions').fetchone() origins = db.execute('SELECT result_position, execution_position FROM result_origins').fetchall() artifacts = db.execute('SELECT result_position, execution_position, path FROM artifacts').fetchall() schema_version = db.execute('PRAGMA user_version').fetchone()[0] assert result_columns == ['run_id', 'position', 'kind', 'value', 'details_json'] assert stored == [('hostname', 'admin.example.com', None)] assert execution == ('vhost', 1) assert origins == [(0, 0)] assert artifacts == [(0, 0, 'screenshots/admin.example.com.png')] assert schema_version == 8 @pytest.mark.asyncio async def test_schema_v7_migrates_unstructured_vhost_without_action_origin_to_hostname(tmp_path) -> None: database = tmp_path / 'stash.sqlite' run_id = UUID('20f4b762-e9cb-4e10-96c6-85f982af50b3') with sqlite3.connect(database) as db: db.executescript(SCHEMA_V7_EVIDENCE_STATUS) db.execute( 'INSERT INTO runs (run_id, target, started_at, completed_at) VALUES (?, ?, ?, ?)', (str(run_id), 'example.com', '2026-08-09T12:00:00+00:00', '2026-08-09T12:01:00+00:00'), ) db.execute( 'INSERT INTO results (run_id, position, kind, value) VALUES (?, ?, ?, ?)', (str(run_id), 0, 'vhost', 'admin.example.com'), ) store = ResultStore(database) await store.initialize() loaded = await store.load_run(run_id) assert loaded.results == (('hostname', 'admin.example.com'),) assert loaded.virtual_hosts == () @pytest.mark.asyncio @pytest.mark.parametrize('details_json', [None, 'not-json', '[]', '[{"endpoint":"http://192.0.2.10:80/"}]']) async def test_loading_malformed_vhost_details_fails_closed(tmp_path, details_json: str | None) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() result = CompletedResult.finish( target='example.com', started_at=datetime(2026, 8, 9, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 9, 12, 1, tzinfo=UTC), groups={}, active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='vhost', status='completed', duration_ms=1, groups={'hostname': ['admin.example.com']}, ), ) ), virtual_hosts=(vhost_observation('http://192.0.2.10', status=200, control_status=404),), ) await store.save_run(result) with sqlite3.connect(database) as db: db.execute( 'UPDATE results SET details_json = ? WHERE run_id = ?', (details_json, str(result.run_id)), ) db.commit() with pytest.raises(ResultStoreError, match='virtual-host details'): await store.load_run(result.run_id) @pytest.mark.asyncio async def test_loading_vhost_details_without_vhost_provenance_preserves_compatibility_error(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() result = CompletedResult.finish( target='example.com', started_at=datetime(2026, 8, 9, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 9, 12, 1, tzinfo=UTC), groups={}, active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='vhost', status='completed', duration_ms=1, groups={'hostname': ['admin.example.com']}, ), ) ), virtual_hosts=(vhost_observation('http://192.0.2.10', status=200, control_status=404),), ) await store.save_run(result) with sqlite3.connect(database) as db: db.execute('DELETE FROM result_origins WHERE run_id = ?', (str(result.run_id),)) db.commit() with pytest.raises( ResultStoreError, match='Persisted virtual-host details require hostname results with vhost provenance', ): await store.load_run(result.run_id) @pytest.mark.asyncio async def test_loading_oversized_network_details_fails_before_json_decode(tmp_path, monkeypatch) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() collected_at = datetime(2026, 8, 11, 12, 1, tzinfo=UTC) result = CompletedResult.finish( target='example.com', started_at=datetime(2026, 8, 11, 12, 0, tzinfo=UTC), completed_at=collected_at, groups={'asn': ['AS64500']}, active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='routeviews', status='completed', duration_ms=1, groups={'prefix': ['192.0.2.0/24']}, ), ) ), network_observations=(PrefixOriginObservation('routeviews', '192.0.2.0/24', 'AS64500', collected_at),), ) await store.save_run(result) with sqlite3.connect(database) as db: db.execute( 'UPDATE results SET details_json = ? WHERE run_id = ? AND kind = ?', ('[]', str(result.run_id), 'prefix'), ) db.commit() monkeypatch.setattr('theHarvester.lib.network_evidence.MAX_NETWORK_DETAILS_BYTES', 1) with pytest.raises(ResultStoreError, match='Persisted network details are invalid'): await store.load_run(result.run_id) @pytest.mark.asyncio @pytest.mark.parametrize('evidence_status', ['partial', 'failed']) async def test_sparse_evidence_status_survives_result_store_round_trip(tmp_path, evidence_status: str) -> None: store = ResultStore(tmp_path / 'stash.sqlite') await store.initialize() result = CompletedResult.finish( target='example.com', started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC), groups={}, evidence_status=evidence_status, ) await store.save_run(result) loaded = await store.load_run(result.run_id) assert loaded == result assert loaded.status == evidence_status @pytest.mark.asyncio async def test_completed_result_round_trip_preserves_source_provenance(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() result = CompletedResult.finish( run_id=UUID('9c024fb7-4877-4f6e-89ef-0bf6af59ade0'), target='example.com', started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC), groups={'hostname': ['api.example.com', 'mail.example.com']}, observations=( ResultObservation('crtsh', 'hostname', 'api.example.com'), ResultObservation('certspotter', 'hostname', 'api.example.com'), ResultObservation('crtsh', 'hostname', 'mail.example.com'), ), source_executions=( SourceExecution('crtsh', 'completed', 12.5, 2), SourceExecution('certspotter', 'completed', 8.0, 1), ), ) await store.save_run(result) assert await store.load_run(result.run_id) == result with sqlite3.connect(database) as db: observations = db.execute( 'SELECT o.run_id, e.name, r.kind, r.value ' 'FROM result_origins AS o ' 'JOIN executions AS e ON e.run_id = o.run_id AND e.position = o.execution_position ' 'JOIN results AS r ON r.run_id = o.run_id AND r.position = o.result_position ' 'ORDER BY e.name, r.value' ).fetchall() executions = db.execute( 'SELECT run_id, producer_kind, name, status, result_count FROM executions ORDER BY position' ).fetchall() assert observations == [ (str(result.run_id), 'certspotter', 'hostname', 'api.example.com'), (str(result.run_id), 'crtsh', 'hostname', 'api.example.com'), (str(result.run_id), 'crtsh', 'hostname', 'mail.example.com'), ] assert executions == [ (str(result.run_id), 'source', 'crtsh', 'completed', 2), (str(result.run_id), 'source', 'certspotter', 'completed', 1), ] @pytest.mark.asyncio async def test_mixed_source_action_artifact_round_trip_uses_unified_tables(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() completed_at = datetime(2026, 8, 9, 12, 1, tzinfo=UTC) result = CompletedResult.finish( run_id=UUID('d721f4c5-1c76-4e7a-904a-23c5d6755834'), target='example.com', started_at=datetime(2026, 8, 9, 12, 0, tzinfo=UTC), completed_at=completed_at, groups={'hostname': ['api.example.com']}, source_executions=(SourceExecution('shared-name', 'completed', 2.0, 1),), observations=(ResultObservation('shared-name', 'hostname', 'api.example.com'),), active_evidence=ActiveEvidence( executions=( ActionExecution.finish( action='shared-name', status='completed', duration_ms=3.0, groups={'ip': ['192.0.2.10']}, ), screenshot_execution(completed_at), ActionExecution.finish( action='takeover', status='completed', duration_ms=1.0, groups={}, ), ) ), ) await store.save_run(result) assert await store.load_run(result.run_id) == result assert [item.to_dict() for item in await store.action_yields(result.run_id)] == [ { 'action': 'screenshot', 'observed_result_count': 0, 'unique_result_count': 0, 'shared_result_count': 0, }, { 'action': 'shared-name', 'observed_result_count': 1, 'unique_result_count': 1, 'shared_result_count': 0, }, { 'action': 'takeover', 'observed_result_count': 0, 'unique_result_count': 0, 'shared_result_count': 0, }, ] with sqlite3.connect(database) as db: tables = { row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'") } executions = db.execute('SELECT position, producer_kind, name, result_count FROM executions ORDER BY position').fetchall() origins = db.execute( 'SELECT e.producer_kind, e.name, r.kind, r.value ' 'FROM result_origins AS o ' 'JOIN executions AS e ON e.run_id = o.run_id AND e.position = o.execution_position ' 'JOIN results AS r ON r.run_id = o.run_id AND r.position = o.result_position ' 'ORDER BY e.producer_kind, e.name' ).fetchall() artifacts = db.execute( 'SELECT e.name, r.kind, r.value, a.kind, a.path, a.media_type, a.size_bytes, a.sha256, a.created_at ' 'FROM artifacts AS a ' 'JOIN executions AS e ON e.run_id = a.run_id AND e.position = a.execution_position ' 'JOIN results AS r ON r.run_id = a.run_id AND r.position = a.result_position' ).fetchall() assert tables == { 'runs', 'executions', 'results', 'result_origins', 'asn_attributions', 'artifacts', 'legacy_observations', 'run_records', 'run_worker_leases', } assert executions == [ (0, 'source', 'shared-name', 1), (1, 'action', 'shared-name', 1), (2, 'action', 'screenshot', 0), (3, 'action', 'takeover', 0), ] assert origins == [ ('action', 'shared-name', 'ip', '192.0.2.10'), ('source', 'shared-name', 'hostname', 'api.example.com'), ] assert artifacts == [ ( 'screenshot', 'hostname', 'api.example.com', 'screenshot', 'screenshots/api.example.com.png', 'image/png', 3, '0' * 64, completed_at.isoformat(), ) ] with sqlite3.connect(database) as db: db.execute('PRAGMA foreign_keys = ON') db.execute('DELETE FROM runs WHERE run_id = ?', (str(result.run_id),)) db.commit() assert db.execute('SELECT COUNT(*) FROM executions').fetchone()[0] == 0 assert db.execute('SELECT COUNT(*) FROM results').fetchone()[0] == 0 assert db.execute('SELECT COUNT(*) FROM result_origins').fetchone()[0] == 0 assert db.execute('SELECT COUNT(*) FROM artifacts').fetchone()[0] == 0 @pytest.mark.asyncio async def test_source_yields_distinguish_unique_and_shared_results(tmp_path) -> None: store = ResultStore(tmp_path / 'stash.sqlite') await store.initialize() result = CompletedResult.finish( run_id=UUID('bb2e9a76-f7fc-4eec-acbc-6da55a389d88'), target='example.com', started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC), groups={'hostname': ['api.example.com', 'mail.example.com', 'www.example.com']}, observations=( ResultObservation('crtsh', 'hostname', 'api.example.com'), ResultObservation('certspotter', 'hostname', 'api.example.com'), ResultObservation('crtsh', 'hostname', 'mail.example.com'), ResultObservation('certspotter', 'hostname', 'www.example.com'), ), source_executions=( SourceExecution('crtsh', 'completed', 12.5, 2), SourceExecution('certspotter', 'completed', 8.0, 2), SourceExecution('empty-source', 'completed', 5.0, 0), ), active_evidence=ActiveEvidence( ( ActionExecution.finish( action='dns-resolve', status='completed', duration_ms=4, groups={'hostname': ['api.example.com', 'mail.example.com']}, ), ) ), ) await store.save_run(result) yields = await store.source_yields(result.run_id) assert [item.to_dict() for item in yields] == [ { 'source': 'certspotter', 'observed_result_count': 2, 'unique_result_count': 1, 'shared_result_count': 1, 'resolved_hostname_count': 1, 'unique_resolved_hostname_count': 0, }, { 'source': 'crtsh', 'observed_result_count': 2, 'unique_result_count': 1, 'shared_result_count': 1, 'resolved_hostname_count': 2, 'unique_resolved_hostname_count': 1, }, { 'source': 'empty-source', 'observed_result_count': 0, 'unique_result_count': 0, 'shared_result_count': 0, 'resolved_hostname_count': 0, 'unique_resolved_hostname_count': 0, }, ] @pytest.mark.asyncio async def test_source_yields_can_measure_hostname_contributions_only(tmp_path) -> None: store = ResultStore(tmp_path / 'stash.sqlite') await store.initialize() result = CompletedResult.finish( run_id=UUID('86eafad7-3308-4786-8e85-1e15b24afafc'), target='example.com', started_at=datetime(2026, 8, 23, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 23, 12, 1, tzinfo=UTC), groups={'hostname': ['shared.example.com'], 'url': ['https://unique.example.com/']}, observations=( ResultObservation('first', 'hostname', 'shared.example.com'), ResultObservation('second', 'hostname', 'shared.example.com'), ResultObservation('first', 'url', 'https://unique.example.com/'), ), source_executions=( SourceExecution('first', 'completed', 10, 2), SourceExecution('second', 'completed', 10, 1), ), ) await store.save_run(result) yields = await store.source_yields(result.run_id, kind='hostname') assert [item.to_dict() for item in yields] == [ { 'source': 'first', 'observed_result_count': 1, 'unique_result_count': 0, 'shared_result_count': 1, 'resolved_hostname_count': 0, 'unique_resolved_hostname_count': 0, }, { 'source': 'second', 'observed_result_count': 1, 'unique_result_count': 0, 'shared_result_count': 1, 'resolved_hostname_count': 0, 'unique_resolved_hostname_count': 0, }, ] @pytest.mark.asyncio async def test_existing_completed_records_survive_initialization(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) existing = completed_result() with sqlite3.connect(database) as db: db.executescript(RELEASED_RESULTS_SCHEMA + RELEASED_COMPLETED_SCHEMA) db.execute( 'INSERT INTO results (domain, resource, type, find_date, source) VALUES (?, ?, ?, ?, ?)', ('example.com', 'legacy.example.com', 'host', '2026-08-08', 'crtsh'), ) db.execute( 'INSERT INTO completed_results (run_id, target, started_at, completed_at) VALUES (?, ?, ?, ?)', (str(existing.run_id), existing.target, existing.started_at.isoformat(), existing.completed_at.isoformat()), ) db.executemany( 'INSERT INTO completed_result_items (run_id, position, kind, value) VALUES (?, ?, ?, ?)', [(str(existing.run_id), position, kind, value) for position, (kind, value) in enumerate(existing.results)], ) await store.initialize() assert await store.load_run(existing.run_id) == existing with sqlite3.connect(database) as db: tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} migrated = db.execute('SELECT resource, kind, source FROM legacy_observations').fetchall() assert {'completed_results', 'completed_result_items', 'legacy_results'}.isdisjoint(tables) assert migrated == [('legacy.example.com', 'hostname', 'crtsh')] @pytest.mark.asyncio async def test_current_schema_reopens_without_running_legacy_migration(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) existing = completed_result() await store.initialize() await store.save_run(existing) await dispose_sqlite_databases() reopened = ResultStore(database) await reopened.initialize() assert await reopened.load_run(existing.run_id) == existing @pytest.mark.asyncio async def test_schema_v2_upgrades_to_v8_without_rewriting_existing_rows(tmp_path) -> None: database = tmp_path / 'stash.sqlite' run_id = UUID('251d4047-190b-4a4d-9c4e-9eed3f23c8c7') with sqlite3.connect(database) as db: db.executescript(SCHEMA_V2_RUN_PROVENANCE) db.execute( 'INSERT INTO runs (run_id, target, started_at, completed_at) VALUES (?, ?, ?, ?)', ( str(run_id), 'example.com', '2026-08-09T12:00:00+00:00', '2026-08-09T12:01:00+00:00', ), ) db.execute( 'INSERT INTO results (run_id, position, kind, value) VALUES (?, ?, ?, ?)', (str(run_id), 0, 'hostname', 'api.example.com'), ) db.execute( 'INSERT INTO executions ' '(run_id, position, producer_kind, name, status, duration_ms, result_count) ' 'VALUES (?, ?, ?, ?, ?, ?, ?)', (str(run_id), 0, 'source', 'crtsh', 'completed', 12.5, 1), ) db.execute( 'INSERT INTO result_origins (run_id, result_position, execution_position) VALUES (?, ?, ?)', (str(run_id), 0, 0), ) store = ResultStore(database) await store.initialize() await store.initialize() loaded = await store.load_run(run_id) assert loaded == CompletedResult.finish( run_id=run_id, target='example.com', started_at=datetime(2026, 8, 9, 12, 0, tzinfo=UTC), completed_at=datetime(2026, 8, 9, 12, 1, tzinfo=UTC), groups={'hostname': ['api.example.com']}, source_executions=(SourceExecution('crtsh', 'completed', 12.5, 1),), observations=(ResultObservation('crtsh', 'hostname', 'api.example.com'),), ) with sqlite3.connect(database) as db: assert db.execute('PRAGMA user_version').fetchone()[0] == 8 assert db.execute('SELECT COUNT(*) FROM artifacts').fetchone()[0] == 0 @pytest.mark.asyncio async def test_schema_v6_adds_nullable_evidence_status_without_rewriting_execution_status(tmp_path) -> None: database = tmp_path / 'stash.sqlite' run_id = UUID('cb34987b-9dd5-44f5-a58a-7ca7d34b0743') with sqlite3.connect(database) as db: db.executescript(SCHEMA_V6_RESULT_KINDS) db.execute( 'INSERT INTO runs (run_id, target, started_at, completed_at) VALUES (?, ?, ?, ?)', ( str(run_id), 'example.com', '2026-08-09T12:00:00+00:00', '2026-08-09T12:01:00+00:00', ), ) db.execute( 'INSERT INTO executions ' '(run_id, position, producer_kind, name, status, duration_ms, result_count) ' 'VALUES (?, ?, ?, ?, ?, ?, ?)', (str(run_id), 0, 'source', 'crtsh', 'partial', 12.5, 0), ) store = ResultStore(database) await store.initialize() loaded = await store.load_run(run_id) with sqlite3.connect(database) as db: run_columns = {row[1] for row in db.execute('PRAGMA table_info(runs)')} stored_status = db.execute('SELECT evidence_status FROM runs WHERE run_id = ?', (str(run_id),)).fetchone()[0] schema_version = db.execute('PRAGMA user_version').fetchone()[0] assert loaded.status == 'partial' assert stored_status is None assert 'evidence_status' in run_columns assert schema_version == 8 @pytest.mark.asyncio async def test_schema_v4_merges_deprecated_url_kinds_without_losing_origins(tmp_path) -> None: database = tmp_path / 'stash.sqlite' run_id = UUID('d299651b-21c1-4511-8cac-63ba70f926f4') target_url = 'https://portal.example.com/login' with sqlite3.connect(database) as db: db.executescript(SCHEMA_V4_URL_KINDS) db.execute( 'INSERT INTO runs (run_id, target, started_at, completed_at) VALUES (?, ?, ?, ?)', (str(run_id), 'example.com', '2026-08-09T12:00:00+00:00', '2026-08-09T12:01:00+00:00'), ) db.executemany( 'INSERT INTO results (run_id, position, kind, value) VALUES (?, ?, ?, ?)', [ (str(run_id), 0, 'api-endpoint', target_url), (str(run_id), 1, 'hostname', 'portal.example.com'), (str(run_id), 2, 'interesting-url', target_url), (str(run_id), 3, 'linkedin-link', target_url), (str(run_id), 4, 'url', target_url), ], ) db.executemany( 'INSERT INTO executions ' '(run_id, position, producer_kind, name, status, duration_ms, result_count, error_type, stop_reason) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', [ (str(run_id), 0, 'source', 'builtwith', 'completed', 1.0, 1, None, None), (str(run_id), 1, 'source', 'rocketreach', 'completed', 1.0, 1, None, None), (str(run_id), 2, 'source', 'gitlab', 'completed', 1.0, 1, None, None), (str(run_id), 3, 'action', 'api-scan', 'completed', 1.0, 3, None, None), (str(run_id), 4, 'action', 'screenshot', 'completed', 1.0, 0, None, None), ], ) db.executemany( 'INSERT INTO result_origins (run_id, result_position, execution_position) VALUES (?, ?, ?)', [ (str(run_id), 2, 0), (str(run_id), 3, 1), (str(run_id), 4, 2), (str(run_id), 0, 3), (str(run_id), 2, 3), (str(run_id), 4, 3), ], ) db.execute( 'INSERT INTO artifacts ' '(run_id, position, result_position, execution_position, kind, path, media_type, size_bytes, sha256, created_at) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', ( str(run_id), 0, 2, 4, 'screenshot', 'screenshots/portal.png', 'image/png', 3, '0' * 64, '2026-08-09T12:01:00+00:00', ), ) db.executemany( 'INSERT INTO legacy_observations (domain, resource, kind, discovered_on, source) VALUES (?, ?, ?, ?, ?)', [ ('example.com', target_url, 'interesting-url', '2026-08-09', 'builtwith'), ('example.com', target_url, 'linkedinlinks', '2026-08-09', 'rocketreach'), ], ) store = ResultStore(database) await store.initialize() loaded = await store.load_run(run_id) assert loaded.results == (('hostname', 'portal.example.com'), ('url', target_url)) assert {(item.source, item.kind, item.value) for item in loaded.observations} == { ('builtwith', 'url', target_url), ('gitlab', 'url', target_url), ('rocketreach', 'url', target_url), } api_scan = next(item for item in loaded.active_evidence.executions if item.action == 'api-scan') assert api_scan.result_count == 1 assert {(item.kind, item.value) for item in api_scan.observations} == {('url', target_url)} screenshot = next(item for item in loaded.active_evidence.executions if item.action == 'screenshot') assert screenshot.artifacts[0].subject_kind == 'url' assert screenshot.artifacts[0].subject_value == target_url with sqlite3.connect(database) as db: assert db.execute('PRAGMA user_version').fetchone()[0] == 8 assert db.execute('SELECT DISTINCT kind FROM legacy_observations').fetchall() == [('url',)] assert db.execute('SELECT result_count FROM executions WHERE name = ?', ('api-scan',)).fetchone()[0] == 1 @pytest.mark.asyncio async def test_schema_v5_merges_ip_address_into_ip_without_losing_provenance_or_artifacts(tmp_path) -> None: database = tmp_path / 'stash.sqlite' run_id = UUID('5b240ef2-e714-45de-b38f-174f20447f8b') address = '192.0.2.1' with sqlite3.connect(database) as db: db.executescript(SCHEMA_V5_RESULT_KINDS) db.execute( 'INSERT INTO runs (run_id, target, started_at, completed_at) VALUES (?, ?, ?, ?)', (str(run_id), 'example.com', '2026-08-09T12:00:00+00:00', '2026-08-09T12:01:00+00:00'), ) db.executemany( 'INSERT INTO results (run_id, position, kind, value) VALUES (?, ?, ?, ?)', [ (str(run_id), 0, 'ip-address', address), (str(run_id), 1, 'ip', address), ], ) db.executemany( 'INSERT INTO executions ' '(run_id, position, producer_kind, name, status, duration_ms, result_count, error_type, stop_reason) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', [ (str(run_id), 0, 'source', 'dns', 'completed', 1.0, 2, None, None), (str(run_id), 1, 'action', 'screenshot', 'completed', 2.0, 2, None, None), ], ) db.executemany( 'INSERT INTO result_origins (run_id, result_position, execution_position) VALUES (?, ?, ?)', [ (str(run_id), 0, 0), (str(run_id), 1, 0), (str(run_id), 0, 1), (str(run_id), 1, 1), ], ) db.execute( 'INSERT INTO artifacts ' '(run_id, position, result_position, execution_position, kind, path, media_type, size_bytes, sha256, created_at) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', ( str(run_id), 0, 0, 1, 'screenshot', 'screenshots/192.0.2.1.png', 'image/png', 3, '0' * 64, '2026-08-09T12:01:00+00:00', ), ) db.executemany( 'INSERT INTO legacy_observations (domain, resource, kind, discovered_on, source) VALUES (?, ?, ?, ?, ?)', [ ('example.com', address, 'ip-address', '2026-08-09', 'dns'), ('example.com', address, 'ip', '2026-08-09', 'dns'), ], ) store = ResultStore(database) await store.initialize() loaded = await store.load_run(run_id) assert loaded.results == (('ip', address),) assert [(item.source, item.kind, item.value) for item in loaded.observations] == [('dns', 'ip', address)] assert loaded.source_executions[0].result_count == 1 screenshot = loaded.active_evidence.executions[0] assert [(item.kind, item.value) for item in screenshot.observations] == [('ip', address)] assert screenshot.artifacts[0].subject_kind == 'ip' assert screenshot.artifacts[0].subject_value == address with sqlite3.connect(database) as db: assert db.execute('PRAGMA user_version').fetchone()[0] == 8 assert db.execute('SELECT kind, value FROM results').fetchall() == [('ip', address)] assert db.execute('SELECT execution_position FROM result_origins ORDER BY execution_position').fetchall() == [ (0,), (1,), ] assert db.execute('SELECT result_count FROM executions ORDER BY position').fetchall() == [(1,), (1,)] assert db.execute('SELECT DISTINCT kind FROM legacy_observations').fetchall() == [('ip',)] @pytest.mark.asyncio async def test_released_results_migrate_to_legacy_observations(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) released_rows = [ ('example.com', 'api.example.com', 'host', '2026-08-08', 'crtsh'), ('example.com', '192.0.2.1', 'ip', '2026-08-08', 'dns'), ('example.com', 'Ada Lovelace', 'people', '2026-08-08', 'hunter'), ('example.com', 'https://linkedin.test/ada', 'linkedinlinks', '2026-08-08', 'linkedin'), ('example.com', 'https://admin.example.com', 'interestingurls', '2026-08-08', 'builtwith'), ('example.com', 'AS64496', 'asns', '2026-08-08', 'shodan'), ('example.com', '/api/v1', 'api_endpoint', '2026-08-08', 'api_scan'), ('example.com', 'admin@example.com', 'email', '2026-08-08', 'hunter'), ] with sqlite3.connect(database) as db: db.executescript(RELEASED_RESULTS_SCHEMA) db.executemany('INSERT INTO results (domain, resource, type, find_date, source) VALUES (?, ?, ?, ?, ?)', released_rows) await store.initialize() await store.initialize() with sqlite3.connect(database) as db: tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} observations = db.execute('SELECT resource, kind FROM legacy_observations ORDER BY id').fetchall() result_columns = [row[1] for row in db.execute('PRAGMA table_info(results)')] schema_version = db.execute('PRAGMA user_version').fetchone()[0] assert 'legacy_results' not in tables assert result_columns == ['run_id', 'position', 'kind', 'value', 'details_json'] assert observations == [ ('api.example.com', 'hostname'), ('192.0.2.1', 'ip'), ('Ada Lovelace', 'person'), ('https://linkedin.test/ada', 'url'), ('https://admin.example.com', 'url'), ('AS64496', 'asn'), ('/api/v1', 'url'), ('admin@example.com', 'email'), ] assert schema_version == 8 @pytest.mark.asyncio async def test_released_null_observation_fields_survive_migration(tmp_path) -> None: database = tmp_path / 'stash.sqlite' with sqlite3.connect(database) as db: db.executescript(RELEASED_RESULTS_SCHEMA) db.execute( 'INSERT INTO results (domain, resource, type, find_date, source) VALUES (?, ?, ?, ?, ?)', (None, None, None, None, None), ) store = ResultStore(database) await store.initialize() with sqlite3.connect(database) as db: migrated = db.execute('SELECT domain, resource, kind, discovered_on, source FROM legacy_observations').fetchall() assert migrated == [(None, None, None, None, None)] @pytest.mark.asyncio async def test_concurrent_initialization_migrates_released_results_once(tmp_path) -> None: database = tmp_path / 'stash.sqlite' first = ResultStore(database) second = ResultStore(database) with sqlite3.connect(database) as db: db.executescript(RELEASED_RESULTS_SCHEMA) db.execute( 'INSERT INTO results (domain, resource, type, find_date, source) VALUES (?, ?, ?, ?, ?)', ('example.com', 'api.example.com', 'host', '2026-08-08', 'crtsh'), ) await asyncio.gather(first.initialize(), second.initialize()) with sqlite3.connect(database) as db: rows = db.execute('SELECT domain, resource, kind, source FROM legacy_observations').fetchall() assert rows == [('example.com', 'api.example.com', 'hostname', 'crtsh')] @pytest.mark.asyncio async def test_completed_result_write_is_atomic_and_rejects_duplicate_run_id(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() result = completed_result() await store.save_run(result) with pytest.raises(DuplicateRunError): await store.save_run(result) failing = completed_result('f9b33a33-e6d6-4a48-b04f-1a4a3012bc1f') with sqlite3.connect(database) as db: db.execute( """ CREATE TRIGGER fail_result BEFORE INSERT ON results WHEN NEW.run_id = 'f9b33a33-e6d6-4a48-b04f-1a4a3012bc1f' BEGIN SELECT RAISE(ABORT, 'forced failure'); END """ ) with pytest.raises(ResultStoreError, match='Could not save enumeration run'): await store.save_run(failing) with sqlite3.connect(database) as db: run_count = db.execute('SELECT COUNT(*) FROM runs').fetchone()[0] result_count = db.execute('SELECT COUNT(*) FROM results').fetchone()[0] assert (run_count, result_count) == (1, 7) artifact_run_id = UUID('7ff120b6-4aec-4d27-b2db-d3ac9fd87340') completed_at = datetime(2026, 8, 9, 12, 1, tzinfo=UTC) failing_artifact = CompletedResult.finish( run_id=artifact_run_id, target='example.com', started_at=datetime(2026, 8, 9, 12, 0, tzinfo=UTC), completed_at=completed_at, groups={'hostname': ['api.example.com']}, active_evidence=ActiveEvidence(executions=(screenshot_execution(completed_at),)), ) with sqlite3.connect(database) as db: db.execute( f""" CREATE TRIGGER fail_artifact BEFORE INSERT ON artifacts WHEN NEW.run_id = '{artifact_run_id}' BEGIN SELECT RAISE(ABORT, 'forced artifact failure'); END """ ) with pytest.raises(ResultStoreError, match='Could not save enumeration run'): await store.save_run(failing_artifact) with sqlite3.connect(database) as db: assert db.execute('SELECT COUNT(*) FROM runs').fetchone()[0] == 1 assert db.execute('SELECT COUNT(*) FROM artifacts').fetchone()[0] == 0 @pytest.mark.asyncio async def test_legacy_observations_keep_the_released_normalized_schema(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) await store.initialize() await store.record_observations('example.com', ['api.example.com', 'www.example.com'], 'hostname', 'crtsh') await store.record_observations('example.com', ['admin@example.com'], 'email', 'hunter') await store.record_observations('example.com', ['192.0.2.1'], 'ip', 'dns') await store.record_observations('example.com', ['{"firstname":"Ada","lastname":"Lovelace"}'], 'person', 'hunter') await store.record_observations('example.com', ['vhost.example.com'], 'hostname', 'virtual-host') await store.record_observations('example.com', ['443'], 'shodan', 'shodan') with sqlite3.connect(database) as db: columns = [row[1] for row in db.execute('PRAGMA table_info(legacy_observations)')] rows = db.execute('SELECT domain, resource, kind, source FROM legacy_observations ORDER BY id').fetchall() assert columns == ['id', 'domain', 'resource', 'kind', 'discovered_on', 'source'] assert rows == [ ('example.com', 'api.example.com', 'hostname', 'crtsh'), ('example.com', 'www.example.com', 'hostname', 'crtsh'), ('example.com', 'admin@example.com', 'email', 'hunter'), ('example.com', '192.0.2.1', 'ip', 'dns'), ('example.com', '{"firstname":"Ada","lastname":"Lovelace"}', 'person', 'hunter'), ('example.com', 'vhost.example.com', 'hostname', 'virtual-host'), ('example.com', '443', 'shodan', 'shodan'), ] @pytest.mark.asyncio async def test_schema_v1_observations_upgrade_without_losing_rows(tmp_path) -> None: database = tmp_path / 'stash.sqlite' store = ResultStore(database) with sqlite3.connect(database) as db: db.executescript(SCHEMA_V1_DISCOVERY_OBSERVATIONS) db.execute( 'INSERT INTO discovery_observations (domain, resource, kind, discovered_on, source) VALUES (?, ?, ?, ?, ?)', ('example.com', 'api.example.com', 'hostname', '2026-08-08', 'crtsh'), ) await store.initialize() with sqlite3.connect(database) as db: tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} rows = db.execute('SELECT domain, resource, kind, source FROM legacy_observations').fetchall() schema_version = db.execute('PRAGMA user_version').fetchone()[0] assert 'discovery_observations' not in tables assert rows == [('example.com', 'api.example.com', 'hostname', 'crtsh')] assert schema_version == 8 @pytest.mark.asyncio async def test_completed_results_are_ordered_by_instant_across_offsets(tmp_path) -> None: store = ResultStore(tmp_path / 'stash.sqlite') await store.initialize() earlier = CompletedResult.finish( run_id=UUID('32c0630c-4af8-421a-9650-10f1472db591'), target='earlier.example', started_at=datetime(2025, 12, 31, 23, 59, tzinfo=timezone(timedelta(hours=2))), completed_at=datetime(2026, 1, 1, 0, 0, tzinfo=timezone(timedelta(hours=2))), groups={'hostname': ['earlier.example']}, ) later = CompletedResult.finish( run_id=UUID('dc0679ee-e52b-4908-8a3c-9ed26ad8f0cc'), target='later.example', started_at=datetime(2025, 12, 31, 22, 59, tzinfo=UTC), completed_at=datetime(2025, 12, 31, 23, 0, tzinfo=UTC), groups={'hostname': ['later.example']}, ) await store.save_run(earlier) await store.save_run(later) rows = await store.list_runs() assert [row['target'] for row in rows] == ['later.example', 'earlier.example']