sashatrask
352 lines
14 KiB
Python
352 lines
14 KiB
Python
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()
|