import contextlib import builtins from concurrent.futures import Future from datetime import datetime, timedelta, timezone import hashlib import json import os from pathlib import Path import sqlite3 import sys import tempfile import threading import time from types import SimpleNamespace import unittest import uuid from unittest import mock ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import console_runner import db_backend from keycheckers import keycheck_common import optimize_dashboard_db import postgres_runtime import runtime_security import scanner from scanner_db import ScannerDB class ControlledExecutor: def __init__(self): self.submissions = [] def submit(self, function): future = Future() self.submissions.append((function.__name__, future)) return future class LifecycleBackend: def __init__(self): self.closed = False def probe(self): raise AssertionError('controlled future should supply probe result') def start(self): raise AssertionError('controlled future should supply start result') def stop(self): raise AssertionError('controlled future should supply stop result') def close(self): self.closed = True class EndpointAuthorityTests(unittest.TestCase): def test_endpoint_identity_ignores_role_and_runtime_temp_roots(self): first = { 'global': { 'runtime_dir': r'D:\copy-one\runtime', 'database_url': 'postgresql://role_one:secret@localhost:5544/shared', }, } second = { 'global': { 'runtime_dir': r'E:\copy-two\other-runtime', 'database_url': 'postgresql://role_two:different@127.0.0.1:5544/shared', }, } cleared = { 'TRUF_MANAGED_POSTGRES_DSN': '', 'SCANNER_DB_URL': '', 'DATABASE_URL': '', 'TRUF_POSTGRES_DB': '', 'TRUF_POSTGRES_USER': '', 'TRUF_POSTGRES_PORT': '', } with mock.patch.dict(os.environ, cleared): first_dsn = first['global']['database_url'] second_dsn = second['global']['database_url'] identity_one = runtime_security.canonical_cluster_endpoint_identity(first, database_url=first_dsn) identity_two = runtime_security.canonical_cluster_endpoint_identity(second, database_url=second_dsn) path_one = runtime_security.cluster_endpoint_authority_lock_path(first, endpoint_dsn=first_dsn) with mock.patch.dict(os.environ, {'TEMP': r'Z:\attacker-temp', 'TMP': r'Y:\other-temp'}): path_two = runtime_security.cluster_endpoint_authority_lock_path(second, endpoint_dsn=second_dsn) self.assertEqual(identity_one, identity_two) self.assertEqual(path_one, path_two) self.assertNotIn('user', json.loads(identity_one)) self.assertEqual(json.loads(identity_one)['schema'], 'public') self.assertTrue(runtime_security.cluster_endpoint_mutex_name(endpoint_identity=identity_one).startswith('Global\\')) def test_managed_dsn_precedence_is_shared_by_connections_and_locks(self): managed = 'postgresql://managed:a@127.0.0.1:6001/effective' conflicting = 'postgresql://other:b@127.0.0.1:6002/wrong' config = {'global': {'runtime_dir': r'D:\runtime', 'database_url': conflicting}} with mock.patch.dict(os.environ, { 'TRUF_MANAGED_POSTGRES_DSN': managed, 'SCANNER_DB_URL': conflicting, 'DATABASE_URL': conflicting, }): self.assertEqual(db_backend.database_url_from_env(), managed) identity = json.loads(runtime_security.canonical_cluster_endpoint_identity(config, database_url=managed)) explicit = json.loads(runtime_security.canonical_cluster_endpoint_identity( config, database_url='postgresql://maintenance:c@127.0.0.1:6003/explicit', )) self.assertEqual((identity['host'], identity['port'], identity['database']), ('127.0.0.1', 6001, 'effective')) self.assertEqual((explicit['port'], explicit['database']), (6003, 'explicit')) class LateStartLifecycleTests(unittest.TestCase): def _controller_with_start(self, timeout=0.5): backend = LifecycleBackend() executor = ControlledExecutor() controller = postgres_runtime.PostgresController( backend, executor=executor, shutdown_timeout_sec=timeout, health_interval_sec=0.1, ) controller.tick(0) executor.submissions[0][1].set_result( postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'offline') ) controller.tick(0) self.assertEqual([name for name, _ in executor.submissions], ['probe', 'start']) return controller, backend, executor def test_close_keeps_unresolved_start_and_marks_authority_unsafe(self): controller, backend, _ = self._controller_with_start(timeout=0.01) self.assertFalse(controller.close(timeout_sec=0.01)) self.assertEqual(controller.state, postgres_runtime.PostgresState.STOP_FAILED) self.assertIsNotNone(controller._future) self.assertFalse(controller.authority_release_safe) self.assertFalse(backend.closed) def test_close_drains_late_acceptance_and_compensating_stop(self): controller, backend, executor = self._controller_with_start(timeout=1) def complete_operations(): time.sleep(0.02) executor.submissions[1][1].set_result(postgres_runtime.StartResult(True, 'late accepted', uncertain=True)) deadline = time.monotonic() + 1 while len(executor.submissions) < 3 and time.monotonic() < deadline: time.sleep(0.005) executor.submissions[2][1].set_result(postgres_runtime.StopResult(True, True, 'compensated')) worker = threading.Thread(target=complete_operations) worker.start() self.assertTrue(controller.close(timeout_sec=1)) worker.join(2) self.assertEqual(controller.state, postgres_runtime.PostgresState.STOPPED) self.assertTrue(backend.closed) def test_owned_recovery_disappearance_enters_backoff_not_foreign(self): class ImmediateExecutor: def submit(self, function): future = Future() try: future.set_result(function()) except BaseException as exc: future.set_exception(exc) return future class Backend: def __init__(self): self.probes = iter([ postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'offline'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.RECOVERING, 'owned recovery'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'briefly disappeared'), ]) def probe(self): return next(self.probes) def start(self): return postgres_runtime.StartResult(True, 'accepted') def stop(self): return postgres_runtime.StopResult(True, True, 'stopped') def close(self): pass controller = postgres_runtime.PostgresController( Backend(), executor=ImmediateExecutor(), health_interval_sec=0.1, stable_ready_interval_sec=1, ) for now in (0, 0, 0, 0, 1, 1): controller.tick(now) self.assertEqual(controller.state, postgres_runtime.PostgresState.BACKOFF) self.assertNotEqual(controller.state, postgres_runtime.PostgresState.FOREIGN_OR_CONFIG_ERROR) def test_uncertain_owned_start_is_compensated_before_retry(self): class ImmediateExecutor: def submit(self, function): future = Future() future.set_result(function()) return future class Backend: def __init__(self): self.probes = iter([ postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'offline'), postgres_runtime.ProbeResult( postgres_runtime.ProbeKind.OWNED_START_UNCERTAIN, 'late side effect uncertain', ), ]) self.starts = 0 self.stops = 0 def probe(self): return next(self.probes) def start(self): self.starts += 1 return postgres_runtime.StartResult(True, 'accepted', uncertain=True) def stop(self): self.stops += 1 return postgres_runtime.StopResult(True, True, 'offline confirmed') def close(self): pass backend = Backend() controller = postgres_runtime.PostgresController( backend, executor=ImmediateExecutor(), health_interval_sec=0.1, ) for now in (0, 0, 0, 0, 0): controller.tick(now) self.assertEqual(controller.state, postgres_runtime.PostgresState.BACKOFF) self.assertEqual((backend.starts, backend.stops), (1, 1)) class OutboxRetentionTests(unittest.TestCase): def test_delivered_payloads_are_compacted_and_rows_bounded(self): with tempfile.TemporaryDirectory() as temp_dir, mock.patch.dict(os.environ, { 'SCANNER_DB_URL': '', 'DATABASE_URL': '', 'TRUF_MANAGED_POSTGRES_DSN': '', 'SCAN_OUTBOX_DELIVERED_MAX_ROWS': '3', 'SCAN_OUTBOX_DELIVERED_MAX_AGE_DAYS': '30', }): db = ScannerDB(db_path=os.path.join(temp_dir, 'scanner.db')) try: run_id = db.start_run('test', ['test']) cycle_id = db.start_source_cycle(run_id, 'source', 'github', 'search', 'q', 1, 1, None, {}, {}) for index in range(8): target_id = db.record_target_result( run_id, cycle_id, 'source', 'q', f'target-{index}', {'scan_event_id': str(uuid.uuid4()), 'target': f'target-{index}', 'findings': [], 'errors': []}, ) owner = f'owner-{index}' now = datetime.now(timezone.utc).isoformat(timespec='seconds') outbox_id = db.conn.insert_returning_id( '''INSERT INTO scan_publication_outbox( target_scan_id, payload_json, status, attempts, lease_owner, created_at, updated_at ) VALUES (?, ?, 'delivering', 1, ?, ?, ?)''', (target_id, json.dumps({'payload': 'x' * (512 * 1024)}), owner, now, now), ) db.conn.commit() self.assertTrue(db.finish_scan_publication(outbox_id, owner, True)) row = db.conn.execute( '''SELECT COUNT(*) AS count, COALESCE(SUM(LENGTH(payload_json)), 0) AS payload_bytes FROM scan_publication_outbox WHERE status = 'delivered' ''' ).fetchone() self.assertLessEqual(row['count'], 3) self.assertEqual(row['payload_bytes'], 0) finally: db.close() class DisabledProjectionTests(unittest.TestCase): def test_postgres_queue_does_not_create_legacy_files_when_disabled(self): class DB: postgres_required = True conn = SimpleNamespace(is_postgres=True) def require_runtime_safety_schema(self): return True def known_target_normalizations_for(self, source, platform, targets): return set() def enqueue_targets(self, source, platform, query, targets, **kwargs): return len(targets) + len(kwargs.get('unresolved_targets') or []) def claim_targets(self, source, platform, limit, owner, lease_seconds, **kwargs): return [] def target_queue_counts(self, source): return {'pending': 1} class Spool: def reserve_claims(self, owner, count, lease_seconds): return 'reservation' def bind_claims(self, reservation, claims): return None with tempfile.TemporaryDirectory() as temp_dir: args = SimpleNamespace( save_dir=os.path.join(temp_dir, 'results'), queue_dir=os.path.join(temp_dir, 'queues'), platform='github', query='q', workers=1, target_claim_batch_size=1, max_targets=0, timeout=60, target_retry_max_attempts=3, sync_file_queues=False, ) todo = os.path.join(args.queue_dir, 'todo_github.txt') checked = os.path.join(args.queue_dir, 'checked_github.txt') with mock.patch.object(console_runner, 'projection_file_lock') as projection_lock, \ mock.patch.object(console_runner, '_append_lines_unlocked') as append, \ mock.patch.object(console_runner, '_write_lines_unlocked') as rewrite: console_runner.prepare_targets( args, ['https://github.com/example/repo'], DB(), 1, 1, 'source', spool=Spool(), ) projection_lock.assert_not_called() append.assert_not_called() rewrite.assert_not_called() self.assertFalse(os.path.exists(todo)) self.assertFalse(os.path.exists(checked)) class JsonlPublicationTests(unittest.TestCase): @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) def private_dir(self, root): scanner.ensure_private_directory(root, reject_reparse=True) return root def test_retry_uses_ledger_without_history_scan(self): with tempfile.TemporaryDirectory() as temp_dir: self.private_dir(temp_dir) path = os.path.join(temp_dir, 'scan_results.jsonl') with mock.patch.object(scanner.scan_config, 'jsonl_rotation_enabled', False): for index in range(40): self.assertTrue(scanner.append_rotating_jsonl_once( path, {'scan_event_id': f'event-{index:016d}', 'value': index}, 'scan_event_id', f'event-{index:016d}', 1, )) with mock.patch.object(scanner, 'projection_segment_paths', side_effect=AssertionError('history scan')): self.assertTrue(scanner.append_rotating_jsonl_once( path, {'scan_event_id': 'event-final-00000001', 'value': 99}, 'scan_event_id', 'event-final-00000001', 1, )) self.assertTrue(scanner.append_rotating_jsonl_once( path, {'scan_event_id': 'event-final-00000001', 'value': 99}, 'scan_event_id', 'event-final-00000001', 1, )) self.assertEqual(len(Path(path).read_text(encoding='utf-8').splitlines()), 41) def test_torn_tail_is_bounded_quarantined_then_repaired(self): with tempfile.TemporaryDirectory() as temp_dir: self.private_dir(temp_dir) path = os.path.join(temp_dir, 'found_secrets.jsonl') first = {'finding_uid': 'finding-00000000000001', 'value': 'first'} second = {'finding_uid': 'finding-00000000000002', 'value': 'second'} with mock.patch.object(scanner.scan_config, 'jsonl_rotation_enabled', False), \ mock.patch.object(scanner.scan_config, 'jsonl_torn_quarantine_max_bytes', 16): self.assertTrue(scanner.append_rotating_jsonl_once(path, first, 'finding_uid', first['finding_uid'], 1)) with open(path, 'ab') as handle: handle.write(b'{"torn":"' + (b'x' * 100)) self.assertTrue(scanner.append_rotating_jsonl_once(path, second, 'finding_uid', second['finding_uid'], 1)) rows = [json.loads(line) for line in Path(path).read_text(encoding='utf-8').splitlines()] self.assertEqual([row['finding_uid'] for row in rows], [first['finding_uid'], second['finding_uid']]) self.assertLessEqual(os.path.getsize(path + '.torn-tail.bin'), 16) def test_manifest_failure_keeps_active_history_and_physical_orphan_visible(self): with tempfile.TemporaryDirectory() as temp_dir: self.private_dir(temp_dir) path = os.path.join(temp_dir, 'scan_results.jsonl') original = b'{"scan_event_id":"event-00000000000001"}\n' Path(path).write_bytes(original) scanner.harden_private_file(path) with mock.patch.object(scanner, 'write_jsonl_manifest', side_effect=OSError('manifest failed')): with self.assertRaisesRegex(OSError, 'manifest failed'): scanner.rotate_jsonl_if_needed(path, 1) self.assertEqual(Path(path).read_bytes(), original) self.assertEqual(len(scanner.projection_segment_paths(path)), 2) scanner.reconcile_jsonl_segments(path) self.assertEqual(Path(path).read_bytes(), b'') self.assertEqual(len(scanner.projection_segment_paths(path)), 2) def test_crash_after_append_before_ledger_ack_is_idempotent(self): with tempfile.TemporaryDirectory() as temp_dir: self.private_dir(temp_dir) path = os.path.join(temp_dir, 'scan_results.jsonl') Path(path).write_bytes(b'') scanner.harden_private_file(path) payload = {'scan_event_id': 'event-crash-00000001', 'value': 'fixture'} real_harden = scanner.harden_private_file failed = {'value': False} def fail_after_append(candidate): if os.path.normcase(os.path.abspath(candidate)) == os.path.normcase(os.path.abspath(path)) and not failed['value']: failed['value'] = True raise OSError('crash after append') return real_harden(candidate) with mock.patch.object(scanner.scan_config, 'jsonl_rotation_enabled', False), \ mock.patch.object(scanner, 'harden_private_file', side_effect=fail_after_append): self.assertFalse(scanner.append_rotating_jsonl_once( path, payload, 'scan_event_id', payload['scan_event_id'], 1, )) with mock.patch.object(scanner.scan_config, 'jsonl_rotation_enabled', False): self.assertTrue(scanner.append_rotating_jsonl_once( path, payload, 'scan_event_id', payload['scan_event_id'], 1, )) self.assertEqual(len(Path(path).read_text(encoding='utf-8').splitlines()), 1) def test_crash_after_prepared_state_before_append_retries_once(self): with tempfile.TemporaryDirectory() as temp_dir: self.private_dir(temp_dir) path = os.path.join(temp_dir, 'scan_results.jsonl') Path(path).write_bytes(b'') scanner.harden_private_file(path) payload = {'scan_event_id': 'event-prepared-000001', 'value': 'fixture'} real_open = builtins.open failed = {'value': False} def fail_append(candidate, mode='r', *args, **kwargs): if ( os.path.normcase(os.path.abspath(os.fspath(candidate))) == os.path.normcase(os.path.abspath(path)) and mode == 'ab' and not failed['value'] ): failed['value'] = True raise OSError('crash before append') return real_open(candidate, mode, *args, **kwargs) with mock.patch.object(scanner.scan_config, 'jsonl_rotation_enabled', False), \ mock.patch('builtins.open', side_effect=fail_append): self.assertFalse(scanner.append_rotating_jsonl_once( path, payload, 'scan_event_id', payload['scan_event_id'], 1, )) ledger = sqlite3.connect(scanner.jsonl_ledger_path(path)) try: self.assertEqual( ledger.execute('SELECT state FROM publication_identity').fetchone()[0], 'prepared', ) finally: ledger.close() with mock.patch.object(scanner.scan_config, 'jsonl_rotation_enabled', False): self.assertTrue(scanner.append_rotating_jsonl_once( path, payload, 'scan_event_id', payload['scan_event_id'], 1, )) self.assertEqual(len(Path(path).read_text(encoding='utf-8').splitlines()), 1) def test_keycheck_reader_discovers_orphans_and_fails_on_torn_segment(self): with tempfile.TemporaryDirectory() as temp_dir: self.private_dir(temp_dir) output_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(output_dir, reject_reparse=True) current = os.path.join(temp_dir, 'found_secrets.jsonl') segment = os.path.join(temp_dir, 'found_secrets.000001.jsonl') Path(segment).write_text('{"finding_uid":"one"}\n', encoding='utf-8') Path(current).write_text('{"finding_uid":"two"}\n', encoding='utf-8') scanner.harden_private_file(segment) scanner.harden_private_file(current) with mock.patch.dict(os.environ, { 'KEYCHECK_OUTPUT_DIR': output_dir, 'KEYCHECK_STATE_DIR': output_dir, 'KEYCHECK_SERVICE': 'fixture', 'KEYCHECK_INPUT_TAIL_BYTES': '0', }): rows = list(keycheck_common.iter_jsonl_input(current)) self.assertEqual([row['data']['finding_uid'] for row in rows], ['one', 'two']) torn = os.path.join(temp_dir, 'found_secrets.000002.jsonl') Path(torn).write_bytes(b'{"finding_uid":"torn"}') scanner.harden_private_file(torn) with self.assertRaisesRegex(RuntimeError, 'torn'): list(keycheck_common.iter_jsonl_input(current)) class BoundedCacheTests(unittest.TestCase): def test_cached_occurrence_load_compacts_100k_rows_to_tiny_limits(self): with tempfile.TemporaryDirectory() as temp_dir: runtime_security.ensure_private_directory(temp_dir, reject_reparse=True) path = os.path.join(temp_dir, 'cached_occurrences.tsv') old = '2000-01-01T00:00:00+00:00' recent = datetime.now(timezone.utc).isoformat(timespec='seconds') with open(path, 'w', encoding='utf-8') as handle: for index in range(99990): handle.write(f'{index:064x}\t{old}\n') for index in range(99990, 100000): handle.write(f'{index:064x}\t{recent}\n') keycheck_common.harden_private_file(path) keycheck_common.CACHED_OCCURRENCE_CACHE.clear() keycheck_common.CACHED_OCCURRENCE_ROWS.clear() keycheck_common.CACHED_OCCURRENCE_SIGNATURES.clear() with mock.patch.dict(os.environ, { 'KEYCHECK_OUTPUT_DIR': temp_dir, 'KEYCHECK_CACHED_OCCURRENCE_MAX_ITEMS': '5', 'KEYCHECK_CACHED_OCCURRENCE_MAX_BYTES': '512', 'KEYCHECK_CACHED_OCCURRENCE_TTL_SEC': '3600', }): seen = keycheck_common.cached_occurrence_set('fixture') self.assertLessEqual(len(seen), 5) self.assertLessEqual(os.path.getsize(path), 512) def test_postman_pretty_expansion_and_context_amplification_fail_closed(self): with tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) cache_dir = os.path.join(temp_dir, 'cache') scanner.ensure_private_directory(cache_dir, reject_reparse=True) compact = ('[' + ','.join('"x"' for _ in range(240000)) + ']').encode('ascii') with mock.patch.object(scanner.scan_config, 'runtime_dir', temp_dir): path, _, _ = scanner.write_postman_cache( compact, cache_dir=cache_dir, max_artifact_size_mb=1, cache_max_items=10, cache_max_bytes=10 * 1024 * 1024, cache_min_free_bytes=0, ) self.assertEqual(Path(path).read_bytes(), compact) context_path = os.path.join(temp_dir, 'context.json') Path(context_path).write_text(json.dumps({'items': list(range(100))}), encoding='utf-8') scanner.harden_private_file(context_path) with mock.patch.object(scanner.scan_config, 'postman_context_max_nodes', 10): with self.assertRaisesRegex(scanner.PostmanCacheValidationError, 'traversal'): scanner.load_postman_context(context_path) def test_dockerhub_cache_has_stable_keys_ttl_and_row_cap(self): with tempfile.TemporaryDirectory() as temp_dir: path = os.path.join(temp_dir, 'dockerhub.sqlite') options = ( mock.patch.object(scanner.scan_config, 'dockerhub_tag_cache_path', path), mock.patch.object(scanner.scan_config, 'dockerhub_tag_cache_max_rows', 3), mock.patch.object(scanner.scan_config, 'dockerhub_tag_cache_max_age_sec', 60), mock.patch.object(scanner.scan_config, 'dockerhub_tag_cache_max_bytes', 1024 * 1024), mock.patch.object(scanner.scan_config, 'dockerhub_tag_cache_min_free_bytes', 0), ) with contextlib.ExitStack() as stack: for option in options: stack.enter_context(option) scanner._dockerhub_tag_cache_initialized.discard(path) first_since = datetime.now(timezone.utc) - timedelta(hours=2) later_since = first_since + timedelta(hours=1) self.assertEqual( scanner.dockerhub_cache_key('owner/repo', first_since, 1), scanner.dockerhub_cache_key('owner/repo', later_since, 1), ) for index in range(10): scanner.put_dockerhub_tag_cache( f'owner/repo-{index}', first_since, 1, 'ok', [f'owner/repo-{index}:tag'], 3600, tag_records=[{'name': 'tag', 'updated_at': first_since.timestamp() + 5400}], ) conn = sqlite3.connect(path) try: self.assertLessEqual(conn.execute('SELECT COUNT(*) FROM dockerhub_tag_cache').fetchone()[0], 3) conn.execute( 'UPDATE dockerhub_tag_cache SET checked_at = 0, expires_at = ?', (time.time() + 3600,), ) conn.commit() finally: conn.close() self.assertIsNone(scanner.get_dockerhub_tag_cache('owner/repo-9', later_since, 1)) conn = sqlite3.connect(path) try: self.assertEqual(conn.execute('SELECT COUNT(*) FROM dockerhub_tag_cache').fetchone()[0], 0) finally: conn.close() self.assertLessEqual(scanner.dockerhub_tag_cache_disk_bytes(path), 1024 * 1024) class RetiredOptimizationCliTests(unittest.TestCase): def test_direct_dashboard_ddl_cli_is_retired(self): with self.assertRaisesRegex(SystemExit, 'retired'): optimize_dashboard_db.main() source = (APP_DIR / 'optimize_dashboard_db.py').read_text(encoding='utf-8') self.assertNotIn('sqlite3', source) self.assertNotIn('CREATE INDEX', source) if __name__ == '__main__': unittest.main()