sashatrask
899 lines
40 KiB
Python
899 lines
40 KiB
Python
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sqlite3
|
|
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 db_backend
|
|
import scanner_db
|
|
from scanner_db import RuntimeSafetySchemaError, ScannerDB, migrate_runtime_safety_schema
|
|
|
|
|
|
class DockerDepthExperimentSchemaTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.environment = mock.patch.dict(
|
|
os.environ,
|
|
{'SCANNER_DB_URL': '', 'DATABASE_URL': ''},
|
|
)
|
|
self.environment.start()
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.db = ScannerDB(db_path=os.path.join(self.temp.name, 'scanner.db'))
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
self.temp.cleanup()
|
|
self.environment.stop()
|
|
|
|
def _insert_repository(self, query, repository, resolver_state='pending'):
|
|
now = scanner_db.utc_now_iso()
|
|
cursor = self.db.conn.execute(
|
|
'''INSERT INTO target_queue(
|
|
source, platform, query, target, normalized_target, status,
|
|
resolver_state, created_at, updated_at
|
|
) VALUES ('dockerhub', 'docker', ?, ?, ?, 'deferred', ?, ?, ?)''',
|
|
(query, repository, repository, resolver_state, now, now),
|
|
)
|
|
self.db.conn.commit()
|
|
return cursor.lastrowid
|
|
|
|
def _insert_experiment(self, state='held', fenced=False):
|
|
now = scanner_db.utc_now_iso()
|
|
digest = 'a' * 64
|
|
cursor = self.db.conn.execute(
|
|
'''INSERT INTO docker_depth_experiments(
|
|
experiment_key, source, state, config_sha256,
|
|
ordered_queries_sha256, selector_version, selector_sha256,
|
|
provenance_policy_sha256, query_count, repositories_per_query,
|
|
images_per_repository, target_limit, fence_owner, fence_token,
|
|
fence_expires_at, created_at, updated_at
|
|
) VALUES (?, 'dockerhub', ?, ?, ?, 'selector-v1', ?, ?, 1, 1, 1, 1,
|
|
?, ?, ?, ?, ?)''',
|
|
(
|
|
'fixture-experiment', state, digest, digest, digest, digest,
|
|
'owner' if fenced else None, 'token' if fenced else None,
|
|
now if fenced else None, now, now,
|
|
),
|
|
)
|
|
self.db.conn.commit()
|
|
return cursor.lastrowid
|
|
|
|
def _remove_latest_marker(self):
|
|
self.db.conn.execute(
|
|
'DELETE FROM runtime_schema_migrations WHERE version = ?',
|
|
(scanner_db.PIPELINE_MIGRATION_VERSIONS[-1],),
|
|
)
|
|
self.db.conn.commit()
|
|
|
|
def test_fresh_sqlite_schema_has_every_registered_model_and_marker(self):
|
|
self.assertEqual(
|
|
scanner_db.PIPELINE_MIGRATION_VERSIONS[-1],
|
|
scanner_db.REMOTE_ASSIGNMENT_CAPACITY_MIGRATION,
|
|
)
|
|
marker = self.db.conn.execute(
|
|
'''SELECT version, code_sha256 FROM runtime_schema_migrations
|
|
WHERE version = ?''',
|
|
(scanner_db.PIPELINE_MIGRATION_VERSIONS[-1],),
|
|
).fetchone()
|
|
self.assertIsNotNone(marker)
|
|
self.assertEqual(
|
|
marker['code_sha256'],
|
|
hashlib.sha256(scanner_db.PIPELINE_SCHEMA_SQL.encode('utf-8')).hexdigest(),
|
|
)
|
|
migration_rows = self.db.conn.execute(
|
|
'SELECT version FROM runtime_schema_migrations'
|
|
).fetchall()
|
|
self.assertEqual(
|
|
{row['version'] for row in migration_rows},
|
|
set(scanner_db.PIPELINE_MIGRATION_VERSIONS),
|
|
)
|
|
|
|
for table in scanner_db.DOCKER_DEPTH_EXPERIMENT_TABLES:
|
|
with self.subTest(table=table):
|
|
specs = scanner_db.RUNTIME_TABLE_SPECS[table]
|
|
details = self.db.conn.table_column_details(table)
|
|
self.assertEqual(set(details), set(specs))
|
|
for column, (expected_type, expected_not_null) in specs.items():
|
|
self.assertTrue(
|
|
scanner_db._schema_type_matches(
|
|
details[column]['type'], expected_type, False,
|
|
),
|
|
f'{table}.{column} type',
|
|
)
|
|
self.assertEqual(
|
|
bool(details[column]['not_null']), bool(expected_not_null),
|
|
f'{table}.{column} nullability',
|
|
)
|
|
primary_key = [
|
|
column for column in specs if details[column]['primary_key']
|
|
]
|
|
self.assertEqual(primary_key, scanner_db.RUNTIME_PRIMARY_KEYS[table])
|
|
generated = scanner_db.GENERATED_ID_COLUMNS.get(table)
|
|
if generated:
|
|
self.assertTrue(scanner_db._column_generates_id(details[generated], False))
|
|
|
|
skip_columns = set(self.db.conn.table_columns(
|
|
'docker_depth_experiment_candidate_skips'
|
|
))
|
|
self.assertTrue(
|
|
{'target', 'normalized_target', 'repository'}.isdisjoint(skip_columns)
|
|
)
|
|
refund_columns = set(self.db.conn.table_columns(
|
|
'docker_depth_resolver_attempt_refunds'
|
|
))
|
|
self.assertTrue(
|
|
{'target', 'normalized_target', 'repository', 'last_error_code'}.isdisjoint(
|
|
refund_columns
|
|
)
|
|
)
|
|
disposition_columns = set(self.db.conn.table_columns(
|
|
'docker_depth_resolver_dispositions'
|
|
))
|
|
self.assertTrue(
|
|
{'target', 'normalized_target', 'repository', 'last_error_code'}.isdisjoint(
|
|
disposition_columns
|
|
)
|
|
)
|
|
|
|
policy = self.db.conn.table_column_details('target_queue_policy_events')
|
|
self.assertIn('experiment_id', policy)
|
|
self.assertFalse(policy['experiment_id']['not_null'])
|
|
self.assertTrue({
|
|
'discovery_retry_queue',
|
|
'target_queue_policy_events',
|
|
'docker_depth_experiment_candidate_skips',
|
|
}.issubset(scanner_db.DOCKER_DEPTH_EXPERIMENT_CHECK_SPECS))
|
|
for table, expected_constraints in scanner_db.DOCKER_DEPTH_EXPERIMENT_CHECK_SPECS.items():
|
|
constraints = self.db.conn.table_check_constraints(table)
|
|
for name, expression in expected_constraints.items():
|
|
with self.subTest(table=table, constraint=name):
|
|
self.assertIn(name, constraints)
|
|
self.assertTrue(constraints[name]['valid'])
|
|
self.assertEqual(
|
|
scanner_db._normalized_check_expression(
|
|
constraints[name]['expression'],
|
|
),
|
|
scanner_db._normalized_check_expression(expression),
|
|
)
|
|
hash_tables = set(scanner_db.DOCKER_DEPTH_EXPERIMENT_TABLES) | {
|
|
'discovery_retry_queue', 'target_queue_policy_events',
|
|
}
|
|
for table in hash_tables:
|
|
expressions = ' '.join(
|
|
scanner_db.DOCKER_DEPTH_EXPERIMENT_CHECK_SPECS.get(table, {}).values()
|
|
)
|
|
for column in scanner_db.RUNTIME_TABLE_SPECS[table]:
|
|
if not column.endswith('_sha256') and not (
|
|
table == 'discovery_retry_queue' and column == 'work_key'
|
|
):
|
|
continue
|
|
with self.subTest(table=table, sha256_column=column):
|
|
self.assertIn(scanner_db._sql_sha256_check(column), expressions)
|
|
self.assertTrue(self.db.runtime_safety_schema_available(), self.db.last_error)
|
|
self.assertTrue(self.db.pipeline_schema_available(), self.db.last_error)
|
|
retry_columns = self.db.conn.table_column_details('discovery_retry_queue')
|
|
self.assertIn('source_cycle_id', retry_columns)
|
|
self.assertFalse(retry_columns['source_cycle_id']['not_null'])
|
|
|
|
def test_sqlite_scarcity_cohort_migration_backfills_existing_queries(self):
|
|
table_sql = re.search(
|
|
r'CREATE TABLE IF NOT EXISTS docker_depth_experiment_queries '
|
|
r'\(.*?\n\);',
|
|
scanner_db.DOCKER_DEPTH_EXPERIMENT_SCHEMA_SQL,
|
|
re.DOTALL,
|
|
).group(0)
|
|
legacy_table_sql = table_sql.replace(
|
|
' selected_repository_count INTEGER NOT NULL DEFAULT 0,\n',
|
|
'',
|
|
)
|
|
legacy_table_sql = re.sub(
|
|
r' CONSTRAINT docker_depth_experiment_queries_selection_check '
|
|
r'CHECK \([^\n]+\),\n',
|
|
'',
|
|
legacy_table_sql,
|
|
)
|
|
self.assertNotIn('selected_repository_count', legacy_table_sql)
|
|
legacy_path = os.path.join(self.temp.name, 'legacy-scarcity.db')
|
|
legacy_db = ScannerDB(db_path=legacy_path, initialize=False)
|
|
try:
|
|
legacy_db.conn.executescript(
|
|
scanner_db.SCHEMA_SQL.replace(table_sql, legacy_table_sql)
|
|
)
|
|
now = scanner_db.utc_now_iso()
|
|
digest = '7' * 64
|
|
experiment_id = legacy_db.conn.execute(
|
|
'''INSERT INTO docker_depth_experiments(
|
|
experiment_key, source, state, config_sha256,
|
|
ordered_queries_sha256, selector_version, selector_sha256,
|
|
provenance_policy_sha256, query_count,
|
|
repositories_per_query, images_per_repository, target_limit,
|
|
created_at, updated_at
|
|
) VALUES ('sqlite-scarcity', 'dockerhub', 'held', ?, ?,
|
|
'selector-v1', ?, ?, 1, 10, 10, 1200, ?, ?)''',
|
|
(digest, digest, digest, digest, now, now),
|
|
).lastrowid
|
|
legacy_db.conn.execute(
|
|
'''INSERT INTO docker_depth_experiment_queries(
|
|
experiment_id, source, query_ordinal, query, query_sha256,
|
|
required_repository_count, created_at
|
|
) VALUES (?, 'dockerhub', 0, 'scarcity-query', ?, 7, ?)''',
|
|
(experiment_id, digest, now),
|
|
)
|
|
legacy_db.conn.commit()
|
|
|
|
self.assertTrue(migrate_runtime_safety_schema(
|
|
legacy_db, initialize_base=True,
|
|
))
|
|
self.assertTrue(migrate_runtime_safety_schema(legacy_db))
|
|
|
|
row = legacy_db.conn.execute(
|
|
'''SELECT required_repository_count, selected_repository_count
|
|
FROM docker_depth_experiment_queries
|
|
WHERE experiment_id = ? AND query_ordinal = 0''',
|
|
(experiment_id,),
|
|
).fetchone()
|
|
self.assertEqual(
|
|
(row['required_repository_count'], row['selected_repository_count']),
|
|
(7, 7),
|
|
)
|
|
self.assertIn(
|
|
'docker_depth_experiment_queries_selection_check',
|
|
legacy_db.conn.table_check_constraints(
|
|
'docker_depth_experiment_queries'
|
|
),
|
|
)
|
|
self.assertTrue(
|
|
legacy_db.runtime_safety_schema_available(), legacy_db.last_error,
|
|
)
|
|
finally:
|
|
legacy_db.close()
|
|
|
|
def test_pre_experiment_pipeline_recovery_skips_absent_binding_schema(self):
|
|
self.db.conn.execute(
|
|
'DELETE FROM runtime_schema_migrations WHERE version = ?',
|
|
(scanner_db.DOCKER_DEPTH_EXPERIMENT_MIGRATION,),
|
|
)
|
|
self.db.conn.execute('DROP TABLE docker_depth_experiment_scan_bindings')
|
|
self.db.conn.commit()
|
|
if hasattr(self.db, '_docker_depth_experiment_schema_installed_cache'):
|
|
del self.db._docker_depth_experiment_schema_installed_cache
|
|
|
|
self.assertIsNone(
|
|
self.db._lock_docker_depth_experiment_for_reservation(1)
|
|
)
|
|
self.assertIsNone(
|
|
self.db._docker_depth_binding_for_reservation_locked({'id': 1})
|
|
)
|
|
|
|
def test_migration_adds_nullable_retry_origin_cycle_and_foreign_key(self):
|
|
table_sql = re.search(
|
|
r'CREATE TABLE IF NOT EXISTS discovery_retry_queue \(.*?\n\);',
|
|
scanner_db.PIPELINE_SCHEMA_SQL,
|
|
re.DOTALL,
|
|
).group(0)
|
|
legacy_sql = table_sql.replace(' source_cycle_id INTEGER,\n', '')
|
|
legacy_sql = re.sub(
|
|
r',\n FOREIGN KEY\(source_cycle_id\) REFERENCES source_cycles\(id\)\n\);$',
|
|
'\n);',
|
|
legacy_sql,
|
|
)
|
|
self.assertNotIn('source_cycle_id', legacy_sql)
|
|
legacy_path = os.path.join(self.temp.name, 'legacy-retry.db')
|
|
connection = sqlite3.connect(legacy_path)
|
|
try:
|
|
connection.executescript(legacy_sql)
|
|
finally:
|
|
connection.close()
|
|
|
|
legacy_db = ScannerDB(db_path=legacy_path, initialize=False)
|
|
try:
|
|
self.assertTrue(migrate_runtime_safety_schema(
|
|
legacy_db, initialize_base=True,
|
|
))
|
|
details = legacy_db.conn.table_column_details('discovery_retry_queue')
|
|
self.assertIn('source_cycle_id', details)
|
|
self.assertFalse(details['source_cycle_id']['not_null'])
|
|
foreign_keys = legacy_db.conn.table_foreign_keys('discovery_retry_queue')
|
|
self.assertTrue(any(
|
|
foreign_key['columns'] == ['source_cycle_id']
|
|
and foreign_key['referenced_table'] == 'source_cycles'
|
|
and foreign_key['referenced_columns'] == ['id']
|
|
for foreign_key in foreign_keys.values()
|
|
))
|
|
marker = legacy_db.conn.execute(
|
|
'SELECT version FROM runtime_schema_migrations WHERE version = ?',
|
|
('20260910_19_docker_retry_provenance',),
|
|
).fetchone()
|
|
self.assertIsNotNone(marker)
|
|
finally:
|
|
legacy_db.close()
|
|
|
|
def test_migration_adds_reason_enum_to_empty_legacy_attribution_table(self):
|
|
table_sql = re.search(
|
|
r'CREATE TABLE IF NOT EXISTS docker_finding_layer_attributions \(.*?\n\);',
|
|
scanner_db.DOCKER_DEPTH_EXPERIMENT_SCHEMA_SQL,
|
|
re.DOTALL,
|
|
).group(0)
|
|
legacy_sql = table_sql.replace(' unattributed_reason TEXT,\n', '')
|
|
legacy_sql = re.sub(
|
|
r' CONSTRAINT docker_finding_layer_attributions_reason_check '
|
|
r'CHECK \([^\n]+\),\n',
|
|
'',
|
|
legacy_sql,
|
|
)
|
|
self.assertNotIn('unattributed_reason', legacy_sql)
|
|
self.db.conn.execute('DROP TABLE docker_finding_layer_attributions')
|
|
self.db.conn.executescript(legacy_sql)
|
|
self._remove_latest_marker()
|
|
|
|
self.assertTrue(migrate_runtime_safety_schema(self.db))
|
|
|
|
details = self.db.conn.table_column_details(
|
|
'docker_finding_layer_attributions'
|
|
)
|
|
constraints = self.db.conn.table_check_constraints(
|
|
'docker_finding_layer_attributions'
|
|
)
|
|
self.assertIn('unattributed_reason', details)
|
|
self.assertIn(
|
|
'docker_finding_layer_attributions_reason_check', constraints,
|
|
)
|
|
self.assertTrue(self.db.runtime_safety_schema_available(), self.db.last_error)
|
|
|
|
def test_registered_foreign_keys_types_and_indexes_are_strictly_validated(self):
|
|
tables = set(scanner_db.DOCKER_DEPTH_EXPERIMENT_TABLES) | {
|
|
'discovery_retry_queue', 'target_queue_policy_events',
|
|
}
|
|
for table in tables:
|
|
foreign_keys = self.db.conn.table_foreign_keys(table)
|
|
for columns, referenced_table, referenced_columns in scanner_db.REQUIRED_FOREIGN_KEYS.get(
|
|
table, ()
|
|
):
|
|
matches = [
|
|
foreign_key for foreign_key in foreign_keys.values()
|
|
if tuple(foreign_key['columns']) == columns
|
|
and foreign_key['referenced_table'] == referenced_table
|
|
and tuple(foreign_key['referenced_columns']) == referenced_columns
|
|
]
|
|
self.assertEqual(
|
|
len(matches), 1,
|
|
f'{table}{columns} -> {referenced_table}{referenced_columns}',
|
|
)
|
|
expected_actions = scanner_db._required_foreign_key_actions(table, columns)
|
|
self.assertEqual(
|
|
(matches[0]['update_action'], matches[0]['delete_action']),
|
|
expected_actions,
|
|
)
|
|
|
|
for table, name, columns, unique, predicate in scanner_db.DOCKER_DEPTH_EXPERIMENT_INDEX_SPECS:
|
|
index = self.db.conn.table_indexes(table).get(name)
|
|
self.assertIsNotNone(index, name)
|
|
self.assertEqual(index['columns'], list(columns), name)
|
|
self.assertEqual(bool(index['unique']), bool(unique), name)
|
|
self.assertEqual(
|
|
scanner_db._normalized_predicate(index['predicate']),
|
|
scanner_db._normalized_predicate(predicate),
|
|
name,
|
|
)
|
|
|
|
converted = db_backend._postgres_schema_sql(scanner_db.SCHEMA_SQL)
|
|
for table in scanner_db.DOCKER_DEPTH_EXPERIMENT_TABLES:
|
|
block = re.search(
|
|
rf'CREATE TABLE IF NOT EXISTS {table} \((.*?)\n\);',
|
|
converted,
|
|
re.DOTALL,
|
|
).group(1)
|
|
for column, (expected_type, _) in scanner_db.RUNTIME_TABLE_SPECS[table].items():
|
|
if expected_type in ('id', 'id_ref'):
|
|
self.assertRegex(block, rf'(?m)^ {column} BIGINT\b', table + '.' + column)
|
|
for table, expected_constraints in scanner_db.DOCKER_DEPTH_EXPERIMENT_CHECK_SPECS.items():
|
|
for name, expression in expected_constraints.items():
|
|
self.assertIn(
|
|
f'CONSTRAINT {name} CHECK ({expression})',
|
|
converted,
|
|
)
|
|
|
|
original = self.db.conn.table_column_details
|
|
|
|
def malformed(table):
|
|
details = original(table)
|
|
if table == 'docker_image_manifests':
|
|
details['target_queue_id'] = dict(details['target_queue_id'], type='text')
|
|
return details
|
|
|
|
with mock.patch.object(self.db.conn, 'table_column_details', side_effect=malformed):
|
|
self.assertFalse(self.db.runtime_safety_schema_available())
|
|
self.assertIn('docker_image_manifests.target_queue_id type', self.db.last_error)
|
|
|
|
original_foreign_keys = self.db.conn.table_foreign_keys
|
|
|
|
def missing_foreign_key(table):
|
|
foreign_keys = original_foreign_keys(table)
|
|
if table == 'docker_depth_experiment_scan_bindings':
|
|
return {
|
|
name: value for name, value in foreign_keys.items()
|
|
if value['columns'] != ['reservation_id']
|
|
}
|
|
return foreign_keys
|
|
|
|
with mock.patch.object(
|
|
self.db.conn, 'table_foreign_keys', side_effect=missing_foreign_key,
|
|
):
|
|
self.assertFalse(self.db.runtime_safety_schema_available())
|
|
self.assertIn(
|
|
'foreign key docker_depth_experiment_scan_bindings(reservation_id)',
|
|
self.db.last_error,
|
|
)
|
|
|
|
def wrong_foreign_key_action(table):
|
|
foreign_keys = original_foreign_keys(table)
|
|
if table == 'docker_depth_experiment_scan_bindings':
|
|
foreign_keys = {
|
|
name: dict(value)
|
|
for name, value in foreign_keys.items()
|
|
}
|
|
for foreign_key in foreign_keys.values():
|
|
if foreign_key['columns'] == ['experiment_target_id']:
|
|
foreign_key['delete_action'] = 'CASCADE'
|
|
return foreign_keys
|
|
|
|
with mock.patch.object(
|
|
self.db.conn,
|
|
'table_foreign_keys',
|
|
side_effect=wrong_foreign_key_action,
|
|
):
|
|
self.assertFalse(self.db.runtime_safety_schema_available())
|
|
self.assertIn(
|
|
'foreign key docker_depth_experiment_scan_bindings(experiment_target_id) '
|
|
'-> docker_depth_experiment_targets(id) actions ON UPDATE NO ACTION '
|
|
'ON DELETE NO ACTION',
|
|
self.db.last_error,
|
|
)
|
|
|
|
def test_every_canonical_check_is_fail_closed_when_malformed(self):
|
|
self.assertNotEqual(
|
|
scanner_db._normalized_check_expression('(a = 1 AND b = 2) OR c = 3'),
|
|
scanner_db._normalized_check_expression('a = 1 AND (b = 2 OR c = 3)'),
|
|
)
|
|
original = self.db.conn.table_check_constraints
|
|
for malformed_table, expected_constraints in (
|
|
scanner_db.DOCKER_DEPTH_EXPERIMENT_CHECK_SPECS.items()
|
|
):
|
|
for malformed_name in expected_constraints:
|
|
with self.subTest(table=malformed_table, constraint=malformed_name):
|
|
def malformed(table):
|
|
constraints = {
|
|
name: dict(value)
|
|
for name, value in original(table).items()
|
|
}
|
|
if table == malformed_table:
|
|
constraints[malformed_name]['expression'] = '1 = 1'
|
|
return constraints
|
|
|
|
with mock.patch.object(
|
|
self.db.conn,
|
|
'table_check_constraints',
|
|
side_effect=malformed,
|
|
):
|
|
self.assertFalse(self.db.runtime_safety_schema_available())
|
|
self.assertIn(malformed_name, self.db.last_error)
|
|
|
|
def test_sqlite_constraints_enforce_hash_state_cap_and_fence_shape(self):
|
|
experiment_id = self._insert_experiment()
|
|
malformed_updates = (
|
|
('config_sha256 = ?', ('g' * 64,)),
|
|
('state = ?', ('unknown',)),
|
|
('query_count = 2', ()),
|
|
('fence_owner = ?', ('owner-only',)),
|
|
)
|
|
for assignment, params in malformed_updates:
|
|
with self.subTest(assignment=assignment):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.db.conn.execute(
|
|
f'UPDATE docker_depth_experiments SET {assignment} WHERE id = ?',
|
|
params + (experiment_id,),
|
|
)
|
|
self.db.conn.rollback()
|
|
|
|
with self.assertRaisesRegex(
|
|
sqlite3.IntegrityError,
|
|
'docker_finding_layer_attributions_evidence_check',
|
|
):
|
|
self.db.conn.execute(
|
|
'''INSERT INTO docker_finding_layer_attributions(
|
|
scan_binding_id, finding_id, manifest_layer_id,
|
|
attribution_state, reported_layer_digest,
|
|
position_from_base, position_from_top, created_at
|
|
) VALUES (1, 1, 1, 'exact', 'sha256:fixture', NULL, 1, ?)''',
|
|
(scanner_db.utc_now_iso(),),
|
|
)
|
|
self.db.conn.rollback()
|
|
|
|
retry = self.db.enqueue_discovery_retry(
|
|
'dockerhub', 'constraint-query', 'b' * 64, 'ordinary', 'page', 1, 1,
|
|
error_category='network',
|
|
)
|
|
for column, malformed_hash in (
|
|
('work_key', 'A' * 64),
|
|
('policy_sha256', 'g' * 64),
|
|
):
|
|
with self.subTest(retry_hash=column), self.assertRaises(
|
|
sqlite3.IntegrityError
|
|
):
|
|
self.db.conn.execute(
|
|
f'UPDATE discovery_retry_queue SET {column} = ? WHERE id = ?',
|
|
(malformed_hash, retry['id']),
|
|
)
|
|
self.db.conn.rollback()
|
|
with self.assertRaisesRegex(
|
|
sqlite3.IntegrityError, 'discovery_retry_queue_lifecycle_check',
|
|
):
|
|
self.db.conn.execute(
|
|
"UPDATE discovery_retry_queue SET status = 'leased' WHERE id = ?",
|
|
(retry['id'],),
|
|
)
|
|
self.db.conn.rollback()
|
|
|
|
queue_id = self._insert_repository('Policy Query', 'fixture/policy')
|
|
now = scanner_db.utc_now_iso()
|
|
event = self.db.conn.execute(
|
|
'''INSERT INTO target_queue_policy_events(
|
|
queue_id, action, prior_status, next_status, source, platform,
|
|
query, reason_code, config_sha256, policy_sha256,
|
|
manifest_sha256, review_audit_sha256, experiment_id,
|
|
prior_updated_at, created_at
|
|
) VALUES (?, 'cold', 'deferred', 'cold', 'dockerhub', 'docker',
|
|
'Policy Query', 'docker_depth_experiment_hold',
|
|
?, ?, ?, ?, ?, ?, ?) RETURNING id''',
|
|
(
|
|
queue_id, 'c' * 64, 'd' * 64, 'e' * 64, 'f' * 64,
|
|
experiment_id, now, now,
|
|
),
|
|
).fetchone()
|
|
self.db.conn.commit()
|
|
for assignment, params, constraint in (
|
|
('config_sha256 = ?', ('A' * 64,), 'target_queue_policy_events_hash_check'),
|
|
(
|
|
"reason_code = 'query_not_in_canonical_policy'", (),
|
|
'target_queue_policy_events_experiment_ownership_check',
|
|
),
|
|
(
|
|
"next_status = 'pending'", (),
|
|
'target_queue_policy_events_transition_check',
|
|
),
|
|
):
|
|
with self.subTest(policy_assignment=assignment), self.assertRaisesRegex(
|
|
sqlite3.IntegrityError, constraint,
|
|
):
|
|
self.db.conn.execute(
|
|
f'UPDATE target_queue_policy_events SET {assignment} WHERE id = ?',
|
|
params + (event['id'],),
|
|
)
|
|
self.db.conn.rollback()
|
|
|
|
generic_queue_id = self._insert_repository(
|
|
'Generic Policy Query', 'fixture/generic-policy',
|
|
)
|
|
self.db.conn.execute(
|
|
'''INSERT INTO target_queue_policy_events(
|
|
queue_id, action, prior_status, next_status, source, platform,
|
|
query, reason_code, config_sha256, policy_sha256,
|
|
manifest_sha256, review_audit_sha256, prior_updated_at, created_at
|
|
) VALUES (?, 'cold', 'deferred', 'cold', 'dockerhub', 'docker',
|
|
'Generic Policy Query', 'query_not_in_canonical_policy',
|
|
?, ?, ?, ?, ?, ?)''',
|
|
(
|
|
generic_queue_id, '1' * 64, '2' * 64, '3' * 64, '4' * 64,
|
|
now, now,
|
|
),
|
|
)
|
|
self.db.conn.commit()
|
|
|
|
with self.assertRaisesRegex(
|
|
sqlite3.IntegrityError,
|
|
'docker_finding_layer_attributions_reason_check',
|
|
):
|
|
self.db.conn.execute(
|
|
'''INSERT INTO docker_finding_layer_attributions(
|
|
scan_binding_id, finding_id, attribution_state, created_at
|
|
) VALUES (1, 1, 'unattributed', ?)''',
|
|
(scanner_db.utc_now_iso(),),
|
|
)
|
|
self.db.conn.rollback()
|
|
|
|
def test_migration_rejects_malformed_sqlite_check(self):
|
|
cases = (
|
|
('docker_depth_experiments', 'docker_depth_experiments_identity_check'),
|
|
('discovery_retry_queue', 'discovery_retry_queue_lifecycle_check'),
|
|
('target_queue_policy_events', 'target_queue_policy_events_transition_check'),
|
|
)
|
|
for table, name in cases:
|
|
with self.subTest(table=table, constraint=name):
|
|
expression = scanner_db.DOCKER_DEPTH_EXPERIMENT_CHECK_SPECS[table][name]
|
|
row = self.db.conn.execute(
|
|
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?",
|
|
(table,),
|
|
).fetchone()
|
|
original_sql = row['sql']
|
|
expected = f'CONSTRAINT {name} CHECK ({expression})'
|
|
self.assertIn(expected, original_sql)
|
|
malformed = original_sql.replace(
|
|
expected, f'CONSTRAINT {name} CHECK (1 = 1)',
|
|
)
|
|
schema_version = self.db.conn.execute(
|
|
'PRAGMA schema_version'
|
|
).fetchone()[0]
|
|
self.db.conn.execute('PRAGMA writable_schema = ON')
|
|
self.db.conn.execute(
|
|
"UPDATE sqlite_master SET sql = ? WHERE type = 'table' AND name = ?",
|
|
(malformed, table),
|
|
)
|
|
self.db.conn.execute('PRAGMA writable_schema = OFF')
|
|
self.db.conn.execute(f'PRAGMA schema_version = {schema_version + 1}')
|
|
self.db.conn.commit()
|
|
|
|
self.assertFalse(self.db.runtime_safety_schema_available())
|
|
self.assertIn(name, self.db.last_error)
|
|
with self.assertRaisesRegex(
|
|
RuntimeSafetySchemaError, 'noncanonical check constraint',
|
|
):
|
|
migrate_runtime_safety_schema(self.db)
|
|
|
|
schema_version = self.db.conn.execute(
|
|
'PRAGMA schema_version'
|
|
).fetchone()[0]
|
|
self.db.conn.execute('PRAGMA writable_schema = ON')
|
|
self.db.conn.execute(
|
|
"UPDATE sqlite_master SET sql = ? WHERE type = 'table' AND name = ?",
|
|
(original_sql, table),
|
|
)
|
|
self.db.conn.execute('PRAGMA writable_schema = OFF')
|
|
self.db.conn.execute(f'PRAGMA schema_version = {schema_version + 1}')
|
|
self.db.conn.commit()
|
|
self.assertTrue(
|
|
self.db.runtime_safety_schema_available(), self.db.last_error,
|
|
)
|
|
|
|
def test_migration_rejects_wrong_sqlite_fk_actions(self):
|
|
original_foreign_keys = self.db.conn.table_foreign_keys
|
|
|
|
def wrong_action(foreign_table):
|
|
foreign_keys = {
|
|
key: dict(value)
|
|
for key, value in original_foreign_keys(foreign_table).items()
|
|
}
|
|
if foreign_table == 'docker_depth_experiment_scan_bindings':
|
|
for foreign_key in foreign_keys.values():
|
|
if foreign_key['columns'] == ['experiment_target_id']:
|
|
foreign_key['update_action'] = 'CASCADE'
|
|
return foreign_keys
|
|
|
|
with mock.patch.object(
|
|
self.db.conn, 'table_foreign_keys', side_effect=wrong_action,
|
|
):
|
|
with self.assertRaisesRegex(RuntimeSafetySchemaError, 'foreign key actions'):
|
|
migrate_runtime_safety_schema(self.db)
|
|
|
|
def test_migration_is_idempotent_and_repairs_registered_index_shape(self):
|
|
self.db.conn.execute('DROP TABLE docker_finding_layer_attributions')
|
|
self.db.conn.execute('DROP INDEX idx_docker_repository_provenance_eligible')
|
|
self.db.conn.execute(
|
|
'''CREATE INDEX idx_docker_repository_provenance_eligible
|
|
ON docker_repository_query_provenance(source)'''
|
|
)
|
|
self.db.conn.commit()
|
|
self.assertFalse(self.db.runtime_safety_schema_available())
|
|
self.assertIn('index idx_docker_repository_provenance_eligible', self.db.last_error)
|
|
self._remove_latest_marker()
|
|
|
|
self.assertTrue(migrate_runtime_safety_schema(self.db))
|
|
self.assertTrue(migrate_runtime_safety_schema(self.db))
|
|
self.assertTrue(self.db.runtime_safety_schema_available(), self.db.last_error)
|
|
marker_count = self.db.conn.execute(
|
|
'SELECT COUNT(*) AS count FROM runtime_schema_migrations WHERE version = ?',
|
|
(scanner_db.PIPELINE_MIGRATION_VERSIONS[-1],),
|
|
).fetchone()['count']
|
|
self.assertEqual(marker_count, 1)
|
|
index = self.db.conn.table_indexes('docker_repository_query_provenance')[
|
|
'idx_docker_repository_provenance_eligible'
|
|
]
|
|
self.assertEqual(
|
|
index['columns'],
|
|
['source', 'query', 'fresh_coverage_eligible', 'best_search_rank', 'repository_queue_id'],
|
|
)
|
|
|
|
def test_hold_snapshot_queue_scan_index_is_additive_and_repaired(self):
|
|
self.db.conn.execute('DROP INDEX idx_target_scans_queue_id')
|
|
self._remove_latest_marker()
|
|
self.db.conn.commit()
|
|
|
|
self.assertFalse(self.db.runtime_safety_schema_available())
|
|
self.assertIn('idx_target_scans_queue_id', self.db.last_error)
|
|
self.assertTrue(migrate_runtime_safety_schema(self.db))
|
|
self.assertTrue(migrate_runtime_safety_schema(self.db))
|
|
self.assertTrue(self.db.runtime_safety_schema_available(), self.db.last_error)
|
|
index = self.db.conn.table_indexes('target_scans')[
|
|
'idx_target_scans_queue_id'
|
|
]
|
|
self.assertEqual(index['columns'], ['queue_id'])
|
|
|
|
def test_legacy_first_inserter_seed_is_bounded_idempotent_and_ineligible(self):
|
|
legacy_id = self._insert_repository(
|
|
'Exact Legacy Query', 'fixture/legacy', resolver_state=None,
|
|
)
|
|
fresh_id = self._insert_repository('Exact Fresh Query', 'fixture/fresh')
|
|
immutable_id = self._insert_repository(
|
|
'Image Query', 'fixture/image@sha256:' + ('b' * 64), resolver_state=None,
|
|
)
|
|
now = scanner_db.utc_now_iso()
|
|
discovery_pass = self.db.conn.execute(
|
|
'''INSERT INTO docker_discovery_passes(
|
|
pass_token, source, pass_kind, policy_sha256,
|
|
ordered_queries_sha256, expected_query_count,
|
|
completed_query_count, state, started_at, completed_at,
|
|
created_at, updated_at
|
|
) VALUES ('fixture-pass-token', 'dockerhub', 'deep', ?, ?, 1, 1,
|
|
'complete', ?, ?, ?, ?) RETURNING id''',
|
|
('c' * 64, 'd' * 64, now, now, now, now),
|
|
).fetchone()
|
|
discovery_page = self.db.conn.execute(
|
|
'''INSERT INTO docker_discovery_pages(
|
|
pass_id, query, query_ordinal, page_number, result_count,
|
|
admitted_count, query_complete, admission_kind, page_sha256,
|
|
observed_at, created_at
|
|
) VALUES (?, 'Exact Fresh Query', 0, 1, 1, 1, 1, 'main', ?, ?, ?)
|
|
RETURNING id''',
|
|
(discovery_pass['id'], 'e' * 64, now, now),
|
|
).fetchone()
|
|
self.db.conn.execute(
|
|
'''INSERT INTO docker_repository_query_provenance(
|
|
source, query, repository_queue_id, provenance_kind,
|
|
first_observed_at, last_observed_at, first_search_rank,
|
|
best_search_rank, last_search_rank, first_page_id, last_page_id,
|
|
first_policy_sha256, last_policy_sha256,
|
|
observation_count, fresh_observation_count,
|
|
fresh_complete_observation_count, fresh_coverage_eligible,
|
|
created_at, updated_at
|
|
) VALUES ('dockerhub', 'Exact Fresh Query', ?, 'fresh_page', ?, ?,
|
|
1, 1, 1, ?, ?, ?, ?, 1, 1, 1, 1, ?, ?)''',
|
|
(
|
|
fresh_id, now, now, discovery_page['id'], discovery_page['id'],
|
|
'c' * 64, 'c' * 64, now, now,
|
|
),
|
|
)
|
|
self.db.conn.execute(
|
|
'''INSERT INTO docker_repository_query_observations(
|
|
page_id, repository_queue_id, source, query, search_rank,
|
|
observed_at
|
|
) VALUES (?, ?, 'dockerhub', 'Exact Fresh Query', 1, ?)''',
|
|
(discovery_page['id'], fresh_id, now),
|
|
)
|
|
self.db.conn.commit()
|
|
|
|
self.db.conn.execute('SAVEPOINT bounded_legacy_seed')
|
|
with self.assertRaisesRegex(RuntimeSafetySchemaError, 'reviewed 1-row bound'):
|
|
scanner_db._migration_seed_legacy_docker_provenance(
|
|
self.db.conn, max_rows=1, page_size=1,
|
|
)
|
|
self.db.conn.execute('ROLLBACK TO SAVEPOINT bounded_legacy_seed')
|
|
self.db.conn.execute('RELEASE SAVEPOINT bounded_legacy_seed')
|
|
|
|
self._remove_latest_marker()
|
|
migrate_runtime_safety_schema(self.db)
|
|
migrate_runtime_safety_schema(self.db)
|
|
legacy = self.db.conn.execute(
|
|
'''SELECT provenance_kind, observation_count, fresh_observation_count,
|
|
fresh_complete_observation_count, fresh_coverage_eligible,
|
|
first_page_id, last_page_id
|
|
FROM docker_repository_query_provenance
|
|
WHERE repository_queue_id = ?''',
|
|
(legacy_id,),
|
|
).fetchone()
|
|
self.assertEqual(
|
|
tuple(legacy),
|
|
('legacy_queue', 1, 0, 0, 0, None, None),
|
|
)
|
|
fresh = self.db.conn.execute(
|
|
'''SELECT provenance_kind, fresh_complete_observation_count,
|
|
fresh_coverage_eligible
|
|
FROM docker_repository_query_provenance
|
|
WHERE repository_queue_id = ?''',
|
|
(fresh_id,),
|
|
).fetchone()
|
|
self.assertEqual(tuple(fresh), ('fresh_page', 1, 1))
|
|
self.assertIsNone(self.db.conn.execute(
|
|
'''SELECT 1 FROM docker_repository_query_provenance
|
|
WHERE repository_queue_id = ?''',
|
|
(immutable_id,),
|
|
).fetchone())
|
|
self.assertEqual(
|
|
self.db.conn.execute(
|
|
'''SELECT COUNT(*) AS count FROM docker_repository_query_provenance
|
|
WHERE repository_queue_id = ?''',
|
|
(legacy_id,),
|
|
).fetchone()['count'],
|
|
1,
|
|
)
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.db.conn.execute(
|
|
'''UPDATE docker_repository_query_provenance
|
|
SET fresh_coverage_eligible = 1 WHERE repository_queue_id = ?''',
|
|
(legacy_id,),
|
|
)
|
|
self.db.conn.rollback()
|
|
|
|
def test_migration_quiescence_blocks_resolvers_authority_and_hold_fences(self):
|
|
queue_id = self._insert_repository('Fence Query', 'fixture/fenced')
|
|
self._remove_latest_marker()
|
|
self.db.conn.execute(
|
|
'''UPDATE target_queue SET resolver_state = 'resolving', resolver_token = 'token'
|
|
WHERE id = ?''',
|
|
(queue_id,),
|
|
)
|
|
self.db.conn.commit()
|
|
with self.assertRaisesRegex(RuntimeSafetySchemaError, 'docker_resolvers=1'):
|
|
migrate_runtime_safety_schema(self.db)
|
|
|
|
self.db.conn.execute(
|
|
'''UPDATE target_queue SET resolver_state = 'pending', resolver_token = NULL
|
|
WHERE id = ?''',
|
|
(queue_id,),
|
|
)
|
|
experiment_id = self._insert_experiment(state='held', fenced=True)
|
|
with self.assertRaisesRegex(RuntimeSafetySchemaError, 'experiment_authority_fences=1'):
|
|
migrate_runtime_safety_schema(self.db)
|
|
|
|
self.db.conn.execute(
|
|
'''UPDATE docker_depth_experiments
|
|
SET fence_owner = NULL, fence_token = NULL, fence_expires_at = NULL
|
|
WHERE id = ?''',
|
|
(experiment_id,),
|
|
)
|
|
now = scanner_db.utc_now_iso()
|
|
self.db.conn.execute(
|
|
'''INSERT INTO target_queue_policy_events(
|
|
queue_id, action, prior_status, next_status, source, platform,
|
|
query, reason_code, config_sha256, policy_sha256,
|
|
manifest_sha256, review_audit_sha256, experiment_id,
|
|
prior_updated_at, created_at
|
|
) VALUES (?, 'cold', 'pending', 'cold', 'dockerhub', 'docker',
|
|
'Fence Query', 'docker_depth_experiment_hold',
|
|
?, ?, ?, ?, ?, ?, ?)''',
|
|
(
|
|
queue_id, 'd' * 64, 'e' * 64, 'f' * 64, '1' * 64,
|
|
experiment_id, now, now,
|
|
),
|
|
)
|
|
self.db.conn.commit()
|
|
self.db.conn.execute(
|
|
"UPDATE docker_depth_experiments SET state = 'planned' WHERE id = ?",
|
|
(experiment_id,),
|
|
)
|
|
self.db.conn.commit()
|
|
with self.assertRaisesRegex(RuntimeSafetySchemaError, 'experiment_holds=1'):
|
|
migrate_runtime_safety_schema(self.db)
|
|
|
|
self.db.conn.execute(
|
|
"UPDATE docker_depth_experiments SET state = 'held' WHERE id = ?",
|
|
(experiment_id,),
|
|
)
|
|
self.db.conn.commit()
|
|
migrate_runtime_safety_schema(self.db)
|
|
marker = self.db.conn.execute(
|
|
'SELECT 1 FROM runtime_schema_migrations WHERE version = ?',
|
|
(scanner_db.PIPELINE_MIGRATION_VERSIONS[-1],),
|
|
).fetchone()
|
|
self.assertIsNotNone(marker)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|