4183 lines
187 KiB
Python
4183 lines
187 KiB
Python
import asyncio
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
import uuid
|
|
from unittest import mock
|
|
from urllib.parse import urlencode, urljoin
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_DIR = ROOT / 'app'
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
from starlette.testclient import TestClient
|
|
from starlette.datastructures import Headers
|
|
from starlette.requests import Request
|
|
from starlette.responses import Response
|
|
|
|
from admin_api import (
|
|
ADMIN_PREFIX, EDGE_MARKER_HEADER, OPERATOR_HEADER, SECURITY_HEADERS,
|
|
AdminAPIError, AdminService, _dispatch_managed_file_mutation,
|
|
_dispatch, _dispatch_runtime_document, _download_managed_file, _form_fields,
|
|
_managed_file_download_response, _ManagedFileStreamingResponse,
|
|
_parse_managed_file_content,
|
|
_trusted_operator, _validate_cap,
|
|
)
|
|
from managed_files import (
|
|
ManagedFileAccessError, ManagedFileDirectoryEntry, ManagedFileDownload,
|
|
ManagedFileIdentity, ManagedFileLimits, ManagedFileListing,
|
|
ManagedFileMutation, ManagedFilePermissions, ManagedFileRoot,
|
|
ManagedFileRootRegistry,
|
|
)
|
|
from scanner_db import (
|
|
RuntimeControlRevisionConflictError, RuntimeOperationIdentityConflictError,
|
|
)
|
|
from runtime_document import RuntimeDocumentError
|
|
from runtime_security import ensure_private_directory
|
|
from worker_api import WorkerAPIError, create_worker_app
|
|
from worker_contracts import (
|
|
MAX_DIAGNOSTIC_BODY_BYTES, MAX_DIAGNOSTIC_LOG_BYTES,
|
|
make_body_material, make_log_material,
|
|
)
|
|
|
|
|
|
ORIGIN = 'https://admin.example.test'
|
|
MARKER = 'fixture-edge-marker-value-32bytes-minimum'
|
|
OPERATOR = 'fixture.operator'
|
|
|
|
|
|
class FakeWorkerService:
|
|
def __init__(self, root):
|
|
self.bundle_root = root
|
|
self.max_bundle_bytes = 1024 * 1024
|
|
self.claim_retry_after_seconds = 5
|
|
|
|
def authenticate(self, authorization):
|
|
raise WorkerAPIError(401, 'unauthorized', 'worker credentials are invalid')
|
|
|
|
def reap(self):
|
|
return []
|
|
|
|
|
|
class AdminValidationTests(unittest.TestCase):
|
|
def test_assignment_cap_accepts_integer_zero(self):
|
|
self.assertEqual(_validate_cap(0), 0)
|
|
|
|
|
|
class FakeManagedFileTraversal:
|
|
def __init__(self):
|
|
self.files = {
|
|
'guide.txt': b'fixture managed content',
|
|
'nested/report.bin': b'\x00\xffreport',
|
|
}
|
|
self.calls = []
|
|
self.closed = 0
|
|
self.error_category = None
|
|
self.stream_downloads = False
|
|
self.last_snapshot = None
|
|
|
|
@staticmethod
|
|
def _identity(content):
|
|
return ManagedFileIdentity(hashlib.sha256(content).hexdigest(), len(content))
|
|
|
|
def _fail(self):
|
|
if self.error_category:
|
|
raise ManagedFileAccessError(self.error_category)
|
|
|
|
def list_directory(self, root_id, relative_path=None):
|
|
self._fail()
|
|
self.calls.append(('list', root_id, relative_path))
|
|
prefix = '' if relative_path is None else relative_path + '/'
|
|
entries = {}
|
|
for path, content in self.files.items():
|
|
if not path.startswith(prefix):
|
|
continue
|
|
remainder = path[len(prefix):]
|
|
name, separator, _tail = remainder.partition('/')
|
|
entries[name] = (
|
|
ManagedFileDirectoryEntry(name, 'directory', None)
|
|
if separator else
|
|
ManagedFileDirectoryEntry(name, 'file', len(content))
|
|
)
|
|
values = tuple(entries[name] for name in sorted(entries))
|
|
return ManagedFileListing(values, sum(len(item.name.encode()) for item in values))
|
|
|
|
def download_file(self, root_id, relative_path):
|
|
self._fail()
|
|
self.calls.append(('download', root_id, relative_path))
|
|
if relative_path not in self.files:
|
|
raise ManagedFileAccessError('not_found')
|
|
content = self.files[relative_path]
|
|
if self.stream_downloads:
|
|
snapshot = FakeManagedFileSnapshot(content)
|
|
self.last_snapshot = snapshot
|
|
return ManagedFileDownload(
|
|
self._identity(content), snapshot=snapshot,
|
|
)
|
|
return ManagedFileDownload(self._identity(content), content)
|
|
|
|
def mutation_file_identity(
|
|
self, root_id, relative_path, operation,
|
|
*, require_private_sha256=None):
|
|
self._fail()
|
|
self.calls.append((
|
|
'identity', root_id, relative_path, operation.value,
|
|
require_private_sha256,
|
|
))
|
|
if relative_path not in self.files:
|
|
raise ManagedFileAccessError('not_found')
|
|
return self._identity(self.files[relative_path])
|
|
|
|
def create_replace_file(self, root_id, relative_path, payload, *, expected_sha256):
|
|
self._fail()
|
|
self.calls.append((
|
|
'create_replace', root_id, relative_path, expected_sha256,
|
|
hashlib.sha256(payload).hexdigest(), len(payload),
|
|
))
|
|
before_content = self.files.get(relative_path)
|
|
if expected_sha256 is None:
|
|
if before_content is not None:
|
|
raise ManagedFileAccessError('hash_conflict')
|
|
elif before_content is None:
|
|
raise ManagedFileAccessError('not_found')
|
|
elif self._identity(before_content).sha256 != expected_sha256:
|
|
raise ManagedFileAccessError('hash_conflict')
|
|
before = self._identity(before_content) if before_content is not None else None
|
|
after = self._identity(payload)
|
|
written = before != after
|
|
self.files[relative_path] = payload
|
|
return ManagedFileMutation(before, after, written)
|
|
|
|
def delete_file(self, root_id, relative_path, *, expected_sha256):
|
|
self._fail()
|
|
self.calls.append(('delete', root_id, relative_path, expected_sha256))
|
|
if relative_path not in self.files:
|
|
raise ManagedFileAccessError('not_found')
|
|
before = self._identity(self.files[relative_path])
|
|
if before.sha256 != expected_sha256:
|
|
raise ManagedFileAccessError('hash_conflict')
|
|
del self.files[relative_path]
|
|
return ManagedFileMutation(before, None, True)
|
|
|
|
def close(self):
|
|
self.closed += 1
|
|
|
|
|
|
class FakeManagedFileSnapshot:
|
|
def __init__(self, content):
|
|
self.content = content
|
|
self.closed = 0
|
|
|
|
def chunks(self):
|
|
try:
|
|
for offset in range(0, len(self.content), 3):
|
|
yield self.content[offset:offset + 3]
|
|
finally:
|
|
self.close()
|
|
|
|
def close(self):
|
|
if not self.closed:
|
|
self.closed = 1
|
|
|
|
|
|
class RecordingDB:
|
|
enabled = True
|
|
calls = []
|
|
fail_queue = False
|
|
source_operations = {}
|
|
worker_admin_operations = {}
|
|
document_operations = {}
|
|
apply_operations = {}
|
|
managed_file_operations = {}
|
|
audit_events = []
|
|
audit_next_before_event_id = None
|
|
source_identity_conflict = False
|
|
completion_failures = 0
|
|
control_identity_conflict = False
|
|
control_revision_conflict = False
|
|
document_completion_failures = 0
|
|
managed_file_completion_failures = 0
|
|
managed_file_execution_lock = threading.Lock()
|
|
|
|
def __init__(self, **kwargs):
|
|
self.managed_file_execution_held = False
|
|
self.calls.append(('open', kwargs))
|
|
|
|
def close(self):
|
|
if self.managed_file_execution_held:
|
|
self.managed_file_execution_held = False
|
|
self.managed_file_execution_lock.release()
|
|
self.calls.append(('close',))
|
|
|
|
def acquire_runtime_managed_file_execution(self, operation_id):
|
|
if self.managed_file_execution_held:
|
|
raise RuntimeError('managed file execution lock is already held')
|
|
self.managed_file_execution_lock.acquire()
|
|
self.managed_file_execution_held = True
|
|
self.calls.append(('managed_file_execution_acquire', operation_id))
|
|
return True
|
|
|
|
def release_runtime_managed_file_execution(self, operation_id):
|
|
if not self.managed_file_execution_held:
|
|
raise RuntimeError('managed file execution lock is not held')
|
|
self.managed_file_execution_held = False
|
|
self.managed_file_execution_lock.release()
|
|
self.calls.append(('managed_file_execution_release', operation_id))
|
|
return True
|
|
|
|
def admin_remote_worker_snapshot(self, limit, filters=None):
|
|
filters = dict(filters or {})
|
|
self.calls.append(('snapshot', limit, filters))
|
|
common = {
|
|
'queue_id': 5, 'user_key': 'fixture-user',
|
|
'active_assignment_cap': 2, 'device_key': 'fixture-device',
|
|
'source': 'gitlab', 'target': 'example/project@commit',
|
|
'issued_at': '2026-09-20T00:01:00+00:00',
|
|
'assignment_deadline_at': '2026-09-21T00:01:00+00:00',
|
|
'remote_result_upload_body_timeout_seconds': 1800,
|
|
'finished_at': None, 'duration_seconds': None,
|
|
'accepted': False, 'ingested': False, 'assignment_code': None,
|
|
'target_scan_id': None, 'scan_error_count': None,
|
|
'first_error_summary': None, 'skipped_reason': None,
|
|
'diagnostic_categories': None, 'diagnostic_codes': None,
|
|
'primary_diagnostic': None, 'phase_started_at': None,
|
|
'last_progress_at': None, 'last_progress_received_at': None,
|
|
'scan_deadline_at': None, 'scan_remaining_seconds': None,
|
|
'assignment_remaining_seconds': 3600,
|
|
'ingestion_state': None, 'projection_state': None,
|
|
'protocol_version': '2', 'bundle_format_version': '2',
|
|
'platform_tag': 'linux-x86_64', 'code_manifest_sha256': '1' * 64,
|
|
'detector_policy_sha256': '2' * 64,
|
|
'effective_config_sha256': '3' * 64,
|
|
}
|
|
assignments = [{
|
|
**common, 'reservation_id': 4,
|
|
'assignment_outcome': 'unfinished', 'scan_outcome': 'unavailable',
|
|
'diagnostic_count': 0, 'diagnostic_projection_version': None,
|
|
'active_phase': 'scanning', 'phase_age_seconds': 17,
|
|
'last_progress_age_seconds': 3, 'slot_id': 0,
|
|
'scan_deadline_at': '2026-09-20T00:11:00+00:00',
|
|
'scan_remaining_seconds': 500,
|
|
}, {
|
|
**common, 'reservation_id': 5, 'target_scan_id': 50,
|
|
'assignment_outcome': 'accepted', 'scan_outcome': 'degraded',
|
|
'finished_at': '2026-09-20T00:10:00+00:00',
|
|
'scan_warning_class': 'detector_timeout',
|
|
'scan_warning_summary': 'bounded detector output timed out',
|
|
'accepted': True, 'ingested': True, 'diagnostic_count': 2,
|
|
'diagnostic_projection_version': 1,
|
|
'diagnostic_categories': 'rate_limit, scanner',
|
|
'diagnostic_codes': 'provider.rate_limit, scanner.exit',
|
|
'primary_diagnostic': 'scanner/scanner.exit',
|
|
'active_phase': 'awaiting_receipt', 'phase_age_seconds': 4,
|
|
'last_progress_age_seconds': 2, 'slot_id': 1,
|
|
'ingestion_state': 'acknowledged', 'projection_state': 'completed',
|
|
}, {
|
|
**common, 'reservation_id': 6,
|
|
'assignment_outcome': 'prebundle_failed', 'scan_outcome': 'unavailable',
|
|
'finished_at': '2026-09-20T00:09:00+00:00',
|
|
'diagnostic_count': 1, 'diagnostic_projection_version': 1,
|
|
'diagnostic_categories': 'storage', 'diagnostic_codes': 'bundle.fsync',
|
|
'primary_diagnostic': 'storage/bundle.fsync', 'active_phase': 'bundling',
|
|
'phase_age_seconds': 9, 'last_progress_age_seconds': 8, 'slot_id': 0,
|
|
}, {
|
|
**common, 'reservation_id': 7,
|
|
'assignment_outcome': 'expired', 'scan_outcome': 'unavailable',
|
|
'finished_at': '2026-09-21T00:01:00+00:00',
|
|
'diagnostic_count': 1, 'diagnostic_projection_version': 1,
|
|
'diagnostic_categories': 'assignment_expired',
|
|
'diagnostic_codes': 'assignment.expired',
|
|
'primary_diagnostic': 'assignment_expired/assignment.expired',
|
|
'active_phase': 'scanning', 'phase_age_seconds': 300,
|
|
'last_progress_age_seconds': 280, 'slot_id': 0,
|
|
}, {
|
|
**common, 'reservation_id': 8, 'target_scan_id': 80,
|
|
'assignment_outcome': 'accepted', 'scan_outcome': 'error',
|
|
'finished_at': '2026-09-20T00:08:00+00:00',
|
|
'accepted': True, 'diagnostic_count': 0,
|
|
'diagnostic_projection_version': None, 'active_phase': None,
|
|
'phase_age_seconds': None, 'last_progress_age_seconds': None,
|
|
'slot_id': None,
|
|
'remote_result_upload_body_timeout_seconds': None,
|
|
'protocol_version': '1',
|
|
}, {
|
|
**common, 'reservation_id': 9,
|
|
'assignment_outcome': 'unfinished', 'scan_outcome': 'unavailable',
|
|
'diagnostic_count': 0, 'diagnostic_projection_version': 1,
|
|
'active_phase': None, 'phase_age_seconds': None,
|
|
'last_progress_age_seconds': None, 'slot_id': None,
|
|
}]
|
|
return {
|
|
'filters': filters,
|
|
'users': [{'user_key': 'fixture-user', 'active_assignment_cap': 2, 'disabled': False}],
|
|
'workers': [{
|
|
'device_key': 'fixture-device', 'user_key': 'fixture-user',
|
|
'active_assignment_cap': 2, 'active_slot_count': 1,
|
|
'current_phases': 'scanning',
|
|
'latest_progress_age_seconds': 3,
|
|
'known_reasons': None, 'pending_local_recovery': False,
|
|
'active_package_identity': 'linux-x86_64:' + ('1' * 12),
|
|
'unfinished_count': 1, 'completed_count': 2, 'failed_count': 0,
|
|
'expired_count': 0, 'last_contact_at': '2026-09-20T00:02:00+00:00',
|
|
'revoked': False,
|
|
}],
|
|
'assignments': assignments,
|
|
'deferred_queue': [],
|
|
}
|
|
|
|
def admin_worker_diagnostic_groups(self, limit, filters=None, *, occurrence_offset=0):
|
|
filters = dict(filters or {})
|
|
self.calls.append(('diagnostic_groups', limit, filters, occurrence_offset))
|
|
occurrences = [{
|
|
'diagnostic_uid': 'b' * 64, 'reservation_id': 5,
|
|
'target_scan_id': 50, 'source': 'gitlab',
|
|
'worker': 'fixture-device', 'user': 'fixture-user',
|
|
'assignment_outcome': 'accepted', 'scan_outcome': 'error',
|
|
'phase': 'scanning', 'kind': 'provider_http',
|
|
'category': 'rate_limit', 'code': 'provider.rate_limit',
|
|
'summary': 'rate limited', 'retryable': True,
|
|
'occurred_at': '2026-09-20T00:03:00Z',
|
|
'received_at': '2026-09-20T00:03:01Z',
|
|
}, {
|
|
'diagnostic_uid': 'c' * 64, 'reservation_id': 6,
|
|
'target_scan_id': None, 'source': 'gitlab',
|
|
'worker': 'fixture-device', 'user': 'fixture-user',
|
|
'assignment_outcome': 'prebundle_failed', 'scan_outcome': 'unavailable',
|
|
'phase': 'scanning', 'kind': 'provider_http',
|
|
'category': 'rate_limit', 'code': 'provider.rate_limit',
|
|
'summary': 'rate limited again', 'retryable': True,
|
|
'occurred_at': '2026-09-20T00:02:00Z',
|
|
'received_at': '2026-09-20T00:02:01Z',
|
|
}]
|
|
return {
|
|
'filters': filters, 'occurrence_limit': limit,
|
|
'matched_occurrence_count': 201, 'page_occurrence_count': 2,
|
|
'occurrence_offset': occurrence_offset,
|
|
'has_previous': occurrence_offset > 0,
|
|
'has_next': occurrence_offset == 0,
|
|
'previous_occurrence_offset': (
|
|
max(0, occurrence_offset - limit) if occurrence_offset else None
|
|
),
|
|
'next_occurrence_offset': limit if occurrence_offset == 0 else None,
|
|
'truncated': occurrence_offset == 0,
|
|
'groups': [{
|
|
'fingerprint': 'f' * 64, 'count': 2,
|
|
'affected_assignment_count': 2, 'page_occurrence_count': 2,
|
|
'affected_assignments': [5, 6], 'occurrences': occurrences,
|
|
}],
|
|
}
|
|
|
|
def admin_worker_duration_metrics(self, limit, filters=None, *, offset=0):
|
|
filters = dict(filters or {})
|
|
self.calls.append(('duration_metrics', limit, filters, offset))
|
|
metrics = [{
|
|
'source': 'gitlab', 'phase': 'scanning', 'outcome': 'error',
|
|
'sample_count': 12, 'sufficient': True, 'minimum_sample_count': 5,
|
|
'p50_seconds': 10.0, 'p95_seconds': 18.0, 'p99_seconds': 21.0,
|
|
}, {
|
|
'source': 'gitlab', 'phase': 'uploading', 'outcome': 'accepted',
|
|
'sample_count': 2, 'sufficient': False, 'minimum_sample_count': 5,
|
|
'p50_seconds': 3.0, 'p95_seconds': 4.0, 'p99_seconds': 4.0,
|
|
}]
|
|
return {
|
|
'filters': filters, 'metrics': metrics,
|
|
'total_group_count': 201, 'metric_limit': limit,
|
|
'metric_offset': offset, 'page_group_count': len(metrics),
|
|
'has_previous': offset > 0, 'has_next': offset == 0,
|
|
'previous_metric_offset': max(0, offset - limit) if offset else None,
|
|
'next_metric_offset': limit if offset == 0 else None,
|
|
'truncated': offset == 0,
|
|
}
|
|
|
|
@staticmethod
|
|
def _diagnostic_fixture(reservation_id=5, diagnostic_uid=None):
|
|
def material_value(material):
|
|
return {
|
|
'encoding': material.encoding.value,
|
|
'head': material.head, 'tail': material.tail,
|
|
'original_size': material.original_size,
|
|
'stored_size': material.stored_size,
|
|
'sha256': material.sha256, 'truncated': material.truncated,
|
|
}
|
|
|
|
material = material_value(make_body_material(
|
|
b'b' * MAX_DIAGNOSTIC_BODY_BYTES,
|
|
))
|
|
truncated = material_value(make_log_material(
|
|
b'l' * (MAX_DIAGNOSTIC_LOG_BYTES + 1),
|
|
))
|
|
diagnostic_uid = diagnostic_uid or ('b' * 64)
|
|
return {
|
|
'schema': 1, 'diagnostic_uid': diagnostic_uid,
|
|
'occurrence_id': 'fixture-occurrence', 'reservation_id': reservation_id,
|
|
'scan_event_id': None, 'slot_id': 0, 'source': 'gitlab',
|
|
'phase': 'scanning', 'kind': 'provider_http',
|
|
'category': 'rate_limit', 'code': 'provider.rate_limit',
|
|
'summary': 'provider returned a limit', 'retryable': True,
|
|
'attempt': 1, 'assignment_outcome': None, 'scan_outcome': None,
|
|
'occurred_at': '2026-09-20T00:03:00Z',
|
|
'captured_at': '2026-09-20T00:03:01Z', 'received_at': None,
|
|
'http': {
|
|
'operation': 'GET', 'status_code': 429,
|
|
'content_type': 'text/plain', 'request_id': 'fixture',
|
|
'body': material, 'headers': None,
|
|
},
|
|
'process': {
|
|
'name': 'trufflehog', 'exit_code': 1, 'signal': None,
|
|
'timed_out': False, 'stdout': None, 'stderr': truncated,
|
|
},
|
|
'exception': None,
|
|
}
|
|
|
|
def admin_worker_assignment_detail(self, reservation_id, **_kwargs):
|
|
self.calls.append(('assignment_detail', reservation_id, _kwargs))
|
|
reservation_id = int(reservation_id)
|
|
states = {
|
|
4: ('unfinished', 'unavailable', None, 'scanning'),
|
|
5: ('accepted', 'error', '2026-09-20T00:06:00Z', 'awaiting_receipt'),
|
|
6: ('prebundle_failed', 'unavailable', '2026-09-20T00:04:00Z', 'bundling'),
|
|
7: ('expired', 'unavailable', '2026-09-20T00:11:00Z', 'scanning'),
|
|
8: ('accepted', 'error', '2026-09-20T00:06:00Z', 'awaiting_receipt'),
|
|
}
|
|
if reservation_id not in states:
|
|
return None
|
|
assignment_outcome, scan_outcome, resolved_at, current_phase = states[reservation_id]
|
|
legacy = reservation_id == 8
|
|
diagnostic_uid = ({5: 'b', 6: 'c', 7: 'd'}.get(reservation_id, 'b')) * 64
|
|
envelope = self._diagnostic_fixture(reservation_id, diagnostic_uid)
|
|
canonical = json.dumps(envelope, ensure_ascii=True, sort_keys=True, separators=(',', ':'))
|
|
diagnostic_items = []
|
|
if reservation_id in {5, 6, 7}:
|
|
diagnostic_items.append({
|
|
'diagnostic_uid': diagnostic_uid, 'phase': current_phase,
|
|
'kind': 'provider_http', 'category': 'rate_limit',
|
|
'code': 'provider.rate_limit', 'retryable': True,
|
|
'summary': 'provider returned a limit',
|
|
'occurred_at': '2026-09-20T00:03:00Z',
|
|
'received_at': '2026-09-20T00:03:01Z',
|
|
'envelope_sha256': '6' * 64, 'diagnostic': envelope,
|
|
'canonical_envelope_json': canonical,
|
|
})
|
|
timeline = [{
|
|
'timestamp': '2026-09-20T00:01:00Z',
|
|
'kind': 'assignment', 'label': 'issued',
|
|
}]
|
|
if not legacy:
|
|
timeline.append({
|
|
'timestamp': '2026-09-20T00:02:00Z', 'kind': 'phase',
|
|
'label': current_phase, 'sequence': 1,
|
|
'received_at': '2026-09-20T00:02:01Z',
|
|
})
|
|
if reservation_id == 5:
|
|
timeline.extend([
|
|
{'timestamp': '2026-09-20T00:04:00Z', 'kind': 'transport', 'label': 'bundle received'},
|
|
{'timestamp': '2026-09-20T00:05:00Z', 'kind': 'ingestion', 'label': 'bundle ingested'},
|
|
{'timestamp': '2026-09-20T00:05:45Z', 'kind': 'settlement', 'label': 'queue settled'},
|
|
{'timestamp': '2026-09-20T00:06:00Z', 'kind': 'projection', 'label': 'projection completed'},
|
|
])
|
|
elif reservation_id == 6:
|
|
timeline.append({
|
|
'timestamp': resolved_at, 'kind': 'receipt', 'label': 'prebundle_report',
|
|
})
|
|
elif reservation_id == 7:
|
|
timeline.append({
|
|
'timestamp': resolved_at, 'kind': 'receipt', 'label': 'expired',
|
|
})
|
|
return {
|
|
'schema': 1,
|
|
'assignment': {
|
|
'reservation_id': int(reservation_id), 'queue_id': 5,
|
|
'source': 'gitlab', 'target': 'example/project@commit',
|
|
'user': 'fixture-user', 'worker': 'fixture-device',
|
|
'active_assignment_cap': 2,
|
|
'assignment_outcome': assignment_outcome,
|
|
'scan_outcome': scan_outcome,
|
|
'issued_at': '2026-09-20T00:01:00Z',
|
|
'resolved_at': resolved_at,
|
|
'assignment_code': None, 'assignment_detail': None,
|
|
},
|
|
'deadlines': {
|
|
'assignment_deadline_at': '2026-09-21T00:01:00Z',
|
|
'scan_deadline_at': '2026-09-20T00:11:00Z',
|
|
'upload_timeout_seconds': None if legacy else 1800,
|
|
'upload_timeout_availability': (
|
|
'legacy/unavailable' if legacy
|
|
else 'persisted at assignment issuance'
|
|
),
|
|
},
|
|
'package': {
|
|
'protocol_version': 1 if legacy else 2,
|
|
'bundle_format_version': 2,
|
|
'platform_tag': 'linux-x86_64',
|
|
'code_manifest_sha256': '1' * 64,
|
|
'detector_policy_sha256': '2' * 64,
|
|
'effective_config_sha256': '3' * 64,
|
|
},
|
|
'transport': {
|
|
'resolution': None, 'receipt_id': (
|
|
'receipt-fixture' if resolved_at else None
|
|
),
|
|
'bundle_state': 'acknowledged' if reservation_id in {5, 8} else None,
|
|
'bundle_ready_at': '2026-09-20T00:04:00Z' if reservation_id in {5, 8} else None,
|
|
'bundle_committed_at': '2026-09-20T00:05:00Z' if reservation_id in {5, 8} else None,
|
|
'bundle_acknowledged_at': '2026-09-20T00:05:30Z' if reservation_id in {5, 8} else None,
|
|
'queue_status': 'done' if resolved_at else 'in_progress',
|
|
'queue_settled_at': '2026-09-20T00:05:45Z' if reservation_id in {5, 8} else None,
|
|
'projection_status': 'completed' if reservation_id in {5, 8} else None,
|
|
'projection_completed_at': '2026-09-20T00:06:00Z' if reservation_id in {5, 8} else None,
|
|
},
|
|
'scan': {
|
|
'available': reservation_id in {5, 8},
|
|
'target_scan_id': reservation_id * 10 if reservation_id in {5, 8} else None,
|
|
'status': 'error' if reservation_id in {5, 8} else None,
|
|
'started_at': '2026-09-20T00:02:00Z',
|
|
'ended_at': '2026-09-20T00:04:00Z',
|
|
'duration_seconds': 120.0, 'findings_count': 0,
|
|
'verified_findings_count': 0, 'error_count': 1,
|
|
'skipped_reason': None,
|
|
'first_error_summary': 'legacy summary' if legacy else None,
|
|
'warning_class': 'detector_timeout' if reservation_id == 5 else None,
|
|
'warning_summary': (
|
|
'bounded detector output timed out' if reservation_id == 5 else None
|
|
),
|
|
},
|
|
'timeline': timeline,
|
|
'durations': [{
|
|
'phase': current_phase, 'duration_seconds': 120.0,
|
|
'outcome': scan_outcome,
|
|
'complete': resolved_at is not None,
|
|
'ended_at': resolved_at or '2026-09-20T00:04:00Z',
|
|
'authority': (
|
|
'assignment_resolution' if resolved_at else 'database_current_time'
|
|
),
|
|
}],
|
|
'progress': {
|
|
'available': not legacy,
|
|
'availability': (
|
|
'legacy/unavailable' if legacy else 'current'
|
|
),
|
|
'truncated': False,
|
|
'total_event_count': 0 if legacy else 1,
|
|
'omitted_older_event_count': 0,
|
|
'current_phase': None if legacy else current_phase,
|
|
'phase_started_at': None if legacy else '2026-09-20T00:02:00Z',
|
|
'phase_age_seconds': None if legacy else 120.0,
|
|
'last_progress_at': None if legacy else '2026-09-20T00:03:55Z',
|
|
'last_progress_age_seconds': None if legacy else 5.0,
|
|
'age_authority': (
|
|
None if legacy else
|
|
'assignment_resolution' if resolved_at
|
|
else 'database_current_time'
|
|
),
|
|
'events': [],
|
|
},
|
|
'diagnostics': {
|
|
'availability': 'legacy/unavailable' if legacy else 'current',
|
|
'projection_version': None if legacy or reservation_id == 4 else 1,
|
|
'declared_count': None if legacy else len(diagnostic_items),
|
|
'truncated': False,
|
|
'items': diagnostic_items,
|
|
},
|
|
'legacy_evidence': {
|
|
'available': legacy, 'explicitly_not_an_envelope': True,
|
|
'truncated': False,
|
|
'errors': [{
|
|
'id': 1, 'category': 'scanner', 'summary': 'legacy summary',
|
|
'raw_error': 'exact legacy raw error',
|
|
'created_at': '2026-09-20T00:04:00Z',
|
|
}] if legacy else [],
|
|
'first_error_summary': 'legacy summary' if legacy else None,
|
|
},
|
|
}
|
|
|
|
def admin_worker_diagnostic_envelope(self, reservation_id, diagnostic_uid):
|
|
self.calls.append(('diagnostic_envelope', reservation_id, diagnostic_uid))
|
|
if int(reservation_id) != 5 or diagnostic_uid != 'b' * 64:
|
|
return None
|
|
envelope = self._diagnostic_fixture(5, 'b' * 64)
|
|
canonical = json.dumps(envelope, ensure_ascii=True, sort_keys=True, separators=(',', ':'))
|
|
return {'canonical_json': canonical, 'sha256': '6' * 64}
|
|
|
|
def admin_target_queue_health(self, sources):
|
|
self.calls.append(('queue_counts', tuple(sources)))
|
|
if self.fail_queue:
|
|
if self.fail_queue == 'malformed':
|
|
return {
|
|
'counts': {}, 'degraded': False, 'stale': False,
|
|
'truncated_statuses': 1,
|
|
}
|
|
if self.fail_queue == 'degraded':
|
|
return {
|
|
'counts': {}, 'degraded': True, 'stale': True,
|
|
'reason': 'bounded_count_query_failed',
|
|
'retry_after_sec': 300, 'truncated_statuses': [],
|
|
}
|
|
raise RuntimeError('sensitive queue failure detail')
|
|
return {
|
|
'counts': {'pending': 3, 'done': 100},
|
|
'degraded': True, 'stale': False,
|
|
'reason': 'bounded_status_sample',
|
|
'sample_limit_per_status': 100, 'timeout_ms': 5000,
|
|
'sampled_rows': 103, 'retry_after_sec': 0,
|
|
'truncated_statuses': ['done'],
|
|
}
|
|
|
|
def runtime_drain_progress(self):
|
|
self.calls.append(('drain_progress',))
|
|
return {
|
|
'revision': 7,
|
|
'discovery_paused': False,
|
|
'dispatch_paused': True,
|
|
'drain_state': 'draining',
|
|
'effective_discovery_paused': True,
|
|
'effective_dispatch_paused': True,
|
|
'actor': 'fixture.operator',
|
|
'operation_id': '00000000-0000-4000-8000-000000000001',
|
|
'created_at': '2026-09-20T00:00:00+00:00',
|
|
'updated_at': '2026-09-20T00:01:00+00:00',
|
|
'live_remote_assignments': 2,
|
|
'precommit_result_bundles': 1,
|
|
'blocker_count': 3,
|
|
}
|
|
|
|
def set_runtime_dispatch_paused(
|
|
self, paused, *, expected_revision, actor, operation_id,
|
|
):
|
|
if self.control_revision_conflict:
|
|
raise RuntimeControlRevisionConflictError(
|
|
expected_revision, {'revision': expected_revision + 1},
|
|
)
|
|
if self.control_identity_conflict:
|
|
raise RuntimeOperationIdentityConflictError('fixture conflict')
|
|
self.calls.append((
|
|
'dispatch_paused', paused, expected_revision, actor, operation_id,
|
|
))
|
|
return {'after': {'revision': expected_revision + 1}}
|
|
|
|
def start_runtime_drain(self, *, expected_revision, actor, operation_id):
|
|
if self.control_revision_conflict:
|
|
raise RuntimeControlRevisionConflictError(
|
|
expected_revision, {'revision': expected_revision + 1},
|
|
)
|
|
if self.control_identity_conflict:
|
|
raise RuntimeOperationIdentityConflictError('fixture conflict')
|
|
self.calls.append(('drain_start', expected_revision, actor, operation_id))
|
|
return {'after': {'revision': expected_revision + 1}}
|
|
|
|
def cancel_runtime_drain(self, *, expected_revision, actor, operation_id):
|
|
if self.control_revision_conflict:
|
|
raise RuntimeControlRevisionConflictError(
|
|
expected_revision, {'revision': expected_revision + 1},
|
|
)
|
|
if self.control_identity_conflict:
|
|
raise RuntimeOperationIdentityConflictError('fixture conflict')
|
|
self.calls.append(('drain_cancel', expected_revision, actor, operation_id))
|
|
return {'after': {'revision': expected_revision + 1}}
|
|
|
|
def recent_runtime_operations(
|
|
self, limit, *, before_updated_at=None, before_operation_id=None,
|
|
):
|
|
self.calls.append((
|
|
'recent_operations', limit, before_updated_at, before_operation_id,
|
|
))
|
|
recorded = (
|
|
list(type(self).apply_operations.values())
|
|
+ list(type(self).document_operations.values())
|
|
+ list(type(self).managed_file_operations.values())
|
|
)
|
|
if recorded:
|
|
return [dict(operation) for operation in recorded[:limit]]
|
|
return [{
|
|
'operation_id': '00000000-0000-4000-8000-000000000002',
|
|
'actor': 'fixture.operator', 'action': 'restart',
|
|
'target_ref': 'runtime', 'status': 'succeeded',
|
|
'safe_category': None, 'safe_detail': None,
|
|
'requested_at': '2026-09-20T00:00:00+00:00',
|
|
'completed_at': '2026-09-20T00:00:05+00:00',
|
|
'updated_at': '2026-09-20T00:00:05+00:00',
|
|
}]
|
|
|
|
def set_runtime_discovery_paused(
|
|
self, paused, *, expected_revision, actor, operation_id,
|
|
):
|
|
if self.control_revision_conflict:
|
|
raise RuntimeControlRevisionConflictError(
|
|
expected_revision, {'revision': expected_revision + 1},
|
|
)
|
|
self.calls.append((
|
|
'discovery_paused', paused, expected_revision, actor, operation_id,
|
|
))
|
|
return {'after': {'revision': expected_revision + 1}}
|
|
|
|
def create_runtime_source_operation(
|
|
self, *, operation_id, actor, source_id, source_action,
|
|
interval_seconds=None, mode=None, restart_enabled=None,
|
|
restart_delay_seconds=None,
|
|
):
|
|
if self.source_identity_conflict:
|
|
raise RuntimeOperationIdentityConflictError('fixture conflict')
|
|
self.calls.append((
|
|
'source_operation_create', operation_id, actor, source_id,
|
|
source_action, interval_seconds, mode, restart_enabled,
|
|
restart_delay_seconds,
|
|
))
|
|
existing = self.source_operations.get(operation_id)
|
|
if existing is not None:
|
|
return {**existing, 'replayed': True}
|
|
operation = {
|
|
'operation_id': operation_id, 'status': 'running', 'replayed': False,
|
|
'resulting_identity': None,
|
|
}
|
|
self.source_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def complete_runtime_source_operation(
|
|
self, operation_id, *, succeeded, outcome=None,
|
|
):
|
|
self.calls.append((
|
|
'source_operation_complete', operation_id, succeeded, outcome,
|
|
))
|
|
if type(self).completion_failures:
|
|
type(self).completion_failures -= 1
|
|
raise RuntimeError('fixture completion failure')
|
|
operation = {
|
|
**self.source_operations[operation_id],
|
|
'status': 'succeeded' if succeeded else 'failed',
|
|
'resulting_identity': (
|
|
{'outcome': outcome} if succeeded else None
|
|
),
|
|
}
|
|
self.source_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def create_runtime_worker_admin_operation(
|
|
self, *, operation_id, actor, action, target_ref, request_sha256,
|
|
):
|
|
self.calls.append((
|
|
'worker_admin_create', operation_id, actor, action, target_ref,
|
|
request_sha256,
|
|
))
|
|
existing = self.worker_admin_operations.get(operation_id)
|
|
if existing is not None:
|
|
return {**existing, 'replayed': True}
|
|
operation = {
|
|
'operation_id': operation_id, 'status': 'running', 'replayed': False,
|
|
'resulting_identity': None,
|
|
}
|
|
self.worker_admin_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def complete_runtime_worker_admin_operation(
|
|
self, operation_id, *, succeeded, affected_count=None,
|
|
):
|
|
self.calls.append((
|
|
'worker_admin_complete', operation_id, succeeded, affected_count,
|
|
))
|
|
operation = {
|
|
**self.worker_admin_operations[operation_id],
|
|
'status': 'succeeded' if succeeded else 'failed',
|
|
'resulting_identity': (
|
|
{'outcome': 'completed', 'affected_count': affected_count}
|
|
if succeeded else None
|
|
),
|
|
}
|
|
self.worker_admin_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def create_runtime_document_operation(self, **kwargs):
|
|
self.calls.append(('document_operation_create', kwargs))
|
|
operation_id = kwargs['operation_id']
|
|
document = kwargs['action'].split('.')[1]
|
|
expected_identity = {
|
|
'document': document,
|
|
'active_config_sha256': kwargs['active_config_sha256'],
|
|
'active_secrets_sha256': kwargs['active_secrets_sha256'],
|
|
'candidate_config_sha256': kwargs['candidate_config_sha256'],
|
|
'candidate_secrets_sha256': kwargs['candidate_secrets_sha256'],
|
|
'candidate_after_sha256': kwargs['candidate_after_sha256'],
|
|
'candidate_before_bytes': kwargs['candidate_before_bytes'],
|
|
'candidate_after_bytes': kwargs['candidate_after_bytes'],
|
|
'candidate_before_present': kwargs['candidate_before_present'],
|
|
}
|
|
existing = self.document_operations.get(operation_id)
|
|
if existing is not None:
|
|
if not (
|
|
existing['actor'] == kwargs['actor']
|
|
and existing['action'] == kwargs['action']
|
|
and existing['target_ref'] == document
|
|
and existing['expected_identity'] == expected_identity
|
|
):
|
|
raise RuntimeOperationIdentityConflictError('fixture identity conflict')
|
|
return {**existing, 'replayed': True}
|
|
if operation_id in self.apply_operations:
|
|
raise RuntimeOperationIdentityConflictError('fixture identity conflict')
|
|
operation = {
|
|
'operation_id': operation_id, 'status': 'running', 'replayed': False,
|
|
'actor': kwargs['actor'], 'action': kwargs['action'],
|
|
'target_kind': 'runtime-document', 'target_ref': document,
|
|
'expected_identity': expected_identity,
|
|
}
|
|
self.document_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def complete_runtime_document_operation(
|
|
self, operation_id, *, succeeded, candidate_sha256=None, written=None,
|
|
):
|
|
self.calls.append((
|
|
'document_operation_complete', operation_id, succeeded,
|
|
candidate_sha256, written,
|
|
))
|
|
if type(self).document_completion_failures:
|
|
type(self).document_completion_failures -= 1
|
|
raise RuntimeError('fixture document completion failure')
|
|
operation = {
|
|
**self.document_operations[operation_id],
|
|
'status': 'succeeded' if succeeded else 'failed',
|
|
}
|
|
self.document_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def create_runtime_operation(self, **kwargs):
|
|
self.calls.append(('runtime_operation_create', kwargs))
|
|
operation_id = kwargs['operation_id']
|
|
existing = self.apply_operations.get(operation_id)
|
|
if existing is not None:
|
|
if not (
|
|
existing['actor'] == kwargs['actor']
|
|
and existing['action'] == kwargs['action']
|
|
and existing['expected_identity'] == kwargs['expected_identity']
|
|
):
|
|
raise RuntimeOperationIdentityConflictError('fixture identity conflict')
|
|
return {**existing, 'replayed': True}
|
|
if operation_id in self.document_operations:
|
|
raise RuntimeOperationIdentityConflictError('fixture identity conflict')
|
|
operation = {
|
|
'operation_id': operation_id, 'status': 'requested', 'replayed': False,
|
|
'actor': kwargs['actor'], 'action': kwargs['action'],
|
|
'expected_identity': dict(kwargs['expected_identity']),
|
|
}
|
|
self.apply_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def create_runtime_managed_file_operation(self, **kwargs):
|
|
self.calls.append(('managed_file_operation_create', dict(kwargs)))
|
|
operation_id = kwargs['operation_id']
|
|
expected_identity = {
|
|
key: kwargs[key] for key in (
|
|
'root_id', 'relative_path', 'expected_sha256',
|
|
'proposed_sha256', 'proposed_byte_count',
|
|
)
|
|
}
|
|
existing = self.managed_file_operations.get(operation_id)
|
|
if existing is not None:
|
|
if not (
|
|
existing['actor'] == kwargs['actor']
|
|
and existing['action'] == kwargs['action']
|
|
and existing['target_ref'] == kwargs['root_id']
|
|
and existing['expected_identity'] == expected_identity
|
|
):
|
|
raise RuntimeOperationIdentityConflictError('fixture identity conflict')
|
|
return {**existing, 'replayed': True}
|
|
operation = {
|
|
'operation_id': operation_id, 'status': 'running',
|
|
'replayed': False, 'actor': kwargs['actor'],
|
|
'action': kwargs['action'], 'target_kind': 'managed-file',
|
|
'target_ref': kwargs['root_id'],
|
|
'expected_identity': expected_identity,
|
|
'resulting_identity': None,
|
|
}
|
|
self.managed_file_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def complete_runtime_managed_file_operation(self, operation_id, **kwargs):
|
|
self.calls.append((
|
|
'managed_file_operation_complete', operation_id, dict(kwargs),
|
|
))
|
|
if type(self).managed_file_completion_failures:
|
|
type(self).managed_file_completion_failures -= 1
|
|
raise RuntimeError('fixture managed file completion failure')
|
|
operation = self.managed_file_operations[operation_id]
|
|
resulting = None
|
|
if kwargs['succeeded']:
|
|
expected = operation['expected_identity']
|
|
resulting = {
|
|
'root_id': expected['root_id'],
|
|
'relative_path': expected['relative_path'],
|
|
'outcome': 'completed',
|
|
'before_sha256': kwargs.get('before_sha256'),
|
|
'before_byte_count': kwargs.get('before_byte_count'),
|
|
'after_sha256': kwargs.get('after_sha256'),
|
|
'after_byte_count': kwargs.get('after_byte_count'),
|
|
'written': kwargs.get('written'),
|
|
}
|
|
operation = {
|
|
**operation,
|
|
'status': 'succeeded' if kwargs['succeeded'] else 'failed',
|
|
'resulting_identity': resulting,
|
|
}
|
|
self.managed_file_operations[operation_id] = operation
|
|
return dict(operation)
|
|
|
|
def runtime_operation(self, operation_id):
|
|
self.calls.append(('runtime_operation', operation_id))
|
|
operation = (
|
|
self.document_operations.get(operation_id)
|
|
or self.apply_operations.get(operation_id)
|
|
or self.managed_file_operations.get(operation_id)
|
|
)
|
|
return None if operation is None else dict(operation)
|
|
|
|
def runtime_audit_events(self, *, before_event_id=None, limit=50):
|
|
self.calls.append(('runtime_audit_events', before_event_id, limit))
|
|
return {
|
|
'events': [dict(event) for event in type(self).audit_events],
|
|
'next_before_event_id': type(self).audit_next_before_event_id,
|
|
}
|
|
|
|
def create_remote_worker_user(self, user_key, cap):
|
|
self.calls.append(('create_user', user_key, cap))
|
|
return {'user_key': user_key}
|
|
|
|
def set_remote_worker_user_cap(self, user_key, cap):
|
|
self.calls.append(('set_cap', user_key, cap))
|
|
return True
|
|
|
|
def set_remote_worker_user_disabled(self, user_key, disabled):
|
|
self.calls.append(('set_disabled', user_key, disabled))
|
|
return True
|
|
|
|
def issue_remote_worker_device(self, user_key, device_key, token_sha256, *, rotate=False):
|
|
self.calls.append(('issue_device', user_key, device_key, token_sha256, rotate))
|
|
return {'user_key': user_key, 'device_key': device_key}
|
|
|
|
def set_remote_worker_device_revoked(self, device_key, revoked=True):
|
|
self.calls.append(('set_revoked', device_key, revoked))
|
|
return True
|
|
|
|
def admin_requeue_deferred_targets(self, queue_ids, *, max_items):
|
|
self.calls.append(('requeue', list(queue_ids), max_items))
|
|
return len(queue_ids)
|
|
|
|
def admin_discard_queued_source(self, source):
|
|
self.calls.append(('discard_source_queue', source))
|
|
return 17
|
|
|
|
|
|
class AdminAPITests(unittest.TestCase):
|
|
def setUp(self):
|
|
RecordingDB.calls = []
|
|
RecordingDB.fail_queue = False
|
|
RecordingDB.source_operations = {}
|
|
RecordingDB.worker_admin_operations = {}
|
|
RecordingDB.document_operations = {}
|
|
RecordingDB.apply_operations = {}
|
|
RecordingDB.managed_file_operations = {}
|
|
RecordingDB.audit_events = []
|
|
RecordingDB.audit_next_before_event_id = None
|
|
RecordingDB.source_identity_conflict = False
|
|
RecordingDB.completion_failures = 0
|
|
RecordingDB.control_identity_conflict = False
|
|
RecordingDB.control_revision_conflict = False
|
|
RecordingDB.document_completion_failures = 0
|
|
RecordingDB.managed_file_completion_failures = 0
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
root = ensure_private_directory(
|
|
os.path.join(self.temp_dir.name, 'bundles'), reject_reparse=True,
|
|
)
|
|
self.worker = FakeWorkerService(root)
|
|
self.runtime_provider = mock.Mock(return_value={
|
|
'snapshot_schema': 2,
|
|
'runtime': {
|
|
'pid': 123, 'phase': 'ACTIVE', 'manages_postgres': True,
|
|
'start_gate_open': True, 'shutdown_requested': False,
|
|
'runtime_failed': False,
|
|
},
|
|
'postgres': {
|
|
'state': 'READY', 'ready': True, 'failures': 0,
|
|
'safe_error_category': '',
|
|
},
|
|
'dashboard': {
|
|
'status': 'running', 'desired_state': 'running', 'healthy': True,
|
|
'pid': 456, 'safe_error_category': '',
|
|
},
|
|
'sources': [
|
|
{
|
|
'id': f'discovery-producer:{source}', 'source': source,
|
|
'role': 'discovery-producer', 'lifecycle_state': 'waiting',
|
|
'process_state': 'stopped', 'desired_state': 'running',
|
|
'safe_error_category': '',
|
|
'interval_seconds': 3600, 'restart_enabled': True,
|
|
'restart_count': 2,
|
|
'restart_delay_seconds': 5, 'restart_streak': 0,
|
|
'mode': 'repeat', 'pid': None,
|
|
'allowed_actions': [
|
|
'start', 'stop', 'restart', 'pause', 'resume', 'set-interval',
|
|
],
|
|
'last_cycle_result': {
|
|
'status': 'completed', 'fetched_count': 4,
|
|
'queued_new_count': 2, 'queued_updated_count': 1,
|
|
},
|
|
'last_successful_discovery_at': '2026-09-20T00:00:00+00:00',
|
|
'next_scheduled_run_at': '2026-09-20T01:00:00+00:00',
|
|
}
|
|
for source in ('gitlab', 'dockerhub', 'huggingface')
|
|
] + [
|
|
{
|
|
'id': source, 'source': source, 'role': source,
|
|
'lifecycle_state': 'running', 'process_state': 'running',
|
|
'desired_state': 'running', 'safe_error_category': '',
|
|
'mode': 'singleton', 'pid': 789, 'interval_seconds': 0,
|
|
'restart_enabled': True, 'restart_delay_seconds': 5,
|
|
'restart_count': 0, 'restart_streak': 0,
|
|
'allowed_actions': [
|
|
'start', 'stop', 'restart', 'pause', 'resume',
|
|
'set-restart', 'set-restart-delay',
|
|
],
|
|
}
|
|
for source in ('result-ingester', 'jsonl-projector', 'janitor', 'worker-api')
|
|
] + [{
|
|
'id': 'keychecks', 'source': 'keychecks', 'role': 'keycheck',
|
|
'lifecycle_state': 'waiting', 'process_state': 'stopped',
|
|
'desired_state': 'running', 'safe_error_category': '', 'pid': None,
|
|
'mode': 'repeat', 'interval_seconds': 3600, 'restart_enabled': True,
|
|
'restart_delay_seconds': 5, 'restart_count': 0, 'restart_streak': 0,
|
|
'allowed_actions': [
|
|
'start', 'stop', 'restart', 'pause', 'resume', 'once',
|
|
'set-mode', 'set-interval', 'set-restart', 'set-restart-delay',
|
|
],
|
|
}, {
|
|
'id': 'docker-shadow', 'source': 'docker-shadow',
|
|
'role': 'docker-shadow', 'lifecycle_state': 'stopped',
|
|
'process_state': 'stopped', 'desired_state': 'stopped',
|
|
'safe_error_category': '', 'pid': None, 'mode': 'manual',
|
|
'interval_seconds': 0, 'restart_enabled': False,
|
|
'restart_delay_seconds': 0, 'restart_count': 0,
|
|
'restart_streak': 0, 'allowed_actions': ['start', 'stop'],
|
|
}],
|
|
'pipeline': {
|
|
'ingester_ready': True, 'projector_ready': True,
|
|
'cutover_ready': True,
|
|
'ingester_state': 'ready', 'projector_state': 'ready',
|
|
'bundle_items': 1, 'bundle_bytes': 4096,
|
|
'projection_items': 0, 'projection_bytes': 0,
|
|
'keycheck_items': 0, 'keycheck_bytes': 0,
|
|
'quarantine_items': 0, 'quarantine_bytes': 0,
|
|
},
|
|
'scan_workers': {
|
|
'active': 0, 'limit': 2, 'base_active': 0, 'base_limit': 2,
|
|
'bonus_active': 0, 'bonus_limit': 0, 'trufflehog': 0,
|
|
'sources': {},
|
|
},
|
|
})
|
|
self.source_action_provider = mock.Mock(return_value={
|
|
'source_action': 'restart', 'outcome': 'completed', 'source': {},
|
|
})
|
|
self.dashboard_action_provider = mock.Mock(return_value={
|
|
'dashboard_action': 'restart', 'outcome': 'completed', 'dashboard': {},
|
|
})
|
|
self.source_log_provider = mock.Mock(return_value={
|
|
'source_id': 'result-ingester', 'line_count': 2,
|
|
'lines': ['line <one>', 'line two'], 'response_truncated': False,
|
|
})
|
|
self.package_compatibility_provider = mock.Mock(return_value={
|
|
'profiles': [{
|
|
'profile_name': 'linux-x86_64', 'protocol_version': 2,
|
|
'bundle_format_version': 2, 'platform_tag': 'linux-x86_64',
|
|
'code_manifest_sha256': '1' * 64,
|
|
'detector_policy_sha256': '2' * 64,
|
|
'sources': ['dockerhub', 'gitlab', 'huggingface'],
|
|
'capabilities': [
|
|
{'source': 'gitlab', 'platform': 'gitlab', 'planning_kind': 'exact_git_v1'},
|
|
{'source': 'dockerhub', 'platform': 'docker', 'planning_kind': 'docker_direct_v1'},
|
|
{'source': 'huggingface', 'platform': 'huggingface', 'planning_kind': 'huggingface_space_v1'},
|
|
],
|
|
}],
|
|
'required_capabilities': [
|
|
{'source': 'gitlab', 'platform': 'gitlab', 'planning_kind': 'exact_git_v1'},
|
|
{'source': 'dockerhub', 'platform': 'docker', 'planning_kind': 'docker_direct_v1'},
|
|
{'source': 'huggingface', 'platform': 'huggingface', 'planning_kind': 'huggingface_space_v1'},
|
|
],
|
|
})
|
|
self.document_state = SimpleNamespace(
|
|
active_config=SimpleNamespace(sha256='a' * 64, byte_count=10, present=True),
|
|
active_secrets=SimpleNamespace(sha256='b' * 64, byte_count=10, present=True),
|
|
candidate_config=SimpleNamespace(sha256='c' * 64, byte_count=10, present=True),
|
|
candidate_secrets=SimpleNamespace(sha256='d' * 64, byte_count=10, present=True),
|
|
)
|
|
self.runtime_config = {
|
|
'global': {'project_dir': '/opt/truf/app'},
|
|
'sources': {
|
|
'gitlab': {'timeout': 600},
|
|
'dockerhub': {'timeout': 900},
|
|
'huggingface': {'timeout': 300},
|
|
},
|
|
'supervisor': {'worker_api': {
|
|
'sources': ['gitlab', 'dockerhub', 'huggingface'],
|
|
'assignment_ttl_seconds': 86400,
|
|
'assignment_ttl_seconds_by_source': {'dockerhub': 90000},
|
|
'bundle_body_timeout_seconds': 1800,
|
|
}},
|
|
}
|
|
self.runtime_config_provider = mock.Mock(return_value=SimpleNamespace(
|
|
config=self.runtime_config,
|
|
))
|
|
self.document_loader = mock.Mock(side_effect=lambda _path, document: SimpleNamespace(
|
|
document=document, source='candidate',
|
|
text=(
|
|
'global:\n project_dir: /opt/truf/app\n'
|
|
'sources:\n gitlab:\n timeout: 600\n'
|
|
' dockerhub:\n timeout: 900\n'
|
|
' huggingface:\n timeout: 300\n'
|
|
'supervisor:\n worker_api:\n'
|
|
' sources: [gitlab, dockerhub, huggingface]\n'
|
|
' assignment_ttl_seconds: 86400\n'
|
|
' assignment_ttl_seconds_by_source:\n dockerhub: 90000\n'
|
|
' bundle_body_timeout_seconds: 1800\n'
|
|
if document == 'config' else
|
|
'auth_pools:\n fixture:\n - name: fixture\n token: editor-secret\n'
|
|
),
|
|
state=self.document_state,
|
|
selected=(
|
|
self.document_state.candidate_config
|
|
if document == 'config' else self.document_state.candidate_secrets
|
|
),
|
|
))
|
|
self.candidate_preview_provider = mock.Mock(side_effect=lambda _path, document, payload: SimpleNamespace(
|
|
document=document, state=self.document_state,
|
|
proposed=SimpleNamespace(
|
|
sha256=hashlib.sha256(payload).hexdigest(),
|
|
byte_count=len(payload), present=True,
|
|
),
|
|
diff=(
|
|
SimpleNamespace(entries=(), truncated=False, format_only_changed=False)
|
|
if document == 'config' else
|
|
SimpleNamespace(**{
|
|
name: value for name, value in (
|
|
('document_changed', True), ('semantic_changed', True),
|
|
('pools_before', 1), ('pools_after', 1), ('pools_added', 0),
|
|
('pools_removed', 0), ('pools_changed', 1),
|
|
('entries_before', 1), ('entries_after', 1),
|
|
('entries_added', 0), ('entries_removed', 0),
|
|
('pools_reordered', 0), ('usernames_added', 0),
|
|
('usernames_removed', 0), ('usernames_changed', 0),
|
|
('tokens_changed', 1),
|
|
)
|
|
})
|
|
),
|
|
))
|
|
def save_candidate(_path, document, payload, **_kwargs):
|
|
proposed = SimpleNamespace(
|
|
sha256=hashlib.sha256(payload).hexdigest(),
|
|
byte_count=len(payload), present=True,
|
|
)
|
|
if document == 'config':
|
|
self.document_state.candidate_config = proposed
|
|
else:
|
|
self.document_state.candidate_secrets = proposed
|
|
return SimpleNamespace(
|
|
document=document, proposed=proposed, written=True,
|
|
)
|
|
|
|
self.candidate_save_provider = mock.Mock(side_effect=save_candidate)
|
|
self.candidate_verify_provider = mock.Mock(return_value=SimpleNamespace(action='apply-config'))
|
|
self.runtime_apply_provider = mock.Mock()
|
|
self.managed_file_traversal = FakeManagedFileTraversal()
|
|
managed_limits = ManagedFileLimits(1024, 255, 16, 100, 65536, 65536)
|
|
self.managed_file_registry = ManagedFileRootRegistry((
|
|
ManagedFileRoot(
|
|
'exports', '/data/managed-files/private-host-path',
|
|
ManagedFilePermissions(True, True, True, True), managed_limits,
|
|
),
|
|
ManagedFileRoot(
|
|
'readonly', '/data/managed-files/private-readonly-path',
|
|
ManagedFilePermissions(True, True, False, False), managed_limits,
|
|
),
|
|
))
|
|
self.admin = AdminService(
|
|
'postgresql://fixture', ORIGIN, MARKER, db_factory=RecordingDB,
|
|
supervisor_metadata={
|
|
'instance_id': 'fixture', 'token': 'supervisor-token-must-not-render',
|
|
},
|
|
runtime_snapshot_provider=self.runtime_provider,
|
|
source_action_provider=self.source_action_provider,
|
|
dashboard_action_provider=self.dashboard_action_provider,
|
|
source_log_provider=self.source_log_provider,
|
|
package_compatibility_provider=self.package_compatibility_provider,
|
|
runtime_config_path='/data/config/runtime.yaml',
|
|
runtime_config_provider=self.runtime_config_provider,
|
|
document_loader=self.document_loader,
|
|
candidate_preview_provider=self.candidate_preview_provider,
|
|
candidate_save_provider=self.candidate_save_provider,
|
|
candidate_verify_provider=self.candidate_verify_provider,
|
|
runtime_apply_provider=self.runtime_apply_provider,
|
|
managed_file_roots=self.managed_file_registry,
|
|
)
|
|
self.traversal_patcher = mock.patch(
|
|
'worker_api.ManagedFileTraversal',
|
|
return_value=self.managed_file_traversal,
|
|
)
|
|
self.traversal_patcher.start()
|
|
|
|
def tearDown(self):
|
|
self.traversal_patcher.stop()
|
|
self.temp_dir.cleanup()
|
|
|
|
def _client(self, enabled=True):
|
|
return TestClient(create_worker_app(
|
|
self.worker, reaper_interval_seconds=3600,
|
|
admin_service=self.admin if enabled else None,
|
|
))
|
|
|
|
def _headers(self, **extra):
|
|
return {
|
|
EDGE_MARKER_HEADER: MARKER,
|
|
OPERATOR_HEADER: OPERATOR,
|
|
'Origin': ORIGIN,
|
|
**extra,
|
|
}
|
|
|
|
def _post(self, client, path, fields):
|
|
fields = dict(fields)
|
|
if path.startswith(('/users/', '/devices/', '/queue/')):
|
|
fields.setdefault('operation_id', str(uuid.uuid4()))
|
|
return client.post(
|
|
ADMIN_PREFIX + path,
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(),
|
|
)
|
|
|
|
def test_managed_file_registry_pages_and_download_are_logical_and_bounded(self):
|
|
self.assertIs(self.admin.managed_file_roots, self.managed_file_registry)
|
|
registry = ManagedFileRootRegistry((object(),))
|
|
service = AdminService(
|
|
'postgresql://fixture', ORIGIN, MARKER, db_factory=RecordingDB,
|
|
managed_file_roots=registry,
|
|
)
|
|
self.assertIs(service.managed_file_roots, registry)
|
|
with self.assertRaisesRegex(ValueError, 'registry is invalid'):
|
|
AdminService(
|
|
'postgresql://fixture', ORIGIN, MARKER, db_factory=RecordingDB,
|
|
managed_file_roots={},
|
|
)
|
|
with self._client() as client:
|
|
index = client.get(ADMIN_PREFIX + '/files', headers=self._headers())
|
|
listing = client.get(
|
|
ADMIN_PREFIX + '/files?root_id=exports', headers=self._headers(),
|
|
)
|
|
nested = client.get(
|
|
ADMIN_PREFIX + '/files?root_id=exports&relative_path=nested',
|
|
headers=self._headers(),
|
|
)
|
|
download = client.get(
|
|
ADMIN_PREFIX
|
|
+ '/files/download?root_id=exports&relative_path=nested%2Freport.bin',
|
|
headers=self._headers(),
|
|
)
|
|
for response in (index, listing, nested, download):
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn('/data/managed-files', response.text)
|
|
self.assertIn('>Files<', index.text)
|
|
self.assertIn('exports', index.text)
|
|
self.assertIn('nested/', listing.text)
|
|
self.assertIn('report.bin', nested.text)
|
|
self.assertEqual(download.content, b'\x00\xffreport')
|
|
self.assertEqual(download.headers['content-type'], 'application/octet-stream')
|
|
self.assertEqual(download.headers['content-length'], '8')
|
|
self.assertEqual(
|
|
download.headers['etag'],
|
|
'"' + hashlib.sha256(b'\x00\xffreport').hexdigest() + '"',
|
|
)
|
|
self.assertIn("filename*=UTF-8''report.bin", download.headers['content-disposition'])
|
|
|
|
def test_managed_file_snapshot_is_streamed_and_closed(self):
|
|
self.managed_file_traversal.stream_downloads = True
|
|
with self._client() as client:
|
|
download = client.get(
|
|
ADMIN_PREFIX
|
|
+ '/files/download?root_id=exports&relative_path=nested%2Freport.bin',
|
|
headers=self._headers(),
|
|
)
|
|
self.assertEqual(download.status_code, 200, download.text)
|
|
self.assertEqual(download.content, b'\x00\xffreport')
|
|
self.assertEqual(download.headers['content-length'], '8')
|
|
self.assertEqual(self.managed_file_traversal.last_snapshot.closed, 1)
|
|
|
|
def test_cancelled_managed_file_download_closes_completed_snapshot(self):
|
|
snapshot = FakeManagedFileSnapshot(b'bounded snapshot')
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
class BlockingService:
|
|
@staticmethod
|
|
def download_managed_file(*_args):
|
|
started.set()
|
|
if not release.wait(5):
|
|
raise RuntimeError('test download timed out')
|
|
return ManagedFileDownload(
|
|
ManagedFileIdentity('0' * 64, len(snapshot.content)),
|
|
snapshot=snapshot,
|
|
)
|
|
|
|
async def scenario():
|
|
task = asyncio.create_task(_download_managed_file(
|
|
BlockingService(), object(), 'runtime-results',
|
|
'scan_results.jsonl',
|
|
))
|
|
self.assertTrue(await asyncio.to_thread(started.wait, 2))
|
|
task.cancel()
|
|
release.set()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await task
|
|
for _ in range(20):
|
|
if snapshot.closed:
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
|
|
asyncio.run(scenario())
|
|
self.assertEqual(snapshot.closed, 1)
|
|
|
|
def test_streaming_response_construction_failure_closes_snapshot(self):
|
|
snapshot = FakeManagedFileSnapshot(b'bounded snapshot')
|
|
download = ManagedFileDownload(
|
|
ManagedFileIdentity('0' * 64, len(snapshot.content)),
|
|
snapshot=snapshot,
|
|
)
|
|
with mock.patch(
|
|
'admin_api._ManagedFileStreamingResponse',
|
|
side_effect=RuntimeError('failed')):
|
|
with self.assertRaisesRegex(RuntimeError, 'failed'):
|
|
_managed_file_download_response(
|
|
download, 'scan_results.jsonl',
|
|
)
|
|
self.assertEqual(snapshot.closed, 1)
|
|
|
|
def test_streaming_send_failure_closes_snapshot(self):
|
|
snapshot = FakeManagedFileSnapshot(b'bounded snapshot')
|
|
response = _ManagedFileStreamingResponse(
|
|
snapshot, snapshot.chunks(), media_type='application/octet-stream',
|
|
)
|
|
|
|
async def scenario():
|
|
async def receive():
|
|
return {'type': 'http.disconnect'}
|
|
|
|
async def send(_message):
|
|
raise RuntimeError('send failed')
|
|
|
|
with self.assertRaisesRegex(RuntimeError, 'send failed'):
|
|
await response(
|
|
{
|
|
'type': 'http',
|
|
'method': 'GET',
|
|
'path': '/',
|
|
'headers': [],
|
|
'asgi': {'version': '3.0', 'spec_version': '2.4'},
|
|
},
|
|
receive,
|
|
send,
|
|
)
|
|
|
|
asyncio.run(scenario())
|
|
self.assertEqual(snapshot.closed, 1)
|
|
|
|
def test_managed_file_mutations_are_typed_audited_and_content_free(self):
|
|
created_payload = b'created-binary-\x00-credential-sentinel'
|
|
replacement_payload = b'replaced-binary-\xff-credential-sentinel'
|
|
with self._client() as client:
|
|
create_id = str(uuid.uuid4())
|
|
created = self._post(client, '/files/create', {
|
|
'operation_id': create_id, 'root_id': 'exports',
|
|
'relative_path': 'new.bin',
|
|
'content_base64': __import__('base64').urlsafe_b64encode(
|
|
created_payload,
|
|
).decode('ascii'),
|
|
})
|
|
created_hash = hashlib.sha256(created_payload).hexdigest()
|
|
replace_id = str(uuid.uuid4())
|
|
replaced = self._post(client, '/files/replace', {
|
|
'operation_id': replace_id, 'root_id': 'exports',
|
|
'relative_path': 'new.bin', 'expected_sha256': created_hash,
|
|
'content_base64': __import__('base64').urlsafe_b64encode(
|
|
replacement_payload,
|
|
).decode('ascii'),
|
|
})
|
|
replacement_hash = hashlib.sha256(replacement_payload).hexdigest()
|
|
delete_id = str(uuid.uuid4())
|
|
deleted = self._post(client, '/files/delete', {
|
|
'operation_id': delete_id, 'root_id': 'exports',
|
|
'relative_path': 'new.bin', 'expected_sha256': replacement_hash,
|
|
})
|
|
|
|
for response in (created, replaced, deleted):
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn('new.bin', self.managed_file_traversal.files)
|
|
operations = [
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
]
|
|
self.assertEqual(
|
|
[call[1]['action'] for call in operations],
|
|
['files.create', 'files.replace', 'files.delete'],
|
|
)
|
|
self.assertEqual([call[1]['actor'] for call in operations], [OPERATOR] * 3)
|
|
self.assertEqual(
|
|
[call[1]['operation_id'] for call in operations],
|
|
[create_id, replace_id, delete_id],
|
|
)
|
|
recorded = repr(RecordingDB.calls)
|
|
self.assertNotIn('credential-sentinel', recorded)
|
|
self.assertNotIn(__import__('base64').urlsafe_b64encode(created_payload).decode(), recorded)
|
|
|
|
def test_managed_file_create_replay_recovers_after_completion_loss(self):
|
|
payload = b'durable replay payload'
|
|
encoded = __import__('base64').urlsafe_b64encode(payload).decode('ascii')
|
|
operation_id = str(uuid.uuid4())
|
|
fields = {
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': operation_id, 'root_id': 'exports',
|
|
'relative_path': 'replayed.bin', 'content_base64': encoded,
|
|
}
|
|
RecordingDB.managed_file_completion_failures = 3
|
|
with self._client() as client:
|
|
first = client.post(
|
|
ADMIN_PREFIX + '/files/create', data=fields,
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
second = client.post(
|
|
ADMIN_PREFIX + '/files/create', data=fields,
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
terminal = client.post(
|
|
ADMIN_PREFIX + '/files/create', data=fields,
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.assertEqual(first.status_code, 503)
|
|
self.assertEqual((second.status_code, terminal.status_code), (303, 303))
|
|
self.assertSecurityHeaders(first)
|
|
self.assertEqual(self.managed_file_traversal.files['replayed.bin'], payload)
|
|
mutations = [
|
|
call for call in self.managed_file_traversal.calls
|
|
if call[0] == 'create_replace' and call[2] == 'replayed.bin'
|
|
]
|
|
self.assertEqual(len(mutations), 1)
|
|
self.assertEqual(
|
|
RecordingDB.managed_file_operations[operation_id]['status'],
|
|
'succeeded',
|
|
)
|
|
replay = self.admin.create_managed_file(
|
|
None, 'exports', 'replayed.bin', payload, OPERATOR, operation_id,
|
|
)
|
|
self.assertEqual(replay['status'], 'succeeded')
|
|
|
|
def test_managed_file_acceptance_precedes_one_concurrent_physical_mutation(self):
|
|
payload = b'concurrent managed payload'
|
|
operation_id = str(uuid.uuid4())
|
|
original = self.managed_file_traversal.create_replace_file
|
|
other_admin = AdminService(
|
|
'postgresql://fixture', ORIGIN, MARKER, db_factory=RecordingDB,
|
|
managed_file_roots=self.managed_file_registry,
|
|
)
|
|
|
|
def accepted_provider(*args, **kwargs):
|
|
self.assertEqual(
|
|
RecordingDB.managed_file_operations[operation_id]['status'],
|
|
'running',
|
|
)
|
|
return original(*args, **kwargs)
|
|
|
|
barrier = threading.Barrier(3)
|
|
results = []
|
|
|
|
def submit(service):
|
|
barrier.wait()
|
|
results.append(service.create_managed_file(
|
|
self.managed_file_traversal, 'exports', 'concurrent.bin',
|
|
payload, OPERATOR, operation_id,
|
|
))
|
|
|
|
with mock.patch.object(
|
|
self.managed_file_traversal, 'create_replace_file',
|
|
side_effect=accepted_provider):
|
|
threads = [
|
|
threading.Thread(target=submit, args=(service,))
|
|
for service in (self.admin, other_admin)
|
|
]
|
|
for thread in threads:
|
|
thread.start()
|
|
barrier.wait()
|
|
for thread in threads:
|
|
thread.join(timeout=5)
|
|
|
|
self.assertTrue(all(not thread.is_alive() for thread in threads))
|
|
self.assertEqual([result['status'] for result in results], [
|
|
'succeeded', 'succeeded',
|
|
])
|
|
mutations = [
|
|
call for call in self.managed_file_traversal.calls
|
|
if call[0] == 'create_replace' and call[2] == 'concurrent.bin'
|
|
]
|
|
self.assertEqual(len(mutations), 1)
|
|
self.assertEqual(
|
|
RecordingDB.managed_file_operations[operation_id]['status'],
|
|
'succeeded',
|
|
)
|
|
|
|
def test_fresh_managed_file_cas_loser_is_not_promoted_to_success(self):
|
|
payload = b'external matching payload'
|
|
operation_id = str(uuid.uuid4())
|
|
|
|
def lose_cas(_root_id, relative_path, value, *, expected_sha256):
|
|
self.managed_file_traversal.files[relative_path] = value
|
|
raise ManagedFileAccessError('hash_conflict')
|
|
|
|
with mock.patch.object(
|
|
self.managed_file_traversal, 'create_replace_file',
|
|
side_effect=lose_cas):
|
|
with self.assertRaises(AdminAPIError) as raised:
|
|
self.admin.create_managed_file(
|
|
self.managed_file_traversal, 'exports', 'external.bin',
|
|
payload, OPERATOR, operation_id,
|
|
)
|
|
|
|
self.assertEqual(raised.exception.status_code, 409)
|
|
self.assertEqual(
|
|
RecordingDB.managed_file_operations[operation_id]['status'],
|
|
'failed',
|
|
)
|
|
|
|
def assertSecurityHeaders(self, response):
|
|
for name, expected in SECURITY_HEADERS.items():
|
|
self.assertEqual(response.headers.get(name), expected)
|
|
|
|
def _document_fields(self, text, operation_id=None):
|
|
return {
|
|
'document_text': text,
|
|
'operation_id': operation_id or str(uuid.uuid4()),
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
'expected_candidate_config_sha256': 'c' * 64,
|
|
'expected_candidate_secrets_sha256': 'd' * 64,
|
|
}
|
|
|
|
def test_runtime_document_pages_preview_save_and_apply_are_exact_and_no_store(self):
|
|
with self._client() as client:
|
|
config = client.get(ADMIN_PREFIX + '/config', headers=self._headers())
|
|
secrets = client.get(ADMIN_PREFIX + '/secrets', headers=self._headers())
|
|
self.assertEqual(config.status_code, 200)
|
|
self.assertEqual(secrets.status_code, 200)
|
|
self.assertSecurityHeaders(config)
|
|
self.assertSecurityHeaders(secrets)
|
|
self.assertIn('Config YAML', config.text)
|
|
self.assertNotIn('editor-secret', config.text)
|
|
self.assertIn('editor-secret', secrets.text)
|
|
self.assertNotIn('<script', secrets.text.lower())
|
|
self.assertIn('./config', config.text)
|
|
self.assertIn('./secrets', config.text)
|
|
|
|
secret_text = 'auth_pools:\n p:\n - name: n\n token: new-secret\n'
|
|
preview = self._post(
|
|
client, '/secrets/preview', self._document_fields(secret_text),
|
|
)
|
|
self.assertEqual(preview.status_code, 200)
|
|
self.assertEqual(preview.text.count('new-secret'), 1)
|
|
self.assertIn('Tokens Changed', preview.text)
|
|
self.assertIn('nothing was saved', preview.text)
|
|
|
|
operation_id = str(uuid.uuid4())
|
|
save = client.post(
|
|
ADMIN_PREFIX + '/config/save',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
**self._document_fields('global:\n project_dir: /opt/truf/app\n', operation_id),
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.assertEqual(save.status_code, 303)
|
|
self.assertEqual(save.headers['location'], '../config')
|
|
creates = [call for call in RecordingDB.calls if call[0] == 'document_operation_create']
|
|
completes = [call for call in RecordingDB.calls if call[0] == 'document_operation_complete']
|
|
self.assertEqual(len(creates), 1)
|
|
self.assertEqual(creates[0][1]['actor'], OPERATOR)
|
|
self.assertEqual(creates[0][1]['operation_id'], operation_id)
|
|
self.assertEqual(len(completes), 1)
|
|
self.assertNotIn('project_dir', repr(creates + completes))
|
|
|
|
apply_id = str(uuid.uuid4())
|
|
apply_response = client.post(
|
|
ADMIN_PREFIX + '/config/apply',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': apply_id,
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
'expected_candidate_config_sha256': 'c' * 64,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.assertEqual(apply_response.status_code, 303)
|
|
self.assertEqual(
|
|
apply_response.headers['location'],
|
|
f'../operations/{apply_id}',
|
|
)
|
|
self.assertEqual(
|
|
urljoin(
|
|
'https://admin.example.test/private-prefix/config/apply',
|
|
apply_response.headers['location'],
|
|
),
|
|
f'https://admin.example.test/private-prefix/operations/{apply_id}',
|
|
)
|
|
self.runtime_apply_provider.assert_called_once_with(
|
|
operation_id=apply_id, action='apply-config',
|
|
active_config_sha256='a' * 64,
|
|
active_secrets_sha256='b' * 64,
|
|
candidate_config_sha256='c' * 64,
|
|
candidate_secrets_sha256=None,
|
|
)
|
|
runtime_create = [call for call in RecordingDB.calls if call[0] == 'runtime_operation_create'][-1]
|
|
self.assertEqual(runtime_create[1]['actor'], OPERATOR)
|
|
self.assertEqual(runtime_create[1]['action'], 'apply-config')
|
|
|
|
def test_runtime_apply_without_agent_fails_before_operation_creation(self):
|
|
self.admin.runtime_apply_provider = None
|
|
with self._client() as client:
|
|
response = client.post(
|
|
ADMIN_PREFIX + '/secrets/apply',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': str(uuid.uuid4()),
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
'expected_candidate_secrets_sha256': 'd' * 64,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.assertEqual(response.status_code, 503)
|
|
self.assertFalse(any(call[0] == 'runtime_operation_create' for call in RecordingDB.calls))
|
|
|
|
def test_runtime_document_hides_absent_and_noop_apply_actions(self):
|
|
self.document_state.candidate_config = SimpleNamespace(
|
|
sha256=self.document_state.active_config.sha256,
|
|
byte_count=10,
|
|
present=True,
|
|
)
|
|
self.document_state.candidate_secrets = SimpleNamespace(
|
|
sha256='', byte_count=0, present=False,
|
|
)
|
|
|
|
with self._client() as client:
|
|
response = client.get(ADMIN_PREFIX + '/config', headers=self._headers())
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertNotIn('action="./config/apply"', response.text)
|
|
self.assertNotIn('action="./runtime/apply-both"', response.text)
|
|
self.assertIn(
|
|
'No staged candidate differs from the active document.',
|
|
response.text,
|
|
)
|
|
|
|
def test_remaining_runtime_document_routes_accept_only_bound_hash_forms(self):
|
|
config_text = 'global:\n project_dir: /opt/truf/app\n'
|
|
secret_text = 'auth_pools:\n p:\n - name: n\n token: route-secret\n'
|
|
secret_sha256 = hashlib.sha256(secret_text.encode('utf-8')).hexdigest()
|
|
with self._client() as client:
|
|
config_preview = self._post(
|
|
client, '/config/preview', self._document_fields(config_text),
|
|
)
|
|
secret_save = client.post(
|
|
ADMIN_PREFIX + '/secrets/save',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
**self._document_fields(secret_text),
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
secret_apply_id = str(uuid.uuid4())
|
|
secret_apply = client.post(
|
|
ADMIN_PREFIX + '/secrets/apply',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': secret_apply_id,
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
'expected_candidate_secrets_sha256': secret_sha256,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
both_apply_id = str(uuid.uuid4())
|
|
both_apply = client.post(
|
|
ADMIN_PREFIX + '/runtime/apply-both',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': both_apply_id,
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
'expected_candidate_config_sha256': 'c' * 64,
|
|
'expected_candidate_secrets_sha256': secret_sha256,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual(config_preview.status_code, 200)
|
|
self.assertEqual(secret_save.status_code, 303)
|
|
self.assertEqual(secret_save.headers['location'], '../secrets')
|
|
self.assertEqual(secret_apply.status_code, 303)
|
|
self.assertEqual(
|
|
secret_apply.headers['location'], f'../operations/{secret_apply_id}',
|
|
)
|
|
self.assertEqual(both_apply.status_code, 303)
|
|
self.assertEqual(
|
|
both_apply.headers['location'], f'../operations/{both_apply_id}',
|
|
)
|
|
self.assertEqual(
|
|
[call.kwargs['action'] for call in self.runtime_apply_provider.call_args_list],
|
|
['apply-secrets', 'apply-both'],
|
|
)
|
|
self.assertEqual(
|
|
[call[1]['action'] for call in RecordingDB.calls
|
|
if call[0] == 'document_operation_create'],
|
|
['runtime.secrets.save'],
|
|
)
|
|
|
|
def test_runtime_document_stale_save_and_apply_forms_fail_before_mutation(self):
|
|
stale_hash = 'e' * 64
|
|
save_cases = [
|
|
(document, field)
|
|
for document in ('config', 'secrets')
|
|
for field in (
|
|
'expected_active_config_sha256',
|
|
'expected_active_secrets_sha256',
|
|
'expected_candidate_config_sha256',
|
|
'expected_candidate_secrets_sha256',
|
|
)
|
|
]
|
|
|
|
correct_verify = {
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
'expected_candidate_config_sha256': 'c' * 64,
|
|
'expected_candidate_secrets_sha256': 'd' * 64,
|
|
}
|
|
|
|
def verify_stale(_path, action, **hashes):
|
|
expected = dict(correct_verify)
|
|
if action == 'apply-config':
|
|
expected['expected_candidate_secrets_sha256'] = None
|
|
elif action == 'apply-secrets':
|
|
expected['expected_candidate_config_sha256'] = None
|
|
if hashes != expected:
|
|
raise RuntimeDocumentError(
|
|
'reference', document='config', path='revision',
|
|
)
|
|
return SimpleNamespace(action=action)
|
|
|
|
self.candidate_verify_provider.side_effect = verify_stale
|
|
apply_cases = (
|
|
('config', 'expected_active_config_sha256'),
|
|
('config', 'expected_active_secrets_sha256'),
|
|
('config', 'expected_candidate_config_sha256'),
|
|
('secrets', 'expected_active_config_sha256'),
|
|
('secrets', 'expected_active_secrets_sha256'),
|
|
('secrets', 'expected_candidate_secrets_sha256'),
|
|
('both', 'expected_active_config_sha256'),
|
|
('both', 'expected_active_secrets_sha256'),
|
|
('both', 'expected_candidate_config_sha256'),
|
|
('both', 'expected_candidate_secrets_sha256'),
|
|
)
|
|
responses = []
|
|
with self._client() as client:
|
|
for document, stale_field in save_cases:
|
|
fields = self._document_fields(
|
|
'global: {}\n' if document == 'config' else 'auth_pools: {}\n',
|
|
)
|
|
fields[stale_field] = stale_hash
|
|
responses.append(client.post(
|
|
ADMIN_PREFIX + f'/{document}/save',
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(), follow_redirects=False,
|
|
))
|
|
for document, stale_field in apply_cases:
|
|
action = f'apply-{document}' if document != 'both' else 'apply-both'
|
|
fields = {
|
|
'operation_id': str(uuid.uuid4()),
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
}
|
|
if action in ('apply-config', 'apply-both'):
|
|
fields['expected_candidate_config_sha256'] = 'c' * 64
|
|
if action in ('apply-secrets', 'apply-both'):
|
|
fields['expected_candidate_secrets_sha256'] = 'd' * 64
|
|
fields[stale_field] = stale_hash
|
|
route = f'/{document}/apply' if document != 'both' else '/runtime/apply-both'
|
|
responses.append(client.post(
|
|
ADMIN_PREFIX + route,
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(), follow_redirects=False,
|
|
))
|
|
|
|
self.assertEqual([response.status_code for response in responses], [409] * 18)
|
|
for response in responses:
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn(stale_hash, response.text)
|
|
self.candidate_save_provider.assert_not_called()
|
|
self.runtime_apply_provider.assert_not_called()
|
|
self.assertFalse(any(
|
|
call[0] in {'document_operation_create', 'runtime_operation_create'}
|
|
for call in RecordingDB.calls
|
|
))
|
|
|
|
def test_runtime_document_save_and_apply_recover_exact_request_replay(self):
|
|
save_id = str(uuid.uuid4())
|
|
document_text = 'global:\n project_dir: /opt/truf/app\n'
|
|
fields = self._document_fields(document_text, save_id)
|
|
proposed_sha256 = hashlib.sha256(document_text.encode('utf-8')).hexdigest()
|
|
RecordingDB.document_completion_failures = 4
|
|
with self._client() as client:
|
|
first = client.post(
|
|
ADMIN_PREFIX + '/config/save',
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.assertEqual(first.status_code, 503)
|
|
self.assertIn(f'name="operation_id" value="{save_id}"', first.text)
|
|
self.assertIn(
|
|
f'name="expected_candidate_config_sha256" value="{proposed_sha256}"',
|
|
first.text,
|
|
)
|
|
second = client.post(
|
|
ADMIN_PREFIX + '/config/save',
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.assertEqual(second.status_code, 303)
|
|
self.assertEqual(self.candidate_save_provider.call_count, 1)
|
|
self.assertEqual(RecordingDB.document_operations[save_id]['status'], 'succeeded')
|
|
|
|
apply_id = str(uuid.uuid4())
|
|
self.runtime_apply_provider.reset_mock()
|
|
self.runtime_apply_provider.side_effect = [RuntimeError('lost response'), None]
|
|
apply_fields = {
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': apply_id,
|
|
'expected_active_config_sha256': 'a' * 64,
|
|
'expected_active_secrets_sha256': 'b' * 64,
|
|
'expected_candidate_config_sha256': 'c' * 64,
|
|
}
|
|
with self._client() as client:
|
|
failed = client.post(
|
|
ADMIN_PREFIX + '/config/apply', data=apply_fields,
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
retried = client.post(
|
|
ADMIN_PREFIX + '/config/apply', data=apply_fields,
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.assertEqual(failed.status_code, 502)
|
|
self.assertEqual(retried.status_code, 303)
|
|
self.assertEqual(self.runtime_apply_provider.call_count, 2)
|
|
self.assertEqual(self.candidate_verify_provider.call_count, 1)
|
|
|
|
def test_runtime_document_save_reconciles_a_concurrent_exact_write(self):
|
|
document_text = 'global:\n project_dir: /opt/truf/app\n'
|
|
candidate_bytes = document_text.encode('utf-8')
|
|
proposed = SimpleNamespace(
|
|
sha256=hashlib.sha256(candidate_bytes).hexdigest(),
|
|
byte_count=len(candidate_bytes), present=True,
|
|
)
|
|
|
|
def concurrent_write(_path, _document, _payload, **_kwargs):
|
|
self.document_state.candidate_config = proposed
|
|
raise RuntimeDocumentError(
|
|
'reference', document='config', path='revision',
|
|
)
|
|
|
|
self.admin.candidate_save_provider = mock.Mock(side_effect=concurrent_write)
|
|
operation_id = str(uuid.uuid4())
|
|
result = self.admin.save_runtime_document_candidate(
|
|
'config', document_text, OPERATOR, operation_id,
|
|
expected_hashes={
|
|
'active_config': 'a' * 64,
|
|
'active_secrets': 'b' * 64,
|
|
'candidate_config': 'c' * 64,
|
|
'candidate_secrets': 'd' * 64,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result['status'], 'succeeded')
|
|
completions = [
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'document_operation_complete'
|
|
]
|
|
self.assertEqual(completions[-1][2:], (True, proposed.sha256, True))
|
|
|
|
def test_runtime_document_replay_remembers_candidate_was_absent(self):
|
|
document_text = 'global:\n project_dir: /opt/truf/app\n'
|
|
digest = hashlib.sha256(document_text.encode('utf-8')).hexdigest()
|
|
byte_count = len(document_text.encode('utf-8'))
|
|
self.document_state.active_config = SimpleNamespace(
|
|
sha256=digest, byte_count=byte_count, present=True,
|
|
)
|
|
self.document_state.candidate_config = SimpleNamespace(
|
|
sha256=digest, byte_count=byte_count, present=False,
|
|
)
|
|
operation_id = str(uuid.uuid4())
|
|
expected_hashes = {
|
|
'active_config': digest,
|
|
'active_secrets': 'b' * 64,
|
|
'candidate_config': digest,
|
|
'candidate_secrets': 'd' * 64,
|
|
}
|
|
RecordingDB.document_completion_failures = 3
|
|
|
|
with self.assertRaisesRegex(AdminAPIError, 'completion is pending'):
|
|
self.admin.save_runtime_document_candidate(
|
|
'config', document_text, OPERATOR, operation_id,
|
|
expected_hashes=expected_hashes,
|
|
)
|
|
result = self.admin.save_runtime_document_candidate(
|
|
'config', document_text, OPERATOR, operation_id,
|
|
expected_hashes=expected_hashes,
|
|
)
|
|
|
|
self.assertEqual(result['status'], 'succeeded')
|
|
self.assertFalse(
|
|
RecordingDB.document_operations[operation_id]['expected_identity'][
|
|
'candidate_before_present'
|
|
]
|
|
)
|
|
completions = [
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'document_operation_complete'
|
|
]
|
|
self.assertEqual(completions[-1][2:], (True, digest, True))
|
|
|
|
def test_runtime_document_running_replay_rejects_every_stale_file(self):
|
|
document_text = 'global:\n project_dir: /opt/truf/app\n'
|
|
expected_hashes = {
|
|
'active_config': 'a' * 64,
|
|
'active_secrets': 'b' * 64,
|
|
'candidate_config': 'c' * 64,
|
|
'candidate_secrets': 'd' * 64,
|
|
}
|
|
mutations = {
|
|
'active': ('active_config', 'e' * 64),
|
|
'counterpart': ('candidate_secrets', 'f' * 64),
|
|
'selected': ('candidate_config', '9' * 64),
|
|
}
|
|
|
|
for label, (attribute, stale_sha256) in mutations.items():
|
|
with self.subTest(label=label):
|
|
RecordingDB.calls = []
|
|
RecordingDB.document_operations = {}
|
|
RecordingDB.document_completion_failures = 3
|
|
self.document_state.active_config = SimpleNamespace(
|
|
sha256='a' * 64, byte_count=10, present=True,
|
|
)
|
|
self.document_state.active_secrets = SimpleNamespace(
|
|
sha256='b' * 64, byte_count=10, present=True,
|
|
)
|
|
self.document_state.candidate_config = SimpleNamespace(
|
|
sha256='c' * 64, byte_count=10, present=True,
|
|
)
|
|
self.document_state.candidate_secrets = SimpleNamespace(
|
|
sha256='d' * 64, byte_count=10, present=True,
|
|
)
|
|
operation_id = str(uuid.uuid4())
|
|
with self.assertRaisesRegex(AdminAPIError, 'completion is pending'):
|
|
self.admin.save_runtime_document_candidate(
|
|
'config', document_text, OPERATOR, operation_id,
|
|
expected_hashes=expected_hashes,
|
|
)
|
|
setattr(self.document_state, attribute, SimpleNamespace(
|
|
sha256=stale_sha256, byte_count=11, present=True,
|
|
))
|
|
|
|
with self.assertRaisesRegex(AdminAPIError, 'revision changed'):
|
|
self.admin.save_runtime_document_candidate(
|
|
'config', document_text, OPERATOR, operation_id,
|
|
expected_hashes=expected_hashes,
|
|
)
|
|
|
|
def test_terminal_apply_replay_needs_no_provider_or_filesystem_verification(self):
|
|
operation_id = str(uuid.uuid4())
|
|
expected_hashes = {
|
|
'active_config': 'a' * 64,
|
|
'active_secrets': 'b' * 64,
|
|
'candidate_config': 'c' * 64,
|
|
}
|
|
first = self.admin.request_runtime_apply(
|
|
'apply-config', OPERATOR, operation_id,
|
|
expected_hashes=expected_hashes,
|
|
)
|
|
self.assertEqual(first['operation']['status'], 'requested')
|
|
RecordingDB.apply_operations[operation_id]['status'] = 'succeeded'
|
|
self.admin.runtime_apply_provider = None
|
|
self.candidate_verify_provider.reset_mock()
|
|
|
|
replay = self.admin.request_runtime_apply(
|
|
'apply-config', OPERATOR, operation_id,
|
|
expected_hashes=expected_hashes,
|
|
)
|
|
|
|
self.assertEqual(replay['operation']['status'], 'succeeded')
|
|
self.assertIsNone(replay['verification'])
|
|
self.candidate_verify_provider.assert_not_called()
|
|
|
|
def test_runtime_document_cancellation_drops_plaintext_from_admin_tracebacks(self):
|
|
secret = 'traceback-document-secret +%/& sentinel'
|
|
|
|
class StreamingRequest:
|
|
def __init__(self, body, *, cancel_after_chunk=False):
|
|
self.headers = Headers({
|
|
'origin': ORIGIN,
|
|
'content-type': 'application/x-www-form-urlencoded',
|
|
'content-length': str(len(body)),
|
|
})
|
|
self.state = SimpleNamespace(admin_actor=OPERATOR)
|
|
self.body = body
|
|
self.cancel_after_chunk = cancel_after_chunk
|
|
|
|
async def stream(self):
|
|
yield self.body
|
|
if self.cancel_after_chunk:
|
|
raise asyncio.CancelledError()
|
|
|
|
encoded_secret = urlencode({'document_text': secret}).split('=', 1)[1]
|
|
|
|
def assert_admin_traceback_is_clean(error):
|
|
current = error.__traceback__
|
|
found_admin_frame = False
|
|
while current is not None:
|
|
if Path(current.tb_frame.f_code.co_filename).name == 'admin_api.py':
|
|
found_admin_frame = True
|
|
self.assertNotIn(secret, repr(current.tb_frame.f_locals))
|
|
self.assertNotIn(encoded_secret, repr(current.tb_frame.f_locals))
|
|
current = current.tb_next
|
|
self.assertTrue(found_admin_frame)
|
|
|
|
encoded_body = urlencode({'document_text': secret}).encode('ascii')
|
|
try:
|
|
asyncio.run(_form_fields(
|
|
StreamingRequest(encoded_body, cancel_after_chunk=True), self.admin,
|
|
{'csrf_token', 'document_text'},
|
|
))
|
|
except asyncio.CancelledError as error:
|
|
assert_admin_traceback_is_clean(error)
|
|
else:
|
|
self.fail('form stream cancellation was not raised')
|
|
|
|
fields = {
|
|
'csrf_token': self.admin.csrf_token,
|
|
**self._document_fields(secret),
|
|
}
|
|
body = urlencode(fields).encode('ascii')
|
|
|
|
async def cancel_after_decode(*_args, **_kwargs):
|
|
raise asyncio.CancelledError()
|
|
|
|
try:
|
|
with mock.patch('admin_api.asyncio.to_thread', new=cancel_after_decode):
|
|
asyncio.run(_dispatch_runtime_document(
|
|
StreamingRequest(body), self.admin, 'secrets', 'preview',
|
|
))
|
|
except asyncio.CancelledError as error:
|
|
assert_admin_traceback_is_clean(error)
|
|
else:
|
|
self.fail('document dispatch cancellation was not raised')
|
|
|
|
def test_managed_file_cancellation_drops_content_from_admin_tracebacks(self):
|
|
content = b'traceback-managed-file-content +%/& sentinel'
|
|
encoded_content = __import__('base64').urlsafe_b64encode(content).decode('ascii')
|
|
body = urlencode({
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': str(uuid.uuid4()),
|
|
'root_id': 'exports',
|
|
'relative_path': 'cancelled.bin',
|
|
'content_base64': encoded_content,
|
|
}).encode('ascii')
|
|
|
|
class Request:
|
|
headers = Headers({
|
|
'origin': ORIGIN,
|
|
'content-type': 'application/x-www-form-urlencoded',
|
|
'content-length': str(len(body)),
|
|
})
|
|
state = SimpleNamespace(admin_actor=OPERATOR)
|
|
app = SimpleNamespace(state=SimpleNamespace(
|
|
managed_file_traversal=self.managed_file_traversal,
|
|
))
|
|
|
|
async def stream(self):
|
|
yield body
|
|
|
|
async def cancel_after_decode(*_args, **_kwargs):
|
|
raise asyncio.CancelledError()
|
|
|
|
try:
|
|
with mock.patch('admin_api.asyncio.to_thread', new=cancel_after_decode):
|
|
asyncio.run(_dispatch_managed_file_mutation(
|
|
Request(), self.admin, 'create',
|
|
))
|
|
except asyncio.CancelledError as error:
|
|
current = error.__traceback__
|
|
found_admin_frame = False
|
|
while current is not None:
|
|
if Path(current.tb_frame.f_code.co_filename).name == 'admin_api.py':
|
|
found_admin_frame = True
|
|
locals_text = repr(current.tb_frame.f_locals)
|
|
self.assertNotIn(content.decode('ascii'), locals_text)
|
|
self.assertNotIn(encoded_content, locals_text)
|
|
current = current.tb_next
|
|
self.assertTrue(found_admin_frame)
|
|
else:
|
|
self.fail('managed file dispatch cancellation was not raised')
|
|
|
|
def test_managed_file_errors_drop_encoded_and_decoded_content_from_tracebacks(self):
|
|
content = b'traceback-managed-file-error-content-sentinel'
|
|
encoded_content = __import__('base64').urlsafe_b64encode(content).decode('ascii')
|
|
|
|
def assert_clean(error):
|
|
pending = [error]
|
|
found_admin_frame = False
|
|
while pending:
|
|
current_error = pending.pop()
|
|
if current_error.__cause__ is not None:
|
|
pending.append(current_error.__cause__)
|
|
current = current_error.__traceback__
|
|
while current is not None:
|
|
if Path(current.tb_frame.f_code.co_filename).name == 'admin_api.py':
|
|
found_admin_frame = True
|
|
locals_text = repr(current.tb_frame.f_locals)
|
|
self.assertNotIn(content.decode('ascii'), locals_text)
|
|
self.assertNotIn(encoded_content, locals_text)
|
|
current = current.tb_next
|
|
self.assertTrue(found_admin_frame)
|
|
|
|
invalid = encoded_content + '!'
|
|
try:
|
|
_parse_managed_file_content(invalid)
|
|
except AdminAPIError as error:
|
|
assert_clean(error)
|
|
else:
|
|
self.fail('invalid Base64 was not rejected')
|
|
|
|
self.managed_file_traversal.error_category = 'filesystem_unavailable'
|
|
try:
|
|
self.admin.create_managed_file(
|
|
self.managed_file_traversal, 'exports', 'failed.bin', content,
|
|
OPERATOR, str(uuid.uuid4()),
|
|
)
|
|
except AdminAPIError as error:
|
|
assert_clean(error)
|
|
else:
|
|
self.fail('managed file provider failure was not raised')
|
|
finally:
|
|
self.managed_file_traversal.error_category = None
|
|
|
|
def test_managed_file_release_cancellation_closes_db_and_clears_payload(self):
|
|
content = b'traceback-managed-file-release-content-sentinel'
|
|
|
|
class CancelReleaseDB(RecordingDB):
|
|
instances = []
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.release_started = False
|
|
self.closed = False
|
|
self.instances.append(self)
|
|
|
|
def release_runtime_managed_file_execution(self, operation_id):
|
|
self.release_started = True
|
|
super().release_runtime_managed_file_execution(operation_id)
|
|
raise asyncio.CancelledError()
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
super().close()
|
|
|
|
service = AdminService(
|
|
'postgresql://fixture', ORIGIN, MARKER,
|
|
db_factory=CancelReleaseDB,
|
|
managed_file_roots=self.managed_file_registry,
|
|
)
|
|
try:
|
|
service.create_managed_file(
|
|
self.managed_file_traversal, 'exports', 'release-cancel.bin',
|
|
content, OPERATOR, str(uuid.uuid4()),
|
|
)
|
|
except asyncio.CancelledError as error:
|
|
current = error.__traceback__
|
|
found_admin_frame = False
|
|
while current is not None:
|
|
if Path(current.tb_frame.f_code.co_filename).name == 'admin_api.py':
|
|
found_admin_frame = True
|
|
self.assertNotIn(
|
|
content.decode('ascii'), repr(current.tb_frame.f_locals),
|
|
)
|
|
current = current.tb_next
|
|
self.assertTrue(found_admin_frame)
|
|
else:
|
|
self.fail('managed file execution release cancellation was not raised')
|
|
|
|
execution_instances = [
|
|
instance for instance in CancelReleaseDB.instances
|
|
if instance.release_started
|
|
]
|
|
self.assertEqual(len(execution_instances), 1)
|
|
self.assertTrue(execution_instances[0].closed)
|
|
self.assertFalse(RecordingDB.managed_file_execution_lock.locked())
|
|
|
|
def test_operation_status_and_audit_pages_are_durable_bounded_and_content_free(self):
|
|
operation_id = str(uuid.uuid4())
|
|
expected_identity = {
|
|
'active_config_sha256': 'a' * 64,
|
|
'active_secrets_sha256': 'b' * 64,
|
|
'candidate_config_sha256': None,
|
|
'candidate_secrets_sha256': 'd' * 64,
|
|
}
|
|
RecordingDB.apply_operations[operation_id] = {
|
|
'operation_id': operation_id,
|
|
'actor': '<operator>',
|
|
'action': 'apply-secrets',
|
|
'target_kind': 'runtime-deployment',
|
|
'target_ref': 'secrets',
|
|
'status': 'failed',
|
|
'safe_category': 'startup_failed',
|
|
'safe_detail': 'startup_failed',
|
|
'expected_revision': None,
|
|
'resulting_revision': None,
|
|
'expected_identity': expected_identity,
|
|
'resulting_identity': None,
|
|
'agent_state': 'failed',
|
|
'agent_result_sha256': 'e' * 64,
|
|
'requested_at': '2026-09-20T00:00:00+00:00',
|
|
'started_at': '2026-09-20T00:00:01+00:00',
|
|
'completed_at': '2026-09-20T00:00:02+00:00',
|
|
'agent_reconciled_at': '2026-09-20T00:00:02+00:00',
|
|
'updated_at': '2026-09-20T00:00:02+00:00',
|
|
}
|
|
RecordingDB.audit_events = [{
|
|
'id': 41,
|
|
'operation_id': operation_id,
|
|
'actor': '<operator>',
|
|
'action': 'apply-secrets',
|
|
'target_kind': 'runtime-deployment',
|
|
'target_ref': 'secrets',
|
|
'result': 'accepted',
|
|
'safe_category': None,
|
|
'before_identity': expected_identity,
|
|
'after_identity': None,
|
|
'before_bytes': None,
|
|
'after_bytes': None,
|
|
'previous_event_id': None,
|
|
'previous_event_sha256': None,
|
|
'event_sha256': 'f' * 64,
|
|
'created_at': '2026-09-20T00:00:00+00:00',
|
|
}]
|
|
RecordingDB.audit_next_before_event_id = 41
|
|
|
|
with self._client() as client:
|
|
index = client.get(
|
|
ADMIN_PREFIX + '/operations', headers=self._headers(),
|
|
)
|
|
status = client.get(
|
|
ADMIN_PREFIX + f'/operations/{operation_id}',
|
|
headers=self._headers(),
|
|
)
|
|
audit = client.get(
|
|
ADMIN_PREFIX + '/audit', headers=self._headers(),
|
|
)
|
|
older = client.get(
|
|
ADMIN_PREFIX + '/audit?before=41', headers=self._headers(),
|
|
)
|
|
|
|
for response in (index, status, audit, older):
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn('editor-secret', response.text)
|
|
self.assertNotIn('supervisor-token-must-not-render', response.text)
|
|
self.assertNotIn('<script', response.text.lower())
|
|
self.assertIn(operation_id, index.text)
|
|
self.assertIn('Persisted state remains queryable', status.text)
|
|
self.assertIn('<operator>', status.text)
|
|
self.assertIn('apply-secrets', audit.text)
|
|
self.assertIn('runtime-deployment:secrets', audit.text)
|
|
self.assertIn('?before=41', audit.text)
|
|
self.assertIn('../operations', status.text)
|
|
self.assertIn('href="./admin.css"', index.text)
|
|
self.assertIn('href="../admin.css"', status.text)
|
|
self.assertEqual(
|
|
urljoin(
|
|
f'https://admin.example.test/private-prefix/operations/{operation_id}',
|
|
'../admin.css',
|
|
),
|
|
'https://admin.example.test/private-prefix/admin.css',
|
|
)
|
|
self.assertIn(
|
|
('runtime_audit_events', None, 50), RecordingDB.calls,
|
|
)
|
|
self.assertIn(
|
|
('runtime_audit_events', 41, 50), RecordingDB.calls,
|
|
)
|
|
|
|
def test_operation_and_audit_routes_reject_unknown_or_noncanonical_input(self):
|
|
unknown = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'
|
|
noncanonical = unknown.upper()
|
|
with self._client() as client:
|
|
responses = [
|
|
client.get(
|
|
ADMIN_PREFIX + f'/operations/{unknown}',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + f'/operations/{noncanonical}',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/operations/not-a-uuid',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/audit?before=0', headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/audit?before=1&before=2',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/audit?unexpected=1', headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/operations?unexpected=1',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/operations?before=not-valid',
|
|
headers=self._headers(),
|
|
),
|
|
]
|
|
method = client.post(
|
|
ADMIN_PREFIX + '/audit', headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(
|
|
[response.status_code for response in responses],
|
|
[404, 404, 404, 400, 400, 400, 400, 400],
|
|
)
|
|
self.assertEqual(method.status_code, 404)
|
|
for response in (*responses, method):
|
|
self.assertSecurityHeaders(response)
|
|
self.assertFalse(any(
|
|
call[0] == 'runtime_audit_events' for call in RecordingDB.calls
|
|
))
|
|
|
|
def test_operations_use_bounded_composite_cursor_pagination(self):
|
|
updated_at = '2026-09-20T00:00:00+00:00'
|
|
RecordingDB.apply_operations = {
|
|
f'00000000-0000-4000-8000-{index:012d}': {
|
|
'operation_id': f'00000000-0000-4000-8000-{index:012d}',
|
|
'actor': 'fixture.operator', 'action': 'restart',
|
|
'target_ref': 'runtime', 'status': 'succeeded',
|
|
'safe_category': None, 'safe_detail': None,
|
|
'requested_at': updated_at, 'completed_at': updated_at,
|
|
'updated_at': updated_at,
|
|
}
|
|
for index in range(1, 52)
|
|
}
|
|
|
|
with self._client() as client:
|
|
first = client.get(
|
|
ADMIN_PREFIX + '/operations', headers=self._headers(),
|
|
)
|
|
cursor = re.search(r'href="\?before=([A-Za-z0-9_-]+)"', first.text)
|
|
self.assertIsNotNone(cursor, first.text)
|
|
older = client.get(
|
|
ADMIN_PREFIX + '/operations?before=' + cursor.group(1),
|
|
headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(first.status_code, 200, first.text)
|
|
self.assertEqual(older.status_code, 200, older.text)
|
|
self.assertIn('Older operations', first.text)
|
|
self.assertIn(
|
|
(
|
|
'recent_operations', 51, updated_at,
|
|
'00000000-0000-4000-8000-000000000050',
|
|
),
|
|
RecordingDB.calls,
|
|
)
|
|
|
|
def test_managed_file_queries_and_forms_fail_closed(self):
|
|
with self._client() as client:
|
|
responses = [
|
|
client.get(
|
|
ADMIN_PREFIX + '/files?relative_path=nested',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/files?root_id=exports&relative_path=',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/files?root_id=exports&root_id=readonly',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/files/download?root_id=exports',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/files/download?relative_path=guide.txt',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX
|
|
+ '/files/download?root_id=missing&relative_path=guide.txt',
|
|
headers=self._headers(),
|
|
),
|
|
client.post(
|
|
ADMIN_PREFIX + '/files/create',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': str(uuid.uuid4()),
|
|
'root_id': 'exports', 'relative_path': 'bad.bin',
|
|
'content_base64': 'not canonical base64',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
),
|
|
client.post(
|
|
ADMIN_PREFIX + '/files/delete',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': str(uuid.uuid4()),
|
|
'root_id': 'exports', 'relative_path': 'guide.txt',
|
|
'expected_sha256': 'A' * 64,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
),
|
|
]
|
|
self.assertEqual(
|
|
[response.status_code for response in responses],
|
|
[400, 400, 400, 400, 400, 404, 400, 400],
|
|
)
|
|
for response in responses:
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn('/data/managed-files', response.text)
|
|
self.assertEqual(self.managed_file_traversal.calls, [])
|
|
self.assertFalse(any(
|
|
call[0] == 'managed_file_operation_create'
|
|
for call in RecordingDB.calls
|
|
))
|
|
|
|
def test_managed_file_percent_encoding_and_encoded_traversal_fail_closed(self):
|
|
invalid_paths = (
|
|
'%2e%2e%2fprivate', '%2E%2e%2Fprivate', 'nested%5cprivate',
|
|
'bad%00name', '%43%3Aprivate', 'nested%2f%2e%2e%2fprivate',
|
|
)
|
|
responses = []
|
|
self.managed_file_traversal.calls.clear()
|
|
operation_calls_before = len([
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
])
|
|
with self._client() as client:
|
|
for encoded_path in invalid_paths:
|
|
responses.append(client.get(
|
|
ADMIN_PREFIX
|
|
+ '/files/download?root_id=exports&relative_path='
|
|
+ encoded_path,
|
|
headers=self._headers(),
|
|
))
|
|
body = (
|
|
urlencode({
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': str(uuid.uuid4()),
|
|
'root_id': 'exports',
|
|
'relative_path': 'PATH_MARKER',
|
|
'content_base64': 'eA==',
|
|
}).replace('PATH_MARKER', encoded_path).encode('ascii')
|
|
)
|
|
responses.append(client.post(
|
|
ADMIN_PREFIX + '/files/create', content=body,
|
|
headers=self._headers(**{
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
}), follow_redirects=False,
|
|
))
|
|
|
|
self.managed_file_traversal.files['%2e%2e%2fprivate'] = b'literal'
|
|
decoded_once = client.get(
|
|
ADMIN_PREFIX
|
|
+ '/files/download?root_id=exports&relative_path='
|
|
+ '%252e%252e%252fprivate',
|
|
headers=self._headers(),
|
|
)
|
|
encoded_routes = (
|
|
client.get(ADMIN_PREFIX + '/%66iles', headers=self._headers()),
|
|
client.get(
|
|
ADMIN_PREFIX + '/files%2Fdownload'
|
|
'?root_id=exports&relative_path=guide.txt',
|
|
headers=self._headers(),
|
|
),
|
|
client.get(
|
|
ADMIN_PREFIX + '/files%252Fdownload', headers=self._headers(),
|
|
),
|
|
)
|
|
|
|
self.assertTrue(all(response.status_code == 400 for response in responses))
|
|
self.assertEqual(decoded_once.status_code, 200, decoded_once.text)
|
|
self.assertEqual(decoded_once.content, b'literal')
|
|
self.assertIn(
|
|
('download', 'exports', '%2e%2e%2fprivate'),
|
|
self.managed_file_traversal.calls,
|
|
)
|
|
self.assertFalse(any(
|
|
call[0] in ('download', 'create_replace')
|
|
and call[2] == '../private'
|
|
for call in self.managed_file_traversal.calls
|
|
))
|
|
self.assertEqual([response.status_code for response in encoded_routes], [404] * 3)
|
|
for response in (*responses, decoded_once, *encoded_routes):
|
|
self.assertSecurityHeaders(response)
|
|
self.assertEqual(len([
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
]), operation_calls_before)
|
|
|
|
missing_raw_path = Request({
|
|
'type': 'http', 'method': 'GET',
|
|
'path': ADMIN_PREFIX + '/files', 'query_string': b'',
|
|
'headers': [], 'state': {},
|
|
})
|
|
with self.assertRaises(AdminAPIError) as raised:
|
|
asyncio.run(_dispatch(missing_raw_path, self.admin))
|
|
self.assertEqual(raised.exception.status_code, 404)
|
|
|
|
def test_generic_files_cannot_address_forbidden_roots_or_actions(self):
|
|
self.managed_file_traversal.calls.clear()
|
|
operation_calls_before = len([
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
])
|
|
responses = []
|
|
with self._client() as client:
|
|
for root_id in (
|
|
'config', 'secrets', 'postgres', 'runtime-document-candidates',
|
|
'/data/config'):
|
|
responses.append(client.get(
|
|
ADMIN_PREFIX + '/files?' + urlencode({'root_id': root_id}),
|
|
headers=self._headers(),
|
|
))
|
|
responses.extend((
|
|
client.get(
|
|
ADMIN_PREFIX + '/files?' + urlencode({
|
|
'root_id': 'exports', 'relative_path': '/data/config',
|
|
}), headers=self._headers(),
|
|
),
|
|
client.post(
|
|
ADMIN_PREFIX + '/files/action', data={}, headers=self._headers(),
|
|
),
|
|
))
|
|
for extra in ('action', 'command', 'path', 'service'):
|
|
responses.append(self._post(client, '/files/create', {
|
|
'operation_id': str(uuid.uuid4()), 'root_id': 'exports',
|
|
'relative_path': 'blocked.bin', 'content_base64': 'eA==',
|
|
extra: 'forbidden',
|
|
}))
|
|
|
|
self.assertEqual(
|
|
[response.status_code for response in responses],
|
|
[404, 404, 404, 404, 404, 400, 404, 400, 400, 400, 400],
|
|
)
|
|
for response in responses:
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn('/data/', response.text)
|
|
self.assertEqual(self.managed_file_traversal.calls, [])
|
|
self.assertEqual(len([
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
]), operation_calls_before)
|
|
|
|
def test_managed_file_body_limit_precedes_decode_audit_and_traversal(self):
|
|
bounded_admin = AdminService(
|
|
'postgresql://fixture', ORIGIN, MARKER, db_factory=RecordingDB,
|
|
max_body_bytes=1024, managed_file_roots=self.managed_file_registry,
|
|
)
|
|
exact_body = payload = None
|
|
operation_id = str(uuid.uuid4())
|
|
for payload_size in range(400, 700):
|
|
candidate_payload = b'x' * payload_size
|
|
encoded = __import__('base64').urlsafe_b64encode(
|
|
candidate_payload,
|
|
).decode('ascii')
|
|
for name_size in range(1, 256):
|
|
candidate = urlencode({
|
|
'csrf_token': bounded_admin.csrf_token,
|
|
'operation_id': operation_id,
|
|
'root_id': 'exports',
|
|
'relative_path': 'f' * name_size,
|
|
'content_base64': encoded,
|
|
}).encode('ascii')
|
|
if len(candidate) == 1024:
|
|
exact_body = candidate
|
|
payload = candidate_payload
|
|
break
|
|
if exact_body is not None:
|
|
break
|
|
self.assertIsNotNone(exact_body)
|
|
self.managed_file_traversal.calls.clear()
|
|
|
|
headers = {
|
|
EDGE_MARKER_HEADER: MARKER, OPERATOR_HEADER: OPERATOR,
|
|
'Origin': ORIGIN,
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
}
|
|
with TestClient(create_worker_app(
|
|
self.worker, reaper_interval_seconds=3600,
|
|
admin_service=bounded_admin,
|
|
)) as client:
|
|
exact = client.post(
|
|
ADMIN_PREFIX + '/files/create', content=exact_body,
|
|
headers=headers, follow_redirects=False,
|
|
)
|
|
calls_after_exact = tuple(self.managed_file_traversal.calls)
|
|
operations_after_exact = len([
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
])
|
|
over = client.post(
|
|
ADMIN_PREFIX + '/files/create', content=exact_body + b'x',
|
|
headers=headers, follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual(exact.status_code, 303, exact.text)
|
|
self.assertEqual(over.status_code, 413, over.text)
|
|
self.assertSecurityHeaders(exact)
|
|
self.assertSecurityHeaders(over)
|
|
self.assertEqual(tuple(self.managed_file_traversal.calls), calls_after_exact)
|
|
self.assertEqual(len([
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
]), operations_after_exact)
|
|
self.assertTrue(any(
|
|
call[0] == 'create_replace' and call[-1] == len(payload)
|
|
for call in calls_after_exact
|
|
))
|
|
|
|
chunks = iter((
|
|
{'type': 'http.request', 'body': b'x' * 700, 'more_body': True},
|
|
{'type': 'http.request', 'body': b'y' * 400, 'more_body': False},
|
|
))
|
|
|
|
async def receive():
|
|
return next(chunks)
|
|
|
|
scope = {
|
|
'type': 'http', 'method': 'POST',
|
|
'path': ADMIN_PREFIX + '/files/create',
|
|
'headers': [
|
|
(b'origin', ORIGIN.encode('ascii')),
|
|
(b'content-type', b'application/x-www-form-urlencoded'),
|
|
],
|
|
'state': {},
|
|
'app': SimpleNamespace(state=SimpleNamespace(
|
|
managed_file_traversal=self.managed_file_traversal,
|
|
)),
|
|
}
|
|
|
|
async def stream_over_limit():
|
|
request = Request(scope, receive)
|
|
request.state.admin_actor = OPERATOR
|
|
return await _dispatch_managed_file_mutation(
|
|
request, bounded_admin, 'create',
|
|
)
|
|
|
|
with self.assertRaises(AdminAPIError) as raised:
|
|
asyncio.run(stream_over_limit())
|
|
self.assertEqual(raised.exception.status_code, 413)
|
|
self.assertEqual(tuple(self.managed_file_traversal.calls), calls_after_exact)
|
|
self.assertEqual(len([
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'managed_file_operation_create'
|
|
]), operations_after_exact)
|
|
|
|
def test_random_prefix_navigation_csp_and_secret_page_are_storage_free(self):
|
|
operation_id = str(uuid.uuid4())
|
|
RecordingDB.apply_operations[operation_id] = {
|
|
'operation_id': operation_id, 'actor': OPERATOR,
|
|
'action': 'apply-config', 'target_kind': 'runtime-deployment',
|
|
'target_ref': 'config', 'status': 'requested',
|
|
'safe_category': None, 'safe_detail': None,
|
|
'expected_identity': {
|
|
'active_config_sha256': 'a' * 64,
|
|
'active_secrets_sha256': 'b' * 64,
|
|
'candidate_config_sha256': 'c' * 64,
|
|
'candidate_secrets_sha256': None,
|
|
},
|
|
'resulting_identity': None, 'requested_at': '2026-09-20T00:00:00+00:00',
|
|
'completed_at': None, 'updated_at': '2026-09-20T00:00:00+00:00',
|
|
}
|
|
paths = (
|
|
'', '/overview', '/search', '/supervisor', '/logs', '/config',
|
|
'/secrets', '/files', '/operations',
|
|
f'/operations/{operation_id}', '/audit',
|
|
)
|
|
with self._client() as client:
|
|
pages = {
|
|
path: client.get(ADMIN_PREFIX + path, headers=self._headers())
|
|
for path in paths
|
|
}
|
|
|
|
for path, response in pages.items():
|
|
with self.subTest(path=path):
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn('<script', response.text.lower())
|
|
self.assertNotRegex(response.text.lower(), r'\son[a-z]+\s*=')
|
|
self.assertNotRegex(
|
|
response.text,
|
|
r'localStorage|sessionStorage|indexedDB|serviceWorker|caches\.open',
|
|
)
|
|
self.assertNotIn(ADMIN_PREFIX, response.text)
|
|
public_path = '/private-prefix/' if not path else '/private-prefix' + path
|
|
base = 'https://admin.example.test' + public_path
|
|
targets = re.findall(r'(?:href|action)="([^"]+)"', response.text)
|
|
self.assertTrue(targets)
|
|
for target in targets:
|
|
self.assertTrue(
|
|
urljoin(base, target).startswith(
|
|
'https://admin.example.test/private-prefix/',
|
|
),
|
|
(path, target, urljoin(base, target)),
|
|
)
|
|
|
|
secrets = pages['/secrets'].text
|
|
self.assertEqual(secrets.count('editor-secret'), 1)
|
|
outside_textarea = re.sub(
|
|
r'(<textarea\b[^>]*>).*?(</textarea>)', r'\1\2', secrets,
|
|
flags=re.DOTALL | re.IGNORECASE,
|
|
)
|
|
self.assertNotIn('editor-secret', outside_textarea)
|
|
self.assertIn('<form method="post" autocomplete="off"', secrets)
|
|
self.assertIn('spellcheck="false" autocomplete="off"', secrets)
|
|
for path, response in pages.items():
|
|
if path != '/secrets':
|
|
self.assertNotIn('editor-secret', response.text)
|
|
|
|
def test_all_admin_routes_reject_unexpected_query_before_side_effects(self):
|
|
get_paths = (
|
|
'', '/', '/overview', '/search', '/supervisor', '/logs', '/config',
|
|
'/secrets', '/files', '/files/download', '/operations', '/audit',
|
|
'/admin.css',
|
|
)
|
|
with self._client() as client:
|
|
get_responses = [
|
|
client.get(
|
|
ADMIN_PREFIX + path + ('&' if '?' in path else '?') + 'unexpected=1',
|
|
headers=self._headers(),
|
|
)
|
|
for path in get_paths
|
|
]
|
|
post = client.post(
|
|
ADMIN_PREFIX + '/users/create?actor=mallory',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'user_key': 'query-injection',
|
|
'active_assignment_cap': '2',
|
|
'operation_id': '00000000-0000-4000-8000-000000000901',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
for response in (*get_responses, post):
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertSecurityHeaders(response)
|
|
self.assertFalse(any(
|
|
call[0] in {'worker_admin_create', 'create_user'}
|
|
for call in RecordingDB.calls
|
|
))
|
|
|
|
def test_every_post_route_rejects_wrong_method_origin_and_missing_fields(self):
|
|
post_paths = (
|
|
'/config/preview', '/config/save', '/config/apply',
|
|
'/secrets/preview', '/secrets/save', '/secrets/apply',
|
|
'/runtime/apply-both',
|
|
'/files/create', '/files/replace', '/files/delete',
|
|
'/dispatch/pause', '/dispatch/resume',
|
|
'/dispatch/drain/start', '/dispatch/drain/cancel',
|
|
'/search/discovery/pause', '/search/discovery/resume',
|
|
'/search/producers/start', '/search/producers/stop',
|
|
'/search/producers/restart', '/search/producers/pause',
|
|
'/search/producers/resume', '/search/producers/interval',
|
|
'/supervisor/sources/start', '/supervisor/sources/stop',
|
|
'/supervisor/sources/restart', '/supervisor/sources/pause',
|
|
'/supervisor/sources/resume', '/supervisor/sources/once',
|
|
'/supervisor/sources/mode/loop',
|
|
'/supervisor/sources/mode/once',
|
|
'/supervisor/sources/mode/repeat',
|
|
'/supervisor/sources/restart-policy/enable',
|
|
'/supervisor/sources/restart-policy/disable',
|
|
'/supervisor/sources/interval',
|
|
'/supervisor/sources/restart-delay',
|
|
'/supervisor/dashboard/start', '/supervisor/dashboard/stop',
|
|
'/supervisor/dashboard/restart', '/logs/tail',
|
|
'/users/create', '/users/cap', '/users/disable', '/users/enable',
|
|
'/devices/issue', '/devices/rotate',
|
|
'/devices/revoke', '/devices/unrevoke', '/queue/requeue',
|
|
)
|
|
responses = []
|
|
with self._client() as client:
|
|
for path in post_paths:
|
|
responses.extend((
|
|
client.get(ADMIN_PREFIX + path, headers=self._headers()),
|
|
client.post(
|
|
ADMIN_PREFIX + path,
|
|
data={'csrf_token': self.admin.csrf_token},
|
|
headers=self._headers(), follow_redirects=False,
|
|
),
|
|
client.post(
|
|
ADMIN_PREFIX + path,
|
|
data={'csrf_token': self.admin.csrf_token},
|
|
headers=self._headers(Origin='https://attacker.invalid'),
|
|
follow_redirects=False,
|
|
),
|
|
))
|
|
|
|
self.assertEqual(
|
|
[response.status_code for response in responses],
|
|
[status for _path in post_paths for status in (404, 400, 403)],
|
|
)
|
|
for response in responses:
|
|
self.assertSecurityHeaders(response)
|
|
mutation_calls = {
|
|
'dispatch_paused', 'drain_start', 'drain_cancel', 'discovery_paused',
|
|
'source_operation_create', 'worker_admin_create', 'create_user',
|
|
'set_cap', 'set_disabled', 'issue_device', 'set_revoked', 'requeue',
|
|
'document_operation_create', 'runtime_operation_create',
|
|
'managed_file_operation_create',
|
|
}
|
|
self.assertFalse(any(call[0] in mutation_calls for call in RecordingDB.calls))
|
|
self.source_action_provider.assert_not_called()
|
|
self.dashboard_action_provider.assert_not_called()
|
|
self.source_log_provider.assert_not_called()
|
|
self.candidate_preview_provider.assert_not_called()
|
|
self.candidate_save_provider.assert_not_called()
|
|
self.candidate_verify_provider.assert_not_called()
|
|
self.runtime_apply_provider.assert_not_called()
|
|
|
|
def test_routes_require_trusted_marker_and_operator_on_every_response(self):
|
|
with self._client(enabled=False) as client:
|
|
self.assertEqual(client.get(ADMIN_PREFIX, headers=self._headers()).status_code, 404)
|
|
with self._client() as client:
|
|
missing = client.get(ADMIN_PREFIX)
|
|
forged_without_marker = client.get(
|
|
ADMIN_PREFIX, headers={OPERATOR_HEADER: 'mallory'},
|
|
)
|
|
wrong = client.get(ADMIN_PREFIX + '/admin.css', headers={
|
|
EDGE_MARKER_HEADER: 'x' * 32, OPERATOR_HEADER: OPERATOR,
|
|
})
|
|
missing_operator = client.get(ADMIN_PREFIX, headers={EDGE_MARKER_HEADER: MARKER})
|
|
invalid_operator = client.get(ADMIN_PREFIX, headers={
|
|
EDGE_MARKER_HEADER: MARKER, OPERATOR_HEADER: 'spoofed operator',
|
|
})
|
|
duplicate_operator = client.get(ADMIN_PREFIX, headers=[
|
|
(EDGE_MARKER_HEADER, MARKER),
|
|
(OPERATOR_HEADER, OPERATOR),
|
|
(OPERATOR_HEADER, 'mallory'),
|
|
])
|
|
duplicate_marker = client.get(ADMIN_PREFIX, headers=[
|
|
(EDGE_MARKER_HEADER, MARKER),
|
|
(EDGE_MARKER_HEADER, 'x' * 32),
|
|
(OPERATOR_HEADER, OPERATOR),
|
|
])
|
|
page = client.get(ADMIN_PREFIX, headers=self._headers())
|
|
css = client.get(ADMIN_PREFIX + '/admin.css', headers=self._headers())
|
|
error = client.get(ADMIN_PREFIX + '/unknown', headers=self._headers())
|
|
method_error = client.put(ADMIN_PREFIX, headers=self._headers())
|
|
custom_method_errors = [
|
|
client.request(method, ADMIN_PREFIX + '/supervisor', headers=self._headers())
|
|
for method in ('PROPFIND', 'FROBULATE')
|
|
]
|
|
|
|
for response in (
|
|
missing, forged_without_marker, wrong, missing_operator, invalid_operator,
|
|
duplicate_operator, duplicate_marker,
|
|
page, css, error, method_error, *custom_method_errors,
|
|
):
|
|
self.assertSecurityHeaders(response)
|
|
self.assertEqual(
|
|
(missing.status_code, forged_without_marker.status_code, wrong.status_code,
|
|
missing_operator.status_code, invalid_operator.status_code,
|
|
duplicate_operator.status_code, duplicate_marker.status_code),
|
|
(404, 404, 404, 404, 404, 404, 404),
|
|
)
|
|
self.assertEqual((page.status_code, css.status_code, error.status_code), (200, 200, 404))
|
|
self.assertEqual(method_error.status_code, 405)
|
|
self.assertEqual([response.status_code for response in custom_method_errors], [405, 405])
|
|
self.assertEqual(css.headers['content-type'].split(';')[0], 'text/css')
|
|
self.assertIn('href="./admin.css"', page.text)
|
|
self.assertIn('href="./overview"', page.text)
|
|
self.assertIn('href="./search"', page.text)
|
|
self.assertIn('action="./users/create"', page.text)
|
|
self.assertNotIn(ADMIN_PREFIX, page.text)
|
|
self.assertIn(self.admin.csrf_token, page.text)
|
|
self.assertNotIn('<script', page.text.lower())
|
|
self.assertNotIn('<style', page.text.lower())
|
|
self.assertNotRegex(page.text, r'https?://')
|
|
|
|
def test_workers_dispatch_page_has_control_progress_packages_and_assignments(self):
|
|
with self._client() as client:
|
|
response = client.get(ADMIN_PREFIX, headers=self._headers())
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertIn('Workers / Dispatch', response.text)
|
|
self.assertIn('aria-current="page">Workers / Dispatch</a>', response.text)
|
|
self.assertIn('Dispatch and drain', response.text)
|
|
self.assertIn('Explicit pause', response.text)
|
|
self.assertIn('Effective pause', response.text)
|
|
self.assertIn('Drain blockers', response.text)
|
|
self.assertIn('value="7"', response.text)
|
|
for action in (
|
|
'./dispatch/pause', './dispatch/resume',
|
|
'./dispatch/drain/start', './dispatch/drain/cancel',
|
|
):
|
|
self.assertIn(f'action="{action}"', response.text)
|
|
self.assertEqual(response.text.count('name="operation_id"'), 14)
|
|
self.assertIn('Authenticated status, terminal reports, uploads', response.text)
|
|
self.assertIn('Protocol-1 workers are completion-only', response.text)
|
|
self.assertIn('linux-x86_64', response.text)
|
|
self.assertIn('exact_git_v1', response.text)
|
|
self.assertIn('docker_direct_v1', response.text)
|
|
self.assertIn('huggingface_space_v1', response.text)
|
|
self.assertIn('fixture-user', response.text)
|
|
self.assertIn('fixture-device', response.text)
|
|
self.assertIn('Accepted', response.text)
|
|
self.assertIn('Ingestion / projection', response.text)
|
|
snapshot_calls = [call for call in RecordingDB.calls if call[0] == 'snapshot']
|
|
self.assertEqual(len(snapshot_calls), 1)
|
|
self.assertEqual(snapshot_calls[0][1], 25)
|
|
self.assertFalse(any(
|
|
call[0] in {'diagnostic_groups', 'duration_metrics'}
|
|
for call in RecordingDB.calls
|
|
))
|
|
self.assertIn('value="assignments" selected', response.text)
|
|
self.assertIn('Not loaded. Select Assignments + diagnostics', response.text)
|
|
self.assertRegex(snapshot_calls[0][2]['since'], r'^\d{4}-\d{2}-\d{2}T')
|
|
self.assertIn(('drain_progress',), RecordingDB.calls)
|
|
self.package_compatibility_provider.assert_called_once_with()
|
|
self.assertNotIn('supervisor-token-must-not-render', response.text)
|
|
|
|
def test_worker_observability_list_separates_final_models_and_keeps_occurrences(self):
|
|
with self._client() as client:
|
|
response = client.get(
|
|
ADMIN_PREFIX + '?details=all', headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
for label in (
|
|
'Assignment outcome', 'Scan outcome', 'Diagnostics',
|
|
'Phase / progress', 'Deadlines', 'Slot / cap', 'Package identity',
|
|
'Ingestion / projection',
|
|
):
|
|
self.assertIn(label, response.text)
|
|
self.assertIn('active progress', response.text)
|
|
self.assertIn('<strong>Observability scope:</strong>', response.text)
|
|
self.assertEqual(response.text.count('<option value="all"'), 2)
|
|
self.assertIn(
|
|
'current phase: scanning; phase age 17s; progress age 3s; '
|
|
'age authority: current time',
|
|
response.text,
|
|
)
|
|
for reservation_id, assignment_outcome, scan_outcome in (
|
|
(4, 'unfinished', 'unavailable'),
|
|
(5, 'accepted', 'degraded'),
|
|
(6, 'prebundle_failed', 'unavailable'),
|
|
(7, 'expired', 'unavailable'),
|
|
(8, 'accepted', 'error'),
|
|
):
|
|
row = re.search(
|
|
rf'<tr data-reservation-id="{reservation_id}">(.*?)</tr>',
|
|
response.text,
|
|
)
|
|
self.assertIsNotNone(row)
|
|
self.assertIn(
|
|
f'<td data-field="assignment_outcome">{assignment_outcome}</td>',
|
|
row.group(1),
|
|
)
|
|
self.assertIn(
|
|
f'<td data-field="scan_outcome">{scan_outcome}',
|
|
row.group(1),
|
|
)
|
|
self.assertIn('latest persisted phase: awaiting_receipt', response.text)
|
|
self.assertIn('age authority: assignment resolution', response.text)
|
|
self.assertIn(
|
|
'persisted warning [detector_timeout]: bounded detector output timed out',
|
|
response.text,
|
|
)
|
|
self.assertIn('current protocol-2: 0 diagnostics observed', response.text)
|
|
self.assertIn('legacy/unavailable', response.text)
|
|
self.assertIn('unavailable (no persisted progress)', response.text)
|
|
self.assertIn(
|
|
'result upload: legacy/unavailable (not persisted)',
|
|
response.text,
|
|
)
|
|
self.assertIn(
|
|
'result upload: 1800s (persisted at assignment issuance)',
|
|
response.text,
|
|
)
|
|
self.assertNotIn('upload policy: 1800s', response.text)
|
|
self.assertIn(
|
|
'2 total filtered occurrences across 2 assignments', response.text,
|
|
)
|
|
self.assertEqual(response.text.count('assignment 5</a>'), 1)
|
|
self.assertEqual(response.text.count('assignment 6</a>'), 1)
|
|
self.assertIn(
|
|
'/assignments/5#diagnostic-' + ('b' * 64), response.text,
|
|
)
|
|
self.assertIn(
|
|
'/assignments/6#diagnostic-' + ('c' * 64), response.text,
|
|
)
|
|
self.assertIn('p50 seconds', response.text)
|
|
self.assertIn('21.000', response.text)
|
|
self.assertIn('2 (minimum 5)', response.text)
|
|
self.assertIn('Future assignments only.', response.text)
|
|
self.assertIn('90000 >= 900 + 1800 + 60', response.text)
|
|
self.assertIn('Matched occurrences: 201.', response.text)
|
|
self.assertIn('Fingerprint counts and affected-assignment counts cover the full filtered set.', response.text)
|
|
self.assertIn('Total source / phase / outcome groups: 201.', response.text)
|
|
self.assertIn('Next diagnostic occurrences', response.text)
|
|
self.assertIn('Next duration groups', response.text)
|
|
|
|
def test_worker_filters_are_independent_bounded_and_reach_all_read_models(self):
|
|
query = urlencode({
|
|
'source': 'gitlab', 'worker': 'fixture-device',
|
|
'assignment': 'accepted', 'scan': 'error', 'phase': 'scanning',
|
|
'category': 'rate_limit', 'code': 'provider.rate_limit',
|
|
'retryable': 'true', 'window': '7d', 'diagnostic_offset': '40',
|
|
'metric_offset': '60', 'limit': '50', 'details': 'all',
|
|
})
|
|
with self._client() as client:
|
|
response = client.get(
|
|
ADMIN_PREFIX + '?' + query, headers=self._headers(),
|
|
)
|
|
invalid = client.get(
|
|
ADMIN_PREFIX + '?retryable=sometimes', headers=self._headers(),
|
|
)
|
|
repeated = client.get(
|
|
ADMIN_PREFIX + '?source=gitlab&source=dockerhub',
|
|
headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertEqual((invalid.status_code, repeated.status_code), (400, 400))
|
|
calls = [call for call in RecordingDB.calls if call[0] in {
|
|
'snapshot', 'diagnostic_groups', 'duration_metrics',
|
|
}]
|
|
self.assertEqual(len(calls), 3)
|
|
for call in calls:
|
|
_name, limit, filters = call[:3]
|
|
self.assertEqual(limit, 50)
|
|
self.assertEqual(filters['source'], 'gitlab')
|
|
self.assertEqual(filters['worker'], 'fixture-device')
|
|
self.assertEqual(filters['assignment_outcome'], 'accepted')
|
|
self.assertEqual(filters['scan_outcome'], 'error')
|
|
self.assertEqual(filters['phase'], 'scanning')
|
|
self.assertEqual(filters['category'], 'rate_limit')
|
|
self.assertEqual(filters['code'], 'provider.rate_limit')
|
|
self.assertIs(filters['retryable'], True)
|
|
self.assertRegex(filters['since'], r'^\d{4}-\d{2}-\d{2}T')
|
|
self.assertIn('value="accepted" selected', response.text)
|
|
self.assertIn('value="7d" selected', response.text)
|
|
self.assertIn('value="50" selected', response.text)
|
|
self.assertIn('value="all" selected', response.text)
|
|
diagnostic_call = next(
|
|
call for call in calls if call[0] == 'diagnostic_groups'
|
|
)
|
|
self.assertEqual(diagnostic_call[3], 40)
|
|
metric_call = next(call for call in calls if call[0] == 'duration_metrics')
|
|
self.assertEqual(metric_call[3], 60)
|
|
self.assertIn('Previous diagnostic occurrences', response.text)
|
|
self.assertIn('Previous duration groups', response.text)
|
|
|
|
def test_worker_filter_accepts_the_full_trimmed_persisted_device_key_domain(self):
|
|
device_key = ' rack / [east] + #1 :: Ω '
|
|
with self._client() as client:
|
|
response = client.get(
|
|
ADMIN_PREFIX + '?' + urlencode({
|
|
'worker': device_key, 'window': 'all',
|
|
}),
|
|
headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
expected = device_key.strip()
|
|
for call in RecordingDB.calls:
|
|
if call[0] in {'snapshot', 'diagnostic_groups', 'duration_metrics'}:
|
|
self.assertEqual(call[2]['worker'], expected)
|
|
self.assertIn(f'value="{html.escape(expected)}"', response.text)
|
|
|
|
def test_assignment_detail_machine_output_material_and_download_are_exact(self):
|
|
diagnostic_uid = 'b' * 64
|
|
with self._client() as client:
|
|
detail = client.get(
|
|
ADMIN_PREFIX + '/assignments/5', headers=self._headers(),
|
|
)
|
|
machine = client.get(
|
|
ADMIN_PREFIX + '/assignments/5.json', headers=self._headers(),
|
|
)
|
|
download = client.get(
|
|
ADMIN_PREFIX + f'/assignments/5/diagnostics/{diagnostic_uid}.json',
|
|
headers=self._headers(),
|
|
)
|
|
script = client.get(
|
|
ADMIN_PREFIX + '/admin.js', headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(detail.status_code, 200, detail.text)
|
|
self.assertSecurityHeaders(detail)
|
|
for value in (
|
|
'Ordered timeline', 'Duration breakdown', 'bundle received',
|
|
'bundle ingested', 'queue settled', 'projection completed',
|
|
'Scan summary', 'Package identity', 'Current effective source policy',
|
|
'Copy canonical JSON', 'Download canonical JSON',
|
|
'truncated transformation', 'Progress authority',
|
|
'Current / latest phase', 'Last progress age seconds',
|
|
'Original bytes', 'Stored bytes', 'Original SHA-256',
|
|
):
|
|
self.assertIn(value, detail.text)
|
|
self.assertIn('[... 1 original bytes omitted', detail.text)
|
|
self.assertIn('1800s (persisted at assignment issuance)', detail.text)
|
|
self.assertIn('bounded detector output timed out', detail.text)
|
|
self.assertIn('src="../admin.js"', detail.text)
|
|
self.assertEqual(machine.status_code, 200, machine.text)
|
|
self.assertEqual(machine.headers['content-type'], 'application/json')
|
|
payload = machine.json()
|
|
self.assertEqual(payload['schema'], 1)
|
|
self.assertEqual(payload['assignment']['reservation_id'], 5)
|
|
self.assertEqual(payload['diagnostics']['items'][0]['diagnostic_uid'], diagnostic_uid)
|
|
body = payload['diagnostics']['items'][0]['diagnostic']['http']['body']
|
|
stderr = payload['diagnostics']['items'][0]['diagnostic']['process']['stderr']
|
|
self.assertEqual(
|
|
(body['original_size'], body['stored_size'], body['truncated']),
|
|
(MAX_DIAGNOSTIC_BODY_BYTES, MAX_DIAGNOSTIC_BODY_BYTES, False),
|
|
)
|
|
self.assertEqual(
|
|
(stderr['original_size'], stderr['stored_size'], stderr['truncated']),
|
|
(MAX_DIAGNOSTIC_LOG_BYTES + 1, MAX_DIAGNOSTIC_LOG_BYTES, True),
|
|
)
|
|
self.assertEqual(payload['deadlines']['upload_timeout_seconds'], 1800)
|
|
self.assertEqual(payload['progress']['current_phase'], 'awaiting_receipt')
|
|
self.assertEqual(payload['progress']['phase_age_seconds'], 120.0)
|
|
self.assertEqual(payload['progress']['last_progress_age_seconds'], 5.0)
|
|
self.assertEqual(payload['scan']['warning_class'], 'detector_timeout')
|
|
expected = json.dumps(
|
|
RecordingDB._diagnostic_fixture(5, diagnostic_uid), ensure_ascii=True,
|
|
sort_keys=True, separators=(',', ':'),
|
|
)
|
|
self.assertEqual(download.status_code, 200, download.text)
|
|
self.assertEqual(download.text, expected)
|
|
self.assertEqual(download.headers['etag'], '"' + ('6' * 64) + '"')
|
|
self.assertIn('attachment;', download.headers['content-disposition'])
|
|
self.assertEqual(script.status_code, 200)
|
|
self.assertIn('navigator.clipboard.writeText', script.text)
|
|
|
|
def test_assignment_details_have_distinct_active_accepted_prebundle_expiry_and_legacy_contracts(self):
|
|
with self._client() as client:
|
|
details = {
|
|
reservation_id: client.get(
|
|
ADMIN_PREFIX + f'/assignments/{reservation_id}.json',
|
|
headers=self._headers(),
|
|
).json()
|
|
for reservation_id in range(4, 9)
|
|
}
|
|
|
|
expected = {
|
|
4: ('unfinished', 'unavailable', 'scanning', False),
|
|
5: ('accepted', 'error', 'awaiting_receipt', True),
|
|
6: ('prebundle_failed', 'unavailable', 'bundling', True),
|
|
7: ('expired', 'unavailable', 'scanning', True),
|
|
8: ('accepted', 'error', None, True),
|
|
}
|
|
for reservation_id, contract in expected.items():
|
|
detail = details[reservation_id]
|
|
assignment_outcome, scan_outcome, phase, complete = contract
|
|
self.assertEqual(detail['assignment']['assignment_outcome'], assignment_outcome)
|
|
self.assertEqual(detail['assignment']['scan_outcome'], scan_outcome)
|
|
self.assertEqual(detail['progress']['current_phase'], phase)
|
|
self.assertEqual(detail['durations'][0]['complete'], complete)
|
|
self.assertEqual(details[4]['diagnostics']['availability'], 'current')
|
|
self.assertEqual(details[4]['diagnostics']['items'], [])
|
|
self.assertIn('bundle ingested', {
|
|
event['label'] for event in details[5]['timeline']
|
|
})
|
|
self.assertIn('prebundle_report', {
|
|
event['label'] for event in details[6]['timeline']
|
|
})
|
|
self.assertIn('expired', {
|
|
event['label'] for event in details[7]['timeline']
|
|
})
|
|
self.assertEqual(details[8]['diagnostics']['availability'], 'legacy/unavailable')
|
|
self.assertIsNone(details[8]['deadlines']['upload_timeout_seconds'])
|
|
self.assertEqual(
|
|
details[8]['deadlines']['upload_timeout_availability'],
|
|
'legacy/unavailable',
|
|
)
|
|
self.assertTrue(details[8]['legacy_evidence']['available'])
|
|
|
|
def test_assignment_detail_marks_legacy_evidence_without_inventing_envelope(self):
|
|
with self._client() as client:
|
|
response = client.get(
|
|
ADMIN_PREFIX + '/assignments/8', headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertIn('Availability: legacy/unavailable.', response.text)
|
|
self.assertIn('legacy evidence, not a synthesized diagnostic envelope', response.text)
|
|
self.assertIn('exact legacy raw error', response.text)
|
|
self.assertNotIn('Copy canonical JSON', response.text)
|
|
|
|
def test_config_editor_and_candidate_preview_show_effective_policy_and_metrics(self):
|
|
config_text = self.document_loader('/unused', 'config').text
|
|
with self._client() as client:
|
|
editor = client.get(
|
|
ADMIN_PREFIX + '/config', headers=self._headers(),
|
|
)
|
|
preview = self._post(
|
|
client, '/config/preview', self._document_fields(config_text),
|
|
)
|
|
|
|
self.assertEqual(editor.status_code, 200, editor.text)
|
|
self.assertEqual(preview.status_code, 200, preview.text)
|
|
self.assertIn('Candidate editor effective values', editor.text)
|
|
self.assertIn('Validated candidate effective values', preview.text)
|
|
for response in (editor, preview):
|
|
self.assertIn('Future assignments only.', response.text)
|
|
self.assertIn('source override', response.text)
|
|
self.assertIn('Observed source / phase / outcome durations', response.text)
|
|
self.assertIn('p95 seconds', response.text)
|
|
|
|
def test_dispatch_and_drain_mutations_use_trusted_actor_and_exact_revision(self):
|
|
operations = (
|
|
('/dispatch/pause', '00000000-0000-4000-8000-000000000201'),
|
|
('/dispatch/resume', '00000000-0000-4000-8000-000000000202'),
|
|
('/dispatch/drain/start', '00000000-0000-4000-8000-000000000203'),
|
|
('/dispatch/drain/cancel', '00000000-0000-4000-8000-000000000204'),
|
|
)
|
|
with self._client() as client:
|
|
responses = [
|
|
client.post(
|
|
ADMIN_PREFIX + path,
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'expected_revision': '7', 'operation_id': operation_id,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
for path, operation_id in operations
|
|
]
|
|
|
|
self.assertEqual(
|
|
[(response.status_code, response.headers.get('location')) for response in responses],
|
|
[(303, '../'), (303, '../'), (303, '../../'), (303, '../../')],
|
|
)
|
|
self.assertEqual([
|
|
call for call in RecordingDB.calls
|
|
if call[0] in {'dispatch_paused', 'drain_start', 'drain_cancel'}
|
|
], [
|
|
('dispatch_paused', True, 7, OPERATOR, operations[0][1]),
|
|
('dispatch_paused', False, 7, OPERATOR, operations[1][1]),
|
|
('drain_start', 7, OPERATOR, operations[2][1]),
|
|
('drain_cancel', 7, OPERATOR, operations[3][1]),
|
|
])
|
|
|
|
def test_dispatch_forms_reject_extra_fields_and_identity_conflicts(self):
|
|
with self._client() as client:
|
|
extra = client.post(
|
|
ADMIN_PREFIX + '/dispatch/pause',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'expected_revision': '7',
|
|
'operation_id': '00000000-0000-4000-8000-000000000205',
|
|
'actor': 'mallory',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
RecordingDB.control_identity_conflict = True
|
|
conflict = client.post(
|
|
ADMIN_PREFIX + '/dispatch/drain/start',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'expected_revision': '7',
|
|
'operation_id': '00000000-0000-4000-8000-000000000206',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual((extra.status_code, conflict.status_code), (400, 409))
|
|
self.assertFalse(any(call[0] == 'dispatch_paused' for call in RecordingDB.calls))
|
|
|
|
def test_stale_control_forms_return_conflict_without_transition(self):
|
|
RecordingDB.control_revision_conflict = True
|
|
operations = (
|
|
'/dispatch/pause', '/dispatch/resume',
|
|
'/dispatch/drain/start', '/dispatch/drain/cancel',
|
|
'/search/discovery/pause', '/search/discovery/resume',
|
|
)
|
|
with self._client() as client:
|
|
responses = [client.post(
|
|
ADMIN_PREFIX + path,
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'expected_revision': '6',
|
|
'operation_id': str(uuid.uuid4()),
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
) for path in operations]
|
|
|
|
self.assertEqual([response.status_code for response in responses], [409] * 6)
|
|
for response in responses:
|
|
self.assertSecurityHeaders(response)
|
|
self.assertNotIn('revision changed from', response.text)
|
|
self.assertFalse(any(
|
|
call[0] in {'dispatch_paused', 'drain_start', 'drain_cancel', 'discovery_paused'}
|
|
for call in RecordingDB.calls
|
|
))
|
|
|
|
def test_workers_page_degrades_package_component_independently(self):
|
|
self.package_compatibility_provider.side_effect = RuntimeError(
|
|
'manifest path secret must not render',
|
|
)
|
|
with self._client() as client:
|
|
response = client.get(ADMIN_PREFIX, headers=self._headers())
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertIn('Package compatibility unavailable.', response.text)
|
|
self.assertIn('fixture-user', response.text)
|
|
self.assertIn('Dispatch and drain', response.text)
|
|
self.assertNotIn('manifest path secret must not render', response.text)
|
|
|
|
def test_overview_uses_bounded_structured_components_and_shared_navigation(self):
|
|
with self._client() as client:
|
|
response = client.get(ADMIN_PREFIX + '/overview', headers=self._headers())
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertSecurityHeaders(response)
|
|
self.assertIn('Runtime overview', response.text)
|
|
self.assertIn('href="./admin.css"', response.text)
|
|
self.assertIn('href="./"', response.text)
|
|
self.assertIn('href="./overview"', response.text)
|
|
self.assertNotIn('href="../', response.text)
|
|
self.assertIn('aria-current="page">Overview</a>', response.text)
|
|
self.assertIn('Supervisor', response.text)
|
|
self.assertIn('PostgreSQL', response.text)
|
|
for source in ('gitlab', 'dockerhub', 'huggingface'):
|
|
self.assertIn(source, response.text)
|
|
for source in ('result-ingester', 'jsonl-projector', 'janitor', 'worker-api'):
|
|
self.assertIn(source, response.text)
|
|
self.assertIn('Pipeline authority', response.text)
|
|
self.assertIn('Ingester lease', response.text)
|
|
self.assertIn('Projector lease', response.text)
|
|
self.assertIn('Final cutover', response.text)
|
|
self.assertIn(
|
|
'<td>Projector lease</td><td>ready</td><td>True</td>',
|
|
response.text,
|
|
)
|
|
self.assertIn(
|
|
'<td>Final cutover</td><td>valid</td><td>True</td>',
|
|
response.text,
|
|
)
|
|
self.assertIn('Active assignments', response.text)
|
|
self.assertIn('Pre-commit bundles', response.text)
|
|
self.assertIn('Recent operation outcomes', response.text)
|
|
self.assertIn('succeeded', response.text)
|
|
self.assertNotIn('supervisor-token-must-not-render', response.text)
|
|
self.assertNotIn(self.admin.csrf_token, response.text)
|
|
self.runtime_provider.assert_called_once_with(
|
|
self.admin.supervisor_metadata, timeout=5,
|
|
)
|
|
self.assertIn(
|
|
('queue_counts', ('gitlab', 'dockerhub', 'huggingface')),
|
|
RecordingDB.calls,
|
|
)
|
|
self.assertIn(('drain_progress',), RecordingDB.calls)
|
|
self.assertIn(('recent_operations', 200, None, None), RecordingDB.calls)
|
|
|
|
def test_overview_degrades_unavailable_components_independently(self):
|
|
self.runtime_provider.side_effect = RuntimeError('runtime secret must not render')
|
|
RecordingDB.fail_queue = 'malformed'
|
|
with self._client() as client:
|
|
response = client.get(ADMIN_PREFIX + '/overview', headers=self._headers())
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertIn('Runtime snapshot unavailable.', response.text)
|
|
self.assertIn('Queue snapshot unavailable.', response.text)
|
|
self.assertIn('Recent operation outcomes', response.text)
|
|
self.assertIn('succeeded', response.text)
|
|
self.assertNotIn('runtime secret must not render', response.text)
|
|
self.assertNotIn('sensitive queue failure detail', response.text)
|
|
|
|
def test_overview_treats_uncached_degraded_queue_as_unavailable(self):
|
|
RecordingDB.fail_queue = 'degraded'
|
|
with self._client() as client:
|
|
response = client.get(ADMIN_PREFIX + '/overview', headers=self._headers())
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertIn('Queue snapshot unavailable.', response.text)
|
|
self.assertIn('Runtime health', response.text)
|
|
|
|
def test_overview_reuses_stale_queue_snapshot_during_retry_backoff(self):
|
|
with self._client() as client:
|
|
fresh = client.get(ADMIN_PREFIX + '/overview', headers=self._headers())
|
|
RecordingDB.fail_queue = 'degraded'
|
|
stale = client.get(ADMIN_PREFIX + '/overview', headers=self._headers())
|
|
queue_calls = RecordingDB.calls.count(
|
|
('queue_counts', ('gitlab', 'dockerhub', 'huggingface')),
|
|
)
|
|
backed_off = client.get(ADMIN_PREFIX + '/overview', headers=self._headers())
|
|
|
|
self.assertEqual((fresh.status_code, stale.status_code, backed_off.status_code), (200, 200, 200))
|
|
self.assertNotIn('Queue snapshot unavailable.', stale.text)
|
|
self.assertIn('stale=True', stale.text)
|
|
self.assertIn('pending', stale.text)
|
|
self.assertEqual(
|
|
RecordingDB.calls.count(
|
|
('queue_counts', ('gitlab', 'dockerhub', 'huggingface')),
|
|
),
|
|
queue_calls,
|
|
)
|
|
|
|
def test_search_page_has_persistent_and_exact_producer_controls(self):
|
|
with self._client() as client:
|
|
response = client.get(ADMIN_PREFIX + '/search', headers=self._headers())
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertSecurityHeaders(response)
|
|
self.assertIn('Search operations', response.text)
|
|
self.assertIn('aria-current="page">Search</a>', response.text)
|
|
self.assertIn('Persistent discovery control', response.text)
|
|
self.assertIn('Explicit pause', response.text)
|
|
self.assertIn('Effective pause', response.text)
|
|
self.assertIn('value="7"', response.text)
|
|
for source in ('gitlab', 'dockerhub', 'huggingface'):
|
|
source_id = f'discovery-producer:{source}'
|
|
self.assertIn(source_id, response.text)
|
|
self.assertIn(f'value="{source_id}"', response.text)
|
|
self.assertIn(f'Set {source} interval', response.text)
|
|
for action in ('start', 'stop', 'restart', 'pause', 'resume', 'interval'):
|
|
self.assertIn(f'action="./search/producers/{action}"', response.text)
|
|
self.assertEqual(
|
|
response.text.count('min="1" max="31536000" step="1"'), 3,
|
|
)
|
|
self.assertEqual(
|
|
response.text.count(
|
|
'name="interval_seconds" type="number" min="1" '
|
|
'max="31536000" step="1" value="3600"'
|
|
),
|
|
3,
|
|
)
|
|
self.assertNotIn('>Submit</button>', response.text)
|
|
self.assertIn('action="./search/discovery/pause"', response.text)
|
|
self.assertIn('action="./search/discovery/resume"', response.text)
|
|
self.assertEqual(response.text.count('name="operation_id"'), 20)
|
|
self.assertIn('Lifecycle and interval changes affect the current Supervisor runtime only.', response.text)
|
|
self.assertNotIn('supervisor-token-must-not-render', response.text)
|
|
|
|
def test_search_mutations_use_trusted_actor_exact_routes_and_durable_operations(self):
|
|
discovery_operation = '00000000-0000-4000-8000-000000000101'
|
|
restart_operation = '00000000-0000-4000-8000-000000000102'
|
|
interval_operation = '00000000-0000-4000-8000-000000000103'
|
|
with self._client() as client:
|
|
pause = client.post(
|
|
ADMIN_PREFIX + '/search/discovery/pause',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token, 'expected_revision': '7',
|
|
'operation_id': discovery_operation,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
restart = client.post(
|
|
ADMIN_PREFIX + '/search/producers/restart',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'discovery-producer:gitlab',
|
|
'operation_id': restart_operation,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
self.source_action_provider.return_value = {
|
|
'source_action': 'set-interval', 'outcome': 'completed', 'source': {},
|
|
}
|
|
interval = client.post(
|
|
ADMIN_PREFIX + '/search/producers/interval',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'discovery-producer:huggingface',
|
|
'interval_seconds': '7200',
|
|
'operation_id': interval_operation,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual(
|
|
[(pause.status_code, pause.headers.get('location')),
|
|
(restart.status_code, restart.headers.get('location')),
|
|
(interval.status_code, interval.headers.get('location'))],
|
|
[(303, '../../search'), (303, '../../search'), (303, '../../search')],
|
|
)
|
|
discovery = next(call for call in RecordingDB.calls if call[0] == 'discovery_paused')
|
|
self.assertEqual(discovery[1:], (True, 7, OPERATOR, discovery_operation))
|
|
creates = [call for call in RecordingDB.calls if call[0] == 'source_operation_create']
|
|
completes = [call for call in RecordingDB.calls if call[0] == 'source_operation_complete']
|
|
self.assertEqual([
|
|
(call[1], call[2], call[3], call[4], call[5]) for call in creates
|
|
], [
|
|
(restart_operation, OPERATOR, 'discovery-producer:gitlab', 'restart', None),
|
|
(interval_operation, OPERATOR, 'discovery-producer:huggingface', 'set-interval', 7200),
|
|
])
|
|
self.assertEqual(
|
|
[(call[1], call[2], call[3]) for call in completes],
|
|
[(creates[0][1], True, 'completed'), (creates[1][1], True, 'completed')],
|
|
)
|
|
self.assertEqual(self.source_action_provider.call_args_list, [
|
|
mock.call(
|
|
self.admin.supervisor_metadata, 'discovery-producer:gitlab',
|
|
'restart', timeout=60,
|
|
),
|
|
mock.call(
|
|
self.admin.supervisor_metadata, 'discovery-producer:huggingface',
|
|
'set-interval', timeout=60, interval_seconds=7200,
|
|
),
|
|
])
|
|
|
|
def test_search_producer_retry_reuses_operation_without_repeating_action(self):
|
|
operation_id = '00000000-0000-4000-8000-000000000104'
|
|
data = {
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'discovery-producer:dockerhub',
|
|
'operation_id': operation_id,
|
|
}
|
|
with self._client() as client:
|
|
first = client.post(
|
|
ADMIN_PREFIX + '/search/producers/restart', data=data,
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
replay = client.post(
|
|
ADMIN_PREFIX + '/search/producers/restart', data=data,
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual((first.status_code, replay.status_code), (303, 303))
|
|
self.source_action_provider.assert_called_once_with(
|
|
self.admin.supervisor_metadata, 'discovery-producer:dockerhub',
|
|
'restart', timeout=60,
|
|
)
|
|
creates = [
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'source_operation_create' and call[1] == operation_id
|
|
]
|
|
completes = [
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'source_operation_complete' and call[1] == operation_id
|
|
]
|
|
self.assertEqual(len(creates), 2)
|
|
self.assertEqual(len(completes), 1)
|
|
|
|
def test_search_producer_completion_retries_and_identity_conflict_is_409(self):
|
|
RecordingDB.completion_failures = 1
|
|
with self._client() as client:
|
|
completed = client.post(
|
|
ADMIN_PREFIX + '/search/producers/start', data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'discovery-producer:gitlab',
|
|
'operation_id': '00000000-0000-4000-8000-000000000108',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
RecordingDB.source_identity_conflict = True
|
|
conflict = client.post(
|
|
ADMIN_PREFIX + '/search/producers/start', data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'discovery-producer:gitlab',
|
|
'operation_id': '00000000-0000-4000-8000-000000000109',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual(completed.status_code, 303)
|
|
self.assertEqual(conflict.status_code, 409)
|
|
completions = [
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'source_operation_complete'
|
|
and call[1] == '00000000-0000-4000-8000-000000000108'
|
|
]
|
|
self.assertEqual(len(completions), 2)
|
|
self.source_action_provider.assert_called_once()
|
|
|
|
def test_search_forms_reject_extra_unknown_and_unbounded_values(self):
|
|
with self._client() as client:
|
|
extra = client.post(
|
|
ADMIN_PREFIX + '/search/discovery/pause',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'expected_revision': '7', 'actor': 'mallory',
|
|
'operation_id': '00000000-0000-4000-8000-000000000105',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
unknown = client.post(
|
|
ADMIN_PREFIX + '/search/producers/start',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'discovery-producer:github',
|
|
'operation_id': '00000000-0000-4000-8000-000000000106',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
unbounded = client.post(
|
|
ADMIN_PREFIX + '/search/producers/interval',
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'discovery-producer:gitlab',
|
|
'interval_seconds': str(365 * 24 * 60 * 60 + 1),
|
|
'operation_id': '00000000-0000-4000-8000-000000000107',
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
)
|
|
generic = self._post(client, '/search/producers/run', {
|
|
'source_id': 'discovery-producer:gitlab',
|
|
})
|
|
|
|
self.assertEqual(
|
|
(extra.status_code, unknown.status_code, unbounded.status_code, generic.status_code),
|
|
(400, 400, 400, 404),
|
|
)
|
|
self.assertFalse(any(
|
|
call[0] in {'discovery_paused', 'source_operation_create'}
|
|
for call in RecordingDB.calls
|
|
))
|
|
self.source_action_provider.assert_not_called()
|
|
|
|
def test_supervisor_and_log_pages_use_structured_allowlisted_controls(self):
|
|
with self._client() as client:
|
|
supervisor = client.get(ADMIN_PREFIX + '/supervisor', headers=self._headers())
|
|
logs = client.get(ADMIN_PREFIX + '/logs', headers=self._headers())
|
|
|
|
self.assertEqual((supervisor.status_code, logs.status_code), (200, 200))
|
|
self.assertIn('Structured managed-process state', supervisor.text)
|
|
self.assertIn('discovery-producer:gitlab', supervisor.text)
|
|
self.assertIn('result-ingester', supervisor.text)
|
|
self.assertIn('keychecks', supervisor.text)
|
|
self.assertIn('docker-shadow', supervisor.text)
|
|
self.assertIn('/supervisor/sources/restart', supervisor.text)
|
|
self.assertIn('/supervisor/sources/mode/repeat', supervisor.text)
|
|
self.assertIn('/supervisor/dashboard/restart', supervisor.text)
|
|
self.assertIn('<summary>Dashboard controls - running</summary>', supervisor.text)
|
|
self.assertIn(
|
|
'<summary>discovery-producer:gitlab - waiting - stopped</summary>',
|
|
supervisor.text,
|
|
)
|
|
self.assertEqual(
|
|
supervisor.text.count('<details class="control-panel">'),
|
|
len(self.runtime_provider.return_value['sources']) + 1,
|
|
)
|
|
self.assertNotIn('name="command"', supervisor.text)
|
|
self.assertNotIn('name="path"', supervisor.text)
|
|
self.assertNotIn('supervisor-token-must-not-render', supervisor.text + logs.text)
|
|
self.assertIn('Bounded allowlisted log tails', logs.text)
|
|
self.assertIn('max="5000"', logs.text)
|
|
self.assertIn('name="line_count" type="number" min="1" max="5000" '
|
|
'step="1" value="40"', logs.text)
|
|
self.assertIn('name="restart_delay_seconds" type="number" min="1" '
|
|
'max="31536000" step="1" value="5"', supervisor.text)
|
|
self.assertNotIn('>Submit</button>', supervisor.text + logs.text)
|
|
self.assertNotIn('name="path"', logs.text)
|
|
|
|
def test_supervisor_typed_actions_and_log_tail_are_attributed_and_escaped(self):
|
|
operations = (
|
|
('/supervisor/sources/mode/repeat', {
|
|
'source_id': 'keychecks',
|
|
'operation_id': '00000000-0000-4000-8000-000000000201',
|
|
}, '../../../supervisor'),
|
|
('/supervisor/sources/restart-delay', {
|
|
'source_id': 'result-ingester', 'restart_delay_seconds': '12',
|
|
'operation_id': '00000000-0000-4000-8000-000000000202',
|
|
}, '../../supervisor'),
|
|
('/supervisor/dashboard/restart', {
|
|
'operation_id': '00000000-0000-4000-8000-000000000203',
|
|
}, '../../supervisor'),
|
|
)
|
|
responses = []
|
|
with self._client() as client:
|
|
for route, fields, _redirect in operations:
|
|
responses.append(client.post(
|
|
ADMIN_PREFIX + route,
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(), follow_redirects=False,
|
|
))
|
|
tail = client.post(
|
|
ADMIN_PREFIX + '/logs/tail', data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'source_id': 'result-ingester', 'line_count': '2',
|
|
}, headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual(
|
|
[(response.status_code, response.headers.get('location')) for response in responses],
|
|
[(303, expected) for _route, _fields, expected in operations],
|
|
)
|
|
self.assertEqual(tail.status_code, 200)
|
|
self.assertIn('line <one>', tail.text)
|
|
self.assertNotIn('line <one>', tail.text)
|
|
creates = [call for call in RecordingDB.calls if call[0] == 'source_operation_create']
|
|
self.assertEqual(
|
|
[(call[2], call[3], call[4]) for call in creates],
|
|
[(OPERATOR, 'keychecks', 'set-mode'),
|
|
(OPERATOR, 'result-ingester', 'set-restart-delay'),
|
|
(OPERATOR, 'dashboard', 'restart')],
|
|
)
|
|
self.source_action_provider.assert_has_calls([
|
|
mock.call(
|
|
self.admin.supervisor_metadata, 'keychecks', 'set-mode',
|
|
timeout=60, mode='repeat',
|
|
),
|
|
mock.call(
|
|
self.admin.supervisor_metadata, 'result-ingester',
|
|
'set-restart-delay', timeout=60, restart_delay_seconds=12,
|
|
),
|
|
])
|
|
self.dashboard_action_provider.assert_called_once_with(
|
|
self.admin.supervisor_metadata, 'restart', timeout=60,
|
|
)
|
|
self.source_log_provider.assert_called_once_with(
|
|
self.admin.supervisor_metadata, 'result-ingester', 2, timeout=60,
|
|
)
|
|
|
|
def test_all_rendered_search_and_supervisor_action_routes_dispatch(self):
|
|
self.source_action_provider.side_effect = lambda *_args, **_kwargs: {
|
|
'outcome': 'completed', 'source': {},
|
|
}
|
|
self.dashboard_action_provider.side_effect = lambda *_args, **_kwargs: {
|
|
'outcome': 'completed', 'dashboard': {},
|
|
}
|
|
requests = [
|
|
('/search/discovery/resume', {'expected_revision': '7'}),
|
|
]
|
|
requests.extend(
|
|
(f'/search/producers/{action}', {
|
|
'source_id': 'discovery-producer:gitlab',
|
|
**({'interval_seconds': '60'} if action == 'interval' else {}),
|
|
})
|
|
for action in ('start', 'stop', 'restart', 'pause', 'resume', 'interval')
|
|
)
|
|
requests.extend(
|
|
(f'/supervisor/sources/{action}', {'source_id': 'keychecks'})
|
|
for action in ('start', 'stop', 'restart', 'pause', 'resume', 'once')
|
|
)
|
|
requests.extend((
|
|
('/supervisor/sources/mode/once', {'source_id': 'keychecks'}),
|
|
('/supervisor/sources/mode/repeat', {'source_id': 'keychecks'}),
|
|
('/supervisor/sources/restart-policy/enable', {'source_id': 'keychecks'}),
|
|
('/supervisor/sources/restart-policy/disable', {'source_id': 'keychecks'}),
|
|
('/supervisor/sources/interval', {
|
|
'source_id': 'keychecks', 'interval_seconds': '60',
|
|
}),
|
|
('/supervisor/sources/restart-delay', {
|
|
'source_id': 'keychecks', 'restart_delay_seconds': '5',
|
|
}),
|
|
))
|
|
requests.extend(
|
|
(f'/supervisor/dashboard/{action}', {})
|
|
for action in ('start', 'stop', 'restart')
|
|
)
|
|
|
|
responses = []
|
|
with self._client() as client:
|
|
for path, fields in requests:
|
|
responses.append(client.post(
|
|
ADMIN_PREFIX + path,
|
|
data={
|
|
'csrf_token': self.admin.csrf_token,
|
|
'operation_id': str(uuid.uuid4()),
|
|
**fields,
|
|
},
|
|
headers=self._headers(), follow_redirects=False,
|
|
))
|
|
|
|
self.assertEqual([response.status_code for response in responses], [303] * len(requests))
|
|
for response in responses:
|
|
self.assertSecurityHeaders(response)
|
|
creates = [
|
|
call for call in RecordingDB.calls if call[0] == 'source_operation_create'
|
|
]
|
|
self.assertEqual(len(creates), len(requests) - 1)
|
|
self.assertTrue(all(call[2] == OPERATOR for call in creates))
|
|
self.assertEqual(self.source_action_provider.call_count, 18)
|
|
self.assertEqual(self.dashboard_action_provider.call_count, 3)
|
|
|
|
def test_supervisor_forms_reject_unknown_generic_and_unbounded_inputs(self):
|
|
cases = (
|
|
('/supervisor/sources/restart', {
|
|
'source_id': 'unknown-source',
|
|
'operation_id': '00000000-0000-4000-8000-000000000211',
|
|
}),
|
|
('/supervisor/sources/restart', {
|
|
'source_id': '../result-ingester',
|
|
'operation_id': '00000000-0000-4000-8000-000000000212',
|
|
}),
|
|
('/supervisor/sources/restart-delay', {
|
|
'source_id': 'result-ingester', 'restart_delay_seconds': '31536001',
|
|
'operation_id': '00000000-0000-4000-8000-000000000213',
|
|
}),
|
|
('/logs/tail', {'source_id': 'result-ingester', 'line_count': '5001'}),
|
|
)
|
|
responses = []
|
|
with self._client() as client:
|
|
for route, fields in cases:
|
|
responses.append(client.post(
|
|
ADMIN_PREFIX + route,
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(), follow_redirects=False,
|
|
))
|
|
generic = client.post(
|
|
ADMIN_PREFIX + '/supervisor/run', data={
|
|
'csrf_token': self.admin.csrf_token, 'command': 'restart all',
|
|
}, headers=self._headers(), follow_redirects=False,
|
|
)
|
|
|
|
self.assertEqual([response.status_code for response in responses], [400, 400, 400, 400])
|
|
self.assertEqual(generic.status_code, 404)
|
|
self.assertFalse(any(
|
|
call[0] == 'source_operation_create' for call in RecordingDB.calls
|
|
))
|
|
self.source_action_provider.assert_not_called()
|
|
self.dashboard_action_provider.assert_not_called()
|
|
self.source_log_provider.assert_not_called()
|
|
|
|
def test_trusted_operator_is_request_scoped_and_preserves_authenticated_identity(self):
|
|
actors = []
|
|
|
|
async def capture_actor(request, service):
|
|
actors.append(request.state.admin_actor)
|
|
return Response('ok')
|
|
|
|
with mock.patch('admin_api._dispatch', side_effect=capture_actor):
|
|
with self._client() as client:
|
|
accepted = client.get(
|
|
ADMIN_PREFIX,
|
|
headers=self._headers(**{OPERATOR_HEADER: 'Alice.Operator'}),
|
|
)
|
|
rejected = [
|
|
client.get(
|
|
ADMIN_PREFIX,
|
|
headers=self._headers(**{OPERATOR_HEADER: value}),
|
|
)
|
|
for value in ('alice:admin', 'alice/admin', ' alice', 'a' * 65)
|
|
]
|
|
|
|
self.assertEqual(accepted.status_code, 200)
|
|
self.assertEqual(actors, ['Alice.Operator'])
|
|
self.assertTrue(all(response.status_code == 404 for response in rejected))
|
|
|
|
def test_non_ascii_marker_fails_closed(self):
|
|
request = Request({
|
|
'type': 'http',
|
|
'headers': [
|
|
(EDGE_MARKER_HEADER.encode('ascii'), b'\xff'),
|
|
(OPERATOR_HEADER.encode('ascii'), OPERATOR.encode('ascii')),
|
|
],
|
|
})
|
|
self.assertIsNone(_trusted_operator(request, self.admin))
|
|
|
|
def test_mutation_response_keeps_assets_and_forms_relative_to_admin_root(self):
|
|
with self._client() as client:
|
|
response = self._post(client, '/users/create', {
|
|
'user_key': 'alice', 'active_assignment_cap': '2',
|
|
})
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertEqual(
|
|
[(item.status_code, item.headers.get('location')) for item in response.history],
|
|
[(303, '../')],
|
|
)
|
|
self.assertIn('href="./admin.css"', response.text)
|
|
self.assertIn('action="./users/create"', response.text)
|
|
self.assertIn('action="./devices/issue"', response.text)
|
|
|
|
def test_configured_origin_must_be_an_exact_https_origin(self):
|
|
for origin in (
|
|
'http://admin.example.test',
|
|
'https://admin.example.test/',
|
|
'https://user:password@admin.example.test',
|
|
'https://admin.example.test/path',
|
|
):
|
|
with self.subTest(origin=origin), self.assertRaisesRegex(ValueError, 'HTTPS origin'):
|
|
AdminService(
|
|
'postgresql://fixture', origin, MARKER, db_factory=RecordingDB,
|
|
)
|
|
|
|
def test_post_requires_exact_origin_and_constant_time_csrf_value(self):
|
|
fields = {
|
|
'user_key': 'alice', 'active_assignment_cap': '2',
|
|
'operation_id': '00000000-0000-4000-8000-000000000301',
|
|
}
|
|
with self._client() as client:
|
|
missing_origin = client.post(
|
|
ADMIN_PREFIX + '/users/create',
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers={EDGE_MARKER_HEADER: MARKER, OPERATOR_HEADER: OPERATOR},
|
|
)
|
|
wrong_origin = client.post(
|
|
ADMIN_PREFIX + '/users/create',
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(Origin=ORIGIN + '/'),
|
|
)
|
|
wrong_csrf = client.post(
|
|
ADMIN_PREFIX + '/users/create',
|
|
data={'csrf_token': self.admin.csrf_token + 'x', **fields},
|
|
headers=self._headers(),
|
|
)
|
|
missing_csrf = client.post(
|
|
ADMIN_PREFIX + '/users/create', data=fields,
|
|
headers=self._headers(),
|
|
)
|
|
duplicate_csrf = client.post(
|
|
ADMIN_PREFIX + '/users/create',
|
|
content=urlencode([
|
|
('csrf_token', self.admin.csrf_token),
|
|
('csrf_token', self.admin.csrf_token),
|
|
*fields.items(),
|
|
]).encode('ascii'),
|
|
headers=self._headers(**{
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
}),
|
|
)
|
|
duplicate_origin = client.post(
|
|
ADMIN_PREFIX + '/users/create',
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=[
|
|
(EDGE_MARKER_HEADER, MARKER), (OPERATOR_HEADER, OPERATOR),
|
|
('Origin', ORIGIN), ('Origin', ORIGIN),
|
|
],
|
|
)
|
|
valid = self._post(client, '/users/create', fields)
|
|
|
|
self.assertEqual(
|
|
(
|
|
missing_origin.status_code, wrong_origin.status_code,
|
|
wrong_csrf.status_code, missing_csrf.status_code,
|
|
duplicate_csrf.status_code, duplicate_origin.status_code,
|
|
),
|
|
(403, 403, 403, 400, 400, 403),
|
|
)
|
|
for response in (
|
|
missing_origin, wrong_origin, wrong_csrf, missing_csrf,
|
|
duplicate_csrf, duplicate_origin,
|
|
):
|
|
self.assertSecurityHeaders(response)
|
|
self.assertEqual(valid.status_code, 200, valid.text)
|
|
self.assertEqual(
|
|
[call for call in RecordingDB.calls if call[0] == 'create_user'],
|
|
[('create_user', 'alice', 2)],
|
|
)
|
|
|
|
def test_urlencoded_body_is_bounded_unique_and_exactly_shaped(self):
|
|
with self._client() as client:
|
|
wrong_type = client.post(
|
|
ADMIN_PREFIX + '/users/create', content=b'{}',
|
|
headers=self._headers(**{'Content-Type': 'application/json'}),
|
|
)
|
|
extra = self._post(client, '/users/create', {
|
|
'user_key': 'alice', 'active_assignment_cap': '2', 'unexpected': 'x',
|
|
})
|
|
duplicate = client.post(
|
|
ADMIN_PREFIX + '/users/create',
|
|
content=(
|
|
f'csrf_token={self.admin.csrf_token}&user_key=alice&user_key=bob'
|
|
'&active_assignment_cap=2'
|
|
).encode('ascii'),
|
|
headers=self._headers(**{
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
}),
|
|
)
|
|
|
|
bounded_admin = AdminService(
|
|
'postgresql://fixture', ORIGIN, MARKER, db_factory=RecordingDB,
|
|
max_body_bytes=1024,
|
|
)
|
|
with TestClient(create_worker_app(
|
|
self.worker, reaper_interval_seconds=3600, admin_service=bounded_admin,
|
|
)) as client:
|
|
oversized = client.post(
|
|
ADMIN_PREFIX + '/users/disable', content=b'x' * 1025,
|
|
headers={
|
|
EDGE_MARKER_HEADER: MARKER, OPERATOR_HEADER: OPERATOR,
|
|
'Origin': ORIGIN,
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
)
|
|
|
|
self.assertEqual(wrong_type.status_code, 415)
|
|
self.assertEqual((extra.status_code, duplicate.status_code), (400, 400))
|
|
self.assertEqual(oversized.status_code, 413)
|
|
for response in (wrong_type, extra, duplicate, oversized):
|
|
self.assertSecurityHeaders(response)
|
|
|
|
def test_only_typed_mutations_are_dispatched(self):
|
|
operations = (
|
|
('/users/create', {'user_key': 'alice', 'active_assignment_cap': '2'}),
|
|
('/users/cap', {'user_key': 'alice', 'active_assignment_cap': '3'}),
|
|
('/users/disable', {'user_key': 'alice'}),
|
|
('/users/enable', {'user_key': 'alice'}),
|
|
('/devices/issue', {'user_key': 'alice', 'device_key': 'laptop'}),
|
|
('/devices/rotate', {'user_key': 'alice', 'device_key': 'laptop'}),
|
|
('/devices/revoke', {'device_key': 'laptop'}),
|
|
('/devices/unrevoke', {'device_key': 'laptop'}),
|
|
('/queue/requeue', {'queue_ids': '11,12'}),
|
|
('/queue/discard-source', {
|
|
'source': 'gitlab', 'confirm_source': 'gitlab',
|
|
}),
|
|
)
|
|
with self._client() as client:
|
|
responses = [self._post(client, path, fields) for path, fields in operations]
|
|
generic = self._post(client, '/supervisor/run', {})
|
|
|
|
self.assertTrue(all(response.status_code == 200 for response in responses))
|
|
self.assertEqual(generic.status_code, 404)
|
|
business_names = {
|
|
'create_user', 'set_cap', 'set_disabled', 'issue_device',
|
|
'set_revoked', 'requeue', 'discard_source_queue',
|
|
}
|
|
mutations = [call for call in RecordingDB.calls if call[0] in business_names]
|
|
self.assertEqual([call[0] for call in mutations], [
|
|
'create_user', 'set_cap', 'set_disabled', 'set_disabled',
|
|
'issue_device', 'issue_device', 'set_revoked', 'set_revoked', 'requeue',
|
|
'discard_source_queue',
|
|
])
|
|
self.assertEqual(mutations[2:], [
|
|
('set_disabled', 'alice', True),
|
|
('set_disabled', 'alice', False),
|
|
mutations[4], mutations[5],
|
|
('set_revoked', 'laptop', True),
|
|
('set_revoked', 'laptop', False),
|
|
('requeue', [11, 12], 100),
|
|
('discard_source_queue', 'gitlab'),
|
|
])
|
|
self.assertFalse(mutations[4][-1])
|
|
self.assertTrue(mutations[5][-1])
|
|
accepted = [call for call in RecordingDB.calls if call[0] == 'worker_admin_create']
|
|
completed = [call for call in RecordingDB.calls if call[0] == 'worker_admin_complete']
|
|
self.assertEqual(len(accepted), len(operations))
|
|
self.assertEqual(len(completed), len(operations))
|
|
self.assertTrue(all(call[2] == OPERATOR for call in accepted))
|
|
self.assertTrue(all(re.fullmatch(r'[a-f0-9]{64}', call[5]) for call in accepted))
|
|
|
|
def test_worker_admin_post_replay_is_attributed_and_does_not_repeat_mutation(self):
|
|
operation_id = '00000000-0000-4000-8000-000000000401'
|
|
fields = {
|
|
'csrf_token': self.admin.csrf_token,
|
|
'user_key': 'replay-user',
|
|
'active_assignment_cap': '2',
|
|
'operation_id': operation_id,
|
|
}
|
|
with self._client() as client:
|
|
first = client.post(
|
|
ADMIN_PREFIX + '/users/create', data=fields,
|
|
headers=self._headers(),
|
|
)
|
|
replay = client.post(
|
|
ADMIN_PREFIX + '/users/create', data=fields,
|
|
headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(first.status_code, 200)
|
|
self.assertEqual(replay.status_code, 409)
|
|
self.assertEqual(
|
|
[call for call in RecordingDB.calls if call[0] == 'create_user'],
|
|
[('create_user', 'replay-user', 2)],
|
|
)
|
|
accepted = [
|
|
call for call in RecordingDB.calls
|
|
if call[0] == 'worker_admin_create' and call[1] == operation_id
|
|
]
|
|
self.assertEqual(len(accepted), 2)
|
|
self.assertTrue(all(call[2] == OPERATOR for call in accepted))
|
|
|
|
def test_non_token_worker_admin_mutations_use_relative_post_redirect_get(self):
|
|
requests = (
|
|
('/users/create', {
|
|
'user_key': 'redirect-user', 'active_assignment_cap': '2',
|
|
'operation_id': '00000000-0000-4000-8000-000000000411',
|
|
}),
|
|
('/devices/revoke', {
|
|
'device_key': 'redirect-device',
|
|
'operation_id': '00000000-0000-4000-8000-000000000412',
|
|
}),
|
|
('/queue/requeue', {
|
|
'queue_ids': '11,12',
|
|
'operation_id': '00000000-0000-4000-8000-000000000413',
|
|
}),
|
|
('/queue/discard-source', {
|
|
'source': 'huggingface', 'confirm_source': 'huggingface',
|
|
'operation_id': '00000000-0000-4000-8000-000000000414',
|
|
}),
|
|
)
|
|
with self._client() as client:
|
|
responses = [client.post(
|
|
ADMIN_PREFIX + path,
|
|
data={'csrf_token': self.admin.csrf_token, **fields},
|
|
headers=self._headers(), follow_redirects=False,
|
|
) for path, fields in requests]
|
|
|
|
self.assertEqual(
|
|
[(response.status_code, response.headers.get('location')) for response in responses],
|
|
[(303, '../'), (303, '../'), (303, '../'), (303, '../')],
|
|
)
|
|
|
|
def test_source_queue_discard_requires_exact_confirmation(self):
|
|
with self._client() as client:
|
|
mismatch = self._post(client, '/queue/discard-source', {
|
|
'source': 'gitlab', 'confirm_source': 'dockerhub',
|
|
})
|
|
unmanaged = self._post(client, '/queue/discard-source', {
|
|
'source': 'github', 'confirm_source': 'github',
|
|
})
|
|
|
|
self.assertEqual((mismatch.status_code, unmanaged.status_code), (400, 400))
|
|
self.assertNotIn(
|
|
'discard_source_queue', [call[0] for call in RecordingDB.calls],
|
|
)
|
|
|
|
def test_plaintext_device_token_is_returned_once_and_only_digest_reaches_db(self):
|
|
with self._client() as client:
|
|
issued = self._post(client, '/devices/issue', {
|
|
'user_key': 'alice', 'device_key': 'laptop',
|
|
})
|
|
later = client.get(ADMIN_PREFIX, headers=self._headers())
|
|
|
|
self.assertEqual(issued.status_code, 200, issued.text)
|
|
token = re.search(r'<code>([^<]+)</code>', issued.text).group(1)
|
|
self.assertGreaterEqual(len(token), 64)
|
|
device_call = next(call for call in RecordingDB.calls if call[0] == 'issue_device')
|
|
self.assertEqual(device_call[3], hashlib.sha256(token.encode('ascii')).hexdigest())
|
|
self.assertNotIn(token, repr(RecordingDB.calls))
|
|
self.assertNotIn(token, later.text)
|
|
|
|
def test_worker_401_has_bearer_challenge(self):
|
|
with self._client(enabled=False) as client:
|
|
response = client.post('/api/v1/worker/claim', json={})
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(response.headers['www-authenticate'], 'Bearer')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|