sashatrask
188 lines
7.8 KiB
Python
188 lines
7.8 KiB
Python
import gzip
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
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))
|
|
|
|
import scanner
|
|
|
|
|
|
def gzip_bytes(lines):
|
|
output = io.BytesIO()
|
|
with gzip.GzipFile(fileobj=output, mode='wb') as archive:
|
|
for line in lines:
|
|
archive.write(line)
|
|
return output.getvalue()
|
|
|
|
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
self.headers = {'Content-Length': str(len(payload))}
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def iter_content(self, chunk_size=1024 * 1024):
|
|
for offset in range(0, len(self.payload), max(1, chunk_size)):
|
|
yield self.payload[offset:offset + chunk_size]
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
|
|
class GHArchiveBoundsTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False)
|
|
|
|
def test_gzip_expansion_line_and_event_bounds_fail_closed(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
path = os.path.join(temp_dir, 'bomb.json.gz')
|
|
Path(path).write_bytes(gzip_bytes([b'x' * 4096 + b'\n']))
|
|
with self.assertRaises(scanner.GHArchiveBoundsError):
|
|
scanner.validate_gharchive_gzip(
|
|
path,
|
|
decompressed_max_bytes=1024,
|
|
max_events=10,
|
|
max_line_bytes=8192,
|
|
)
|
|
with self.assertRaises(scanner.GHArchiveBoundsError):
|
|
scanner.validate_gharchive_gzip(
|
|
path,
|
|
decompressed_max_bytes=8192,
|
|
max_events=10,
|
|
max_line_bytes=512,
|
|
)
|
|
|
|
event_path = os.path.join(temp_dir, 'events.json.gz')
|
|
Path(event_path).write_bytes(gzip_bytes([b'{}\n', b'{}\n', b'{}\n']))
|
|
with self.assertRaises(scanner.GHArchiveBoundsError):
|
|
scanner.validate_gharchive_gzip(
|
|
event_path,
|
|
decompressed_max_bytes=8192,
|
|
max_events=2,
|
|
max_line_bytes=512,
|
|
)
|
|
|
|
def test_twenty_four_hours_many_events_obey_tiny_cache_and_repo_bounds(self):
|
|
events = []
|
|
for index in range(200):
|
|
event = {
|
|
'type': 'PushEvent',
|
|
'repo': {'name': f'owner/repo-{index}'},
|
|
'payload': {'ref': 'refs/heads/main', 'commits': []},
|
|
}
|
|
events.append(json.dumps(event, separators=(',', ':')).encode('utf-8') + b'\n')
|
|
payload = gzip_bytes(events)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
runtime_dir = os.path.join(temp_dir, 'runtime')
|
|
cache_dir = os.path.join(runtime_dir, 'gharchive')
|
|
scanner.ensure_private_directory(runtime_dir, reject_reparse=True)
|
|
scanner.ensure_private_directory(cache_dir, reject_reparse=True)
|
|
settings = {
|
|
'runtime_dir': runtime_dir,
|
|
'gharchive_cache_dir': cache_dir,
|
|
'gharchive_cache_max_items': 2,
|
|
'gharchive_cache_max_bytes': 1024 * 1024,
|
|
'gharchive_cache_min_free_bytes': 0,
|
|
'gharchive_download_max_bytes': 1024 * 1024,
|
|
'gharchive_decompressed_max_bytes': 4 * 1024 * 1024,
|
|
'gharchive_max_events': 1000,
|
|
'gharchive_max_line_bytes': 64 * 1024,
|
|
'gharchive_cache_lock_timeout_sec': 5,
|
|
}
|
|
with mock.patch.multiple(scanner.scan_config, **settings), \
|
|
mock.patch.object(scanner.requests, 'request', side_effect=lambda *_args, **_kwargs: Response(payload)) as request:
|
|
targets = scanner.fetch_github_archive_repos(
|
|
hours_back=24,
|
|
max_repos=3,
|
|
event_types=['PushEvent'],
|
|
request_timeout=1,
|
|
archive_cache_dir=cache_dir,
|
|
)
|
|
usage = scanner.gharchive_cache_usage(cache_dir)
|
|
self.assertEqual(len(targets), 3)
|
|
self.assertEqual(request.call_count, 24)
|
|
self.assertTrue(all(call.kwargs['proxies']['no_proxy'] == '*' for call in request.call_args_list))
|
|
self.assertLessEqual(usage['items'], 2)
|
|
self.assertLessEqual(usage['bytes'], 1024 * 1024)
|
|
|
|
def test_compressed_download_cap_leaves_no_cache_artifact(self):
|
|
payload = b'x' * 512
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
runtime_dir = os.path.join(temp_dir, 'runtime')
|
|
cache_dir = os.path.join(runtime_dir, 'gharchive')
|
|
scanner.ensure_private_directory(runtime_dir, reject_reparse=True)
|
|
scanner.ensure_private_directory(cache_dir, reject_reparse=True)
|
|
settings = {
|
|
'runtime_dir': runtime_dir,
|
|
'gharchive_cache_dir': cache_dir,
|
|
'gharchive_cache_max_items': 2,
|
|
'gharchive_cache_max_bytes': 4096,
|
|
'gharchive_cache_min_free_bytes': 0,
|
|
'gharchive_download_max_bytes': 128,
|
|
'gharchive_decompressed_max_bytes': 4096,
|
|
'gharchive_max_events': 100,
|
|
'gharchive_max_line_bytes': 1024,
|
|
'gharchive_cache_lock_timeout_sec': 2,
|
|
}
|
|
with mock.patch.multiple(scanner.scan_config, **settings), \
|
|
mock.patch.object(scanner.requests, 'request', return_value=Response(payload)):
|
|
with self.assertRaises(scanner.ApiRequestError):
|
|
scanner.cached_gharchive_hour(
|
|
scanner.datetime(2026, 7, 19, 1, tzinfo=scanner.timezone.utc),
|
|
cache_dir,
|
|
request_timeout=1,
|
|
retries=1,
|
|
)
|
|
self.assertEqual(list(Path(cache_dir).glob('*.json.gz')), [])
|
|
|
|
def test_eviction_skips_an_active_oldest_cache_entry(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
runtime_dir = os.path.join(temp_dir, 'runtime')
|
|
cache_dir = os.path.join(runtime_dir, 'gharchive')
|
|
scanner.ensure_private_directory(runtime_dir, reject_reparse=True)
|
|
scanner.ensure_private_directory(cache_dir, reject_reparse=True)
|
|
first = os.path.join(cache_dir, '2026-07-19-01.json.gz')
|
|
hour = 2
|
|
second = os.path.join(cache_dir, f'2026-07-19-{hour:02d}.json.gz')
|
|
while scanner.gharchive_item_lock_path(cache_dir, first) == scanner.gharchive_item_lock_path(cache_dir, second):
|
|
hour += 1
|
|
second = os.path.join(cache_dir, f'2026-07-19-{hour:02d}.json.gz')
|
|
for index, path in enumerate((first, second), 1):
|
|
Path(path).write_bytes(gzip_bytes([b'{}\n']))
|
|
scanner.harden_private_file(path)
|
|
os.utime(path, (index, index))
|
|
settings = {
|
|
'runtime_dir': runtime_dir,
|
|
'gharchive_cache_dir': cache_dir,
|
|
'gharchive_cache_max_items': 2,
|
|
'gharchive_cache_max_bytes': 1024 * 1024,
|
|
'gharchive_cache_min_free_bytes': 0,
|
|
}
|
|
held = scanner.PrivateFileLock(scanner.gharchive_item_lock_path(cache_dir, first)).acquire()
|
|
try:
|
|
with mock.patch.multiple(scanner.scan_config, **settings):
|
|
scanner._gharchive_evict_for_capacity(cache_dir, required_items=1)
|
|
finally:
|
|
held.release()
|
|
self.assertTrue(os.path.exists(first))
|
|
self.assertFalse(os.path.exists(second))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|