import hashlib import io import os from pathlib import Path import subprocess import sys import tarfile import tempfile import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) from keycheck_candidates import extract_candidates from parity_helpers import ( bundle_evidence_difference_paths, configured_trufflehog, native_streamed_command, normalized_bundle_evidence, ) from result_bundle import BundleReservation, ResultBundleReader from runtime_security import ensure_private_directory import scan_execution import scanner import scanner_db TRUFFLEHOG = configured_trufflehog() def synthetic_material(label, length, alphabet): seed = hashlib.sha512(label.encode('ascii')).hexdigest() output = '' while len(output) < length: output += ''.join( alphabet[int(seed[index:index + 2], 16) % len(alphabet)] for index in range(0, len(seed), 2) ) seed = hashlib.sha512(seed.encode('ascii')).hexdigest() return output[:length] def synthetic_llm_tokens(): alphabet = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789' return { 'openai': ( 'sk-proj-' + synthetic_material('local-openai-canary-prefix', 24, alphabet) + 'T3BlbkFJ' + synthetic_material('local-openai-canary-suffix', 24, alphabet) ), 'openrouter': ( 'sk-or-v1-' + synthetic_material('local-openrouter-canary', 64, '0123456789abcdef') ), 'anthropic': ( 'sk-ant-api03-' + synthetic_material('local-anthropic-canary', 93, alphabet + '-_') + 'AA' ), } def run_synthetic_scan(root): fixture_path = os.path.join(root, 'synthetic.env') with open(fixture_path, 'w', encoding='utf-8', newline='\n') as handle: for name, value in synthetic_llm_tokens().items(): handle.write(f'{name.upper()}_API_KEY={value}\n') completed = subprocess.run( [ str(TRUFFLEHOG), 'filesystem', root, '--json', '--no-update', '--no-verification', ], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, timeout=60, ) if completed.returncode != 0: raise AssertionError(f'TruffleHog exited with {completed.returncode}') result = {'findings': [], 'errors': []} scanner.append_trufflehog_findings( result, [line.decode('utf-8', errors='replace') for line in completed.stdout.splitlines()], ) scanner.attach_nearby_context(result) scanner.apply_finding_filters(result, root) return result @unittest.skipUnless(TRUFFLEHOG, 'configured TruffleHog binary is unavailable') class SyntheticLLMPipelineTests(unittest.TestCase): def test_real_scanner_detects_and_routes_synthetic_llm_keys(self): with tempfile.TemporaryDirectory() as root: result = run_synthetic_scan(root) scanner.strip_nearby_context_for_persistence(result) detectors = { str(finding.get('DetectorName') or finding.get('DetectorType') or '') for finding in result['findings'] } services = { candidate.service for finding in result['findings'] for candidate in extract_candidates(finding) } self.assertEqual(detectors, {'OpenAI', 'OpenRouter', 'Anthropic'}) self.assertEqual(services, {'openai', 'openrouter', 'anthropic'}) def test_synthetic_llm_candidates_survive_durable_bundle_staging(self): with tempfile.TemporaryDirectory() as root: bundle_root = os.path.join(root, 'bundles') ensure_private_directory(bundle_root, reject_reparse=True) result = run_synthetic_scan(root) result.update({ 'scan_event_id': 'a' * 32, 'target': 'https://example.invalid/synthetic-llm-fixture', 'scan_type': 'github', 'timestamp': '2026-09-07T00:00:00+00:00', 'scan_started_at': '2026-09-07T00:00:00+00:00', 'duration_sec': 1.0, }) scanner.assign_finding_uids(result) reservation = BundleReservation( reservation_id=7, reservation_token='synthetic-reservation-token', bundle_id='b' * 32, scan_event_id=result['scan_event_id'], queue_id=11, claim_lease_token='synthetic-claim-token', declared_bytes=4 * 1024 * 1024, ready_path='ready/bb/' + ('b' * 32) + '.trb', source='github', platform='github', target=result['target'], normalized_target=result['target'], ) staged = scanner.stage_result_bundle( result, reservation, bundle_root, {'no_verification': True}, {'queue_status': 'done', 'queue_error': None, 'available_after': None}, candidate_max_items=10, candidate_max_bytes=1024 * 1024, ) self.assertEqual(result['findings'], []) reader = ResultBundleReader( os.path.join(bundle_root, *staged.relative_path.split('/')) ) metadata = reader.validate() candidate_services = { str(candidate.get('service') or '') for candidate in reader.iter_candidates() } self.assertEqual(metadata.finding_count, 3) self.assertEqual(metadata.candidate_count, 3) self.assertEqual(candidate_services, {'openai', 'openrouter', 'anthropic'}) def test_docker_layer_fallback_detects_the_same_synthetic_llm_keys(self): payload = '\n'.join( f'{name.upper()}_API_KEY={value}' for name, value in synthetic_llm_tokens().items() ).encode('utf-8') archive_buffer = io.BytesIO() with tarfile.open(fileobj=archive_buffer, mode='w:gz') as archive: member = tarfile.TarInfo('app/synthetic.env') member.size = len(payload) member.mode = 0o600 archive.addfile(member, io.BytesIO(payload)) layer_blob = archive_buffer.getvalue() limits = { 'config_max_bytes': 1024, 'layer_max_bytes': 1024 * 1024, 'image_max_bytes': 2 * 1024 * 1024, 'max_layers': 1, 'archive_max_size_bytes': 1024 * 1024, 'archive_max_depth': 4, 'archive_timeout_sec': 10, 'blob_timeout_sec': 30, 'filesystem_concurrency': 1, 'blob_max_attempts': 3, } config_blob = b'{}' plan = { 'version': 2, 'image': 'owner/synthetic@sha256:' + ('a' * 64), 'repository': 'owner/synthetic', 'manifest_digest': 'sha256:' + ('a' * 64), 'platform_os': 'linux', 'platform_arch': 'amd64', 'manifest_media_type': 'application/vnd.oci.image.manifest.v1+json', 'limits': limits, 'selector_version': scanner_db.DOCKER_ADAPTIVE_SELECTOR_VERSION, 'selection_policy_sha256': scanner_db.docker_layer_selection_policy_sha256(limits), 'scan_policy_sha256': 'c' * 64, 'execution_policy_sha256': scanner_db.docker_layer_execution_policy_sha256( 'c' * 64, limits, ), 'checkpoint': {'max_blobs': 1, 'max_bytes': 1024 * 1024}, 'descriptors': [ { 'digest': 'sha256:' + hashlib.sha256(config_blob).hexdigest(), 'size': len(config_blob), 'media_type': 'application/vnd.oci.image.config.v1+json', 'kind': 'config', 'position': 0, 'payload_class': 'config', 'selected': True, 'selection_reason': 'already_covered', 'coverage_state': 'covered', 'lease_token': None, 'attempt': 0, 'max_attempts': 3, }, { 'digest': 'sha256:' + hashlib.sha256(layer_blob).hexdigest(), 'size': len(layer_blob), 'media_type': 'application/vnd.oci.image.layer.v1.tar+gzip', 'kind': 'layer', 'position': 1, 'payload_class': 'copy_add', 'selected': True, 'selection_reason': 'selected_copy_add', 'coverage_state': 'leased', 'lease_token': 'l' * 43, 'attempt': 1, 'max_attempts': 3, }, ], } plan = scanner_db.validate_docker_layer_plan(plan) class StreamResponse: status_code = 200 headers = { 'Content-Length': str(len(layer_blob)), 'Content-Encoding': 'identity', } url = 'https://registry-1.docker.io/v2/owner/synthetic/blobs/private' @staticmethod def iter_content(chunk_size=1): del chunk_size yield layer_blob @staticmethod def close(): return None plan_sha256 = hashlib.sha256( scanner_db.canonical_docker_layer_plan_bytes(plan) ).hexdigest() work = { 'plan': plan, 'plan_sha256': plan_sha256, 'bearer_auth': scanner.DockerRegistryAuth(token=''), 'min_free_bytes': 0, } claim = BundleReservation( reservation_id=9, reservation_token='docker-reservation-token', bundle_id='d' * 32, scan_event_id='e' * 32, queue_id=13, claim_lease_token='docker-claim-token', declared_bytes=4 * 1024 * 1024, ready_path='ready/dd/' + ('d' * 32) + '.trb', source='docker', platform='docker', query='fixture', target=plan['image'], normalized_target=scanner.normalize_target(plan['image'], 'docker'), ) kwargs = { 'timeout_sec': 60, 'docker_layer_work': work, 'no_verification': True, } with tempfile.TemporaryDirectory() as root: work_root = os.path.join(root, 'work') local_root = os.path.join(root, 'local') remote_root = os.path.join(root, 'remote') for path in (work_root, local_root, remote_root): ensure_private_directory(path, reject_reparse=True) with mock.patch.object( scanner, 'get_work_dir', return_value=work_root, ), mock.patch.object( scanner, 'get_trufflehog_cmd', return_value=str(TRUFFLEHOG), ), mock.patch.object( scanner, '_docker_registry_blob_response', return_value=(StreamResponse(), scanner.DockerRegistryAuth(token='')), ), mock.patch.object( scanner, 'run_command_streamed', side_effect=native_streamed_command, ), mock.patch.object( scanner, 'require_scanner_runtime_initialized', return_value=None, ): local_result = scanner.scan_target_result( claim.target, 'docker', claim.scan_event_id, kwargs, ) local = scan_execution.stage_scan_result_in_scope( local_result, claim, local_root, {}, scan_execution.QueueDispositionPolicy(), attempts=1, ) remote = scan_execution.execute_planned_result_in_scope( claim, remote_root, kwargs, {}, scan_execution.QueueDispositionPolicy(), attempts=1, ) local_path = os.path.join(local_root, local.relative_path) remote_path = os.path.join(remote_root, remote.relative_path) local_metadata = ResultBundleReader(local_path).metadata() execution = scanner_db.validate_docker_layer_execution( local_metadata['docker_layer_execution'], plan, plan_sha256, ) for bundle_path in (local_path, remote_path): reader = ResultBundleReader(bundle_path) finding_uids = { finding['finding_uid'] for finding in reader.iter_findings() } self.assertEqual(len(finding_uids), 3) for candidate in reader.iter_candidates(): attribution = candidate['attribution'] self.assertIn(attribution['finding_uid'], finding_uids) self.assertEqual( candidate['metadata']['finding_uid'], attribution['finding_uid'], ) local_evidence = normalized_bundle_evidence(local_path) remote_evidence = normalized_bundle_evidence(remote_path) self.assertEqual( bundle_evidence_difference_paths(local_evidence, remote_evidence), [], ) self.assertEqual(local.queue_status, remote.queue_status) self.assertEqual(local.queue_status, 'done') detectors = { str(finding.get('DetectorName') or finding.get('DetectorType') or '') for finding in local_evidence['findings'] } services = { str(candidate.get('service') or '') for candidate in local_evidence['candidates'] } diagnostics = { 'errors': local_evidence['errors'], 'metadata': local_evidence['metadata'], } self.assertEqual( detectors, {'OpenAI', 'OpenRouter', 'Anthropic'}, diagnostics, ) self.assertEqual(services, {'openai', 'openrouter', 'anthropic'}) self.assertEqual(execution['blobs'][0]['status'], 'covered') self.assertEqual(execution['blobs'][0]['finding_count'], 3) self.assertEqual(execution['blobs'][0]['lease_token'], 'l' * 43) if __name__ == '__main__': unittest.main()