truf-server/tests/container_unit.py

"""Explicit offline regressions for the Docker test target.

Run as UID 10001 with read-only /opt/truf/app, writable /tmp tmpfs, and
Docker --network none: python -I -S -B /opt/truf/tests/container_unit.py.
No /data mount, runtime configuration, credentials, PostgreSQL server, external
Git source, or provider entrypoint is needed. SQLite fixtures stay in /tmp;
native OwnedProcess children and loopback control sockets are intentional.

--list and --check-selection inspect source with the stdlib only, including
on the host. -k only narrows the reviewed selection; pytest arguments and
additional paths/plugins are not accepted. New test files are never discovered.
"""

import sys

sys.dont_write_bytecode = True

import argparse
import ast
from contextlib import ExitStack
import os
from pathlib import Path
import runpy
import stat
import tempfile
from unittest import mock


TESTS = Path(__file__).resolve().parent
ROOT = TESTS.parent
APP = ROOT / 'app'

SELECTION = {
    'test_db_backend_safety.py': (
        'PostgresTimeoutTests',
    ),
    'test_docker_foundation.py': (
        'PortablePathTests',
        'DockerStagingTests::test_native_checkout_default_is_derived_from_this_copy',
        'DockerStagingTests::test_python_entrypoints_refuse_before_application_imports',
        'DockerStagingTests::test_all_copied_powershell_tools_have_static_refusals',
        'DockerStagingTests::test_docker_context_exceptions_are_explicit_source_files_only',
    ),
    # Linux metadata and runtime I/O are mocked; filesystem fixtures stay in tmp_path.
    'test_container_runtime.py': (
        'test_private_path_accepts_exact_owner_only_mode',
        'test_private_path_rejects_nonabsolute_or_parent_traversal_before_stat',
        'test_private_path_rejects_symlink_at_every_component',
        'test_private_path_fails_closed_on_missing_or_uninspectable_components',
        'test_private_path_rejects_wrong_owner_group_type_or_exact_mode',
        'test_container_requires_private_image_native_data_and_dedicated_tmpfs',
        'test_container_refuses_before_mount_inventory_when_identity_gate_fails',
        'test_root_is_permitted_only_for_explicit_provisioning',
        'test_container_rejects_nonprivate_required_paths',
        'test_container_rejects_missing_shared_or_host_style_storage',
        'test_new_private_file_is_exclusive_nofollow_and_durable',
        'test_new_private_file_never_overwrites_existing_file',
        'test_fresh_provision_generates_private_password_once_without_initializing_pg',
        'test_provision_refuses_nonempty_or_partial_volume_without_repair',
        'test_provision_lock_conflict_never_creates_layout_or_credentials',
        'test_provision_does_not_regenerate_incomplete_marked_volume',
        'test_provision_adds_only_managed_files_to_legacy_marked_volume',
        'test_provision_refuses_wrong_existing_managed_files_path',
        'test_bad_provision_marker_prevents_environment_or_application_imports',
        'test_prepare_environment_scrubs_dsn_and_runtime_overrides_before_imports',
        'test_prepare_rejects_invalid_generated_password_before_imports',
        'test_prepare_refuses_config_escaping_fixed_storage_contract',
        'test_only_default_image_config_or_private_data_config_is_accepted',
        'test_bootstrap_target_is_the_real_runtime_safety_migration_module',
        'test_initialize_publishes_marker_only_after_migration_and_confirmed_stop',
        'test_failed_initialization_always_stops_maintenance_and_never_marks_success',
        'test_shutdown_during_initialization_never_publishes_early_marker',
        'test_partial_initialization_is_not_adopted_or_repaired',
        'test_initialized_marker_must_match_bound_cluster_identity',
        'test_health_does_not_import_database_or_control_before_valid_marker',
        'test_health_requires_active_supervisor_and_explicit_postgres_ready',
        'test_health_rejects_missing_or_unready_pipeline_processes_before_db',
        'test_health_requires_each_enabled_pipeline_worker',
        'test_health_allows_explicitly_disabled_janitor',
        'test_health_requires_worker_api_only_when_explicitly_enabled',
        'test_strict_health_requires_enabled_worker_api_before_database',
        'test_strict_health_probes_exact_unauthorized_worker_endpoint',
        'test_strict_worker_probe_fails_closed_before_database',
        'test_health_rejects_failed_or_uncertain_managed_source',
        'test_health_checks_readonly_schema_cutover_leases_identity_then_storage',
        'test_health_fails_closed_and_closes_database_on_each_database_gate',
        'test_health_requires_writable_nonfull_persistent_storage',
        'test_secret_import_refuses_active_runtime_before_reading_stdin',
        'test_secret_import_cluster_lock_conflict_precedes_stopped_check_and_stdin',
        'test_secret_import_is_locked_atomic_private_and_never_prints_credentials',
        'test_invalid_secret_input_is_bounded_and_does_not_leak_or_replace',
        'test_secret_import_validation_error_clears_candidate_traceback_locals',
        'test_secret_import_rejects_config_hash_drift_before_temporary_write',
        'test_secret_import_rechecks_config_hash_before_replacement',
        'test_write_new_clears_secret_payload_from_traceback_locals',
        'test_initialize_rejects_config_hash_drift_before_mutation',
        'test_secret_import_error_removes_temporary_file_and_preserves_old_credentials',
        'test_main_container_gate_precedes_every_action',
        'test_main_provision_does_not_enter_nonroot_environment_or_database',
        'test_main_import_secrets_uses_preliminary_environment_validation',
        'test_main_initialize_binds_validated_config_hash',
        'test_main_run_rechecks_config_hash_before_exec',
        'test_main_snapshot_requires_approved_digest_and_default_config',
        'test_main_snapshot_dispatch_preserves_module_signal_state_without_initialize',
        'test_manifest_digest_is_rejected_for_other_actions',
        'test_worker_api_requirement_flag_is_restricted_to_health',
        'test_strict_status_dispatches_worker_api_requirement',
        'test_status_stdout_contains_readiness_but_no_credentials',
    ),
    'test_runtime_document.py': (
        'RuntimeDocumentTests',
    ),
    'test_managed_files.py': (
        'ManagedFileConfigurationTests',
        'ManagedFileTraversalTests',
        'ManagedFileTraversalLinuxTests',
    ),
    'test_operations_schema.py': ('OperationsSchemaTests',),
    'test_operations_control.py': ('OperationsControlTests',),
    'test_host_agent_protocol.py': (
        'HostAgentProtocolTests',
        'HostAgentClientTests',
        'HostAgentServerTests',
    ),
    'test_host_agent_linux.py': (
        'HostAgentLinuxTests',
    ),
    'test_host_agent_apply.py': (
        'HostAgentApplyTests',
    ),
    'test_host_agent_deploy.py': (
        'HostAgentDeployTests',
        'HostAgentInstallerTests',
    ),
    'test_host_agent_lifecycle.py': (
        'HostAgentLifecycleTests',
    ),
    'test_host_agent_reconcile.py': (
        'HostAgentReconcileTests',
    ),
    'test_host_agent_runtime.py': (
        'HostAgentRuntimeTests',
    ),
    'test_host_agent_state.py': (
        'HostAgentStateTests',
    ),
    'test_runtime_document_io.py': (
        'RuntimeDocumentIOTests',
    ),
    'test_operations_service.py': (
        'OperationsServiceTests',
    ),
    'test_owned_process.py': (
        'OwnedProcessTests',
        'WindowsOwnedProcessTests',
        'StaticProcessSafetyTests',
    ),
    'test_owned_process_linux.py': (
        'LinuxIdentityTests',
        'StartupHandshakeTests',
        'LinuxContainmentTests',
        'LinuxOwnedProcessIntegrationTests',
    ),
    'test_owned_process_boundary.py': (
        'OwnedProcessHostBoundaryTests',
        'CredentialBoundaryTests',
    ),
    'test_runtime_bootstrap_authority.py': (
        'RuntimeBootstrapAuthorityTests::test_authenticated_docker_shadow_dispatches_exact_entrypoint',
        'RuntimeBootstrapAuthorityTests::test_provider_bootstrap_consumes_one_required_separator_before_provider_parse',
        'RuntimeBootstrapAuthorityTests::test_provider_bootstrap_rejects_missing_and_duplicate_separators',
        'RuntimeBootstrapAuthorityTests::test_authenticated_entrypoint_exception_is_classified_as_runtime_failure',
        'RuntimeBootstrapAuthorityTests::test_command_builders_use_exact_isolation_flag_order',
        'RuntimeBootstrapAuthorityTests::test_direct_mutating_supervisor_fails_before_config_env_locks_or_children',
    ),
    'test_supervisor_foreground_shutdown.py': (
        'test_signal_callback_is_only_an_idempotent_flag_assignment',
        'test_receipt_parity_and_finalization_order',
        'test_windows_does_not_register_signals',
        'test_mutating_launch_gates_precede_config_locks_children_and_signals',
        'test_non_owning_dispatch_does_not_register_signals',
        'test_pending_term_cannot_admit_startup',
        'test_background_activation_wait_consumes_term',
        'test_admission_rejects_pending_term_before_checkpoint',
        'test_control_activation_cannot_reopen_pending_shutdown',
        'test_background_activation_callback_failure_is_sticky',
        'test_term_between_pipeline_starts_admits_no_more_children',
        'test_checkpoints_exit_without_polling_or_starting',
        'test_term_between_source_polls_rejects_next_poll',
        'test_foreground_waits_consume_term_without_input_or_next_refresh',
        'test_metadata_publication_failure_leaves_closed_memory_gates',
        'test_repeated_term_checkpoint_does_not_republish_or_rearm_retry',
        'test_repeated_shutdown_does_not_discard_confirmed_stop_proof',
        'test_initial_cleanup_interruption_retains_authority_and_failure',
        'test_interrupted_cleanup_lock_acquisition_retains_authority',
        'test_partially_published_activation_uses_full_cleanup',
        'test_coordinated_shutdown_interruption_closes_gate_before_propagating',
        'test_failed_hold_survives_interruption_and_logging_failure',
        'test_failed_hold_does_not_release_with_unconfirmed_children',
        'test_receipt_write_failure_is_nonzero',
        'test_late_cleanup_failures_are_in_receipt',
        'test_failure_latched_by_draining_control_worker_is_in_receipt',
        'test_terminal_foreground_failure_is_sticky_without_changing_restart_policy',
        'test_foreground_source_failure_survives_cleanup_status_reset',
        'test_foreground_missing_autostart_is_nonzero',
        'test_system_exit_status_is_preserved',
        'test_interrupt_message_failure_cannot_publish_success',
        'test_authority_drift_failure_remains_sticky',
        'test_preactivation_close_failure_is_sticky',
        'test_noninteractive_keeps_existing_completion_policy',
        'test_stopper_requires_posix_receipt_and_preserves_windows_crosscheck',
        'test_unconfirmed_child_defers_all_postgres_actions',
        'test_children_precede_postgres_and_stop_close_share_budget',
        'test_postgres_wait_consumes_close_budget',
    ),
    'test_supervisor_startup_rollback.py': (
        'test_unconfirmed_launch_rollback_retains_owner_until_stop_retry',
        'test_retained_rollback_refuses_start_schedule_and_poll',
        'test_confirmed_launch_rollback_clears_owner_and_allows_explicit_retry',
        'test_constructor_failure_without_returned_owner_still_closes_log',
        'test_rollback_interruption_propagates_without_losing_owner',
        'test_log_cleanup_error_does_not_undo_confirmed_exit',
        'test_log_pump_with_failed_thread_start_can_close_on_rollback_retry',
        'test_rollback_pending_closes_admission_before_failed_hold_publication',
        'test_owned_child_uncertain_defers_postgres_until_retained_retry_confirms_exit',
        'test_main_retains_locks_and_control_until_startup_rollback_is_confirmed',
        'test_locked_stale_metadata_recovery_uses_exact_identity_not_pid_or_timestamp',
        'test_foreground_reconciliation_uses_existing_exact_identity_api',
        'test_live_or_uncertain_owner_metadata_is_never_replaced',
        'test_metadata_bound_to_another_instance_path_is_not_removed',
        'test_reconciliation_error_or_interruption_preserves_metadata_for_retry',
        'test_changed_instance_during_matching_removal_fails_closed',
    ),
    'test_supervisor_managed_postgres_gate.py': ('SupervisorManagedPostgresGateTests',),
    'test_observer_only_coordinated_shutdown.py': ('ObserverOnlyCoordinatedShutdownTests',),
    'test_discovery_producer_supervisor.py': ('DiscoveryProducerSupervisorTests',),
    'test_distributed_core_profile.py': (
        'test_default_distributed_core_profile_is_exact',
        'test_keychecks_are_independent_from_discovery_profile',
        'test_core_wrapper_selects_only_discovery_producers',
    ),
    'test_discovery_only_cycle.py': ('DiscoveryOnlyCycleTests',),
    'test_discovery_request_budgets.py': (
        'test_each_discovery_page_has_one_shared_attempt_deadline',
        'test_discovery_proxy_uses_source_read_timeout_and_at_most_three_attempts',
        'test_late_discovery_success_fails_closed_without_advancing_page',
    ),
    'test_supervisor_safety.py': (
        'AuthenticatedControlTests', 'DependencyFailureTests', 'DashboardManagerTests',
    ),
    'test_postgres_runtime.py': (
        'PostgresRuntimePathTests',
        'PostgresControllerTests',
        'PostgresBackendTests',
        'PostgresEntrypointTests',
    ),
    'test_container_security.py': (
        'NativeExecutablePolicyTests',
        'WindowsNativePolicyTests',
        'ContainerLifecyclePolicyTests',
        'LinuxFilesystemSecurityTests',
    ),
    'test_runtime_security.py': ('RuntimeSecurityTests',),
    'test_postgres_empty_initialization.py': (
        'NativePostgresPathTests',
        'InitializeEmptyTests',
        'InitializationEntrypointTests::test_cli_dispatches_to_lock_owning_initializer_after_preflight',
        'InitializationEntrypointTests::test_container_refusal_and_full_layout_gate_precede_application_imports',
        'InitializationEntrypointTests::test_bounded_maintenance_stop_requires_proof_and_never_closes_a_supplied_backend',
        'InitializationEntrypointTests::test_bootstrap_cannot_replace_an_existing_identity',
        'MaintenanceEntrypointTests',
    ),
    'test_process_identity_linux.py': ('LinuxProcessIdentityTests',),
    'test_container_migration_paths.py': (
        'LegacySpoolPathTests',
        'ImmutableNativeHardeningTests',
        'LegacyCutoverChecksTests',
        'NormalMigrationContractTests',
    ),
    'test_container_provider_portability.py': ('AskpassTests', 'ProviderDeadlineTests'),
    'test_container_e2e_helpers.py': ('ContainerE2EHelperTests', 'ContainerE2EFreshVolumeTests'),
    'test_container_import.py': ('ContainerImportTests',),
    'test_container_import_config.py': ('ContainerImportConfigTests',),
    'test_container_projection_recovery.py': ('ProjectionRecoveryTests',),
    'test_result_bundle_v2.py': ('ResultBundleV2Tests',),
    'test_pipeline_cutover_invariants.py': ('PipelineCutoverInvariantTests',),
    'test_custom_provider_detector_compatibility.py': (
        'CustomProviderDetectorPolicyTests', 'CustomProviderDetectorCLITests',
    ),
    'test_scan_execution.py': ('ScanExecutionTests',),
    'test_synthetic_llm_pipeline.py': ('SyntheticLLMPipelineTests',),
    'test_worker_api.py': (
        'WorkerServiceTests', 'WorkerAPIRouteTests', 'WorkerSlotRecoveryTests',
    ),
    'test_worker_api_runtime.py': (
        'ConfiguredWorkerServiceTests', 'WorkerSupervisorWiringTests',
    ),
    'test_worker_assignment.py': ('RemoteGitAssignmentBuilderTests',),
    'test_worker_assignment_runner.py': (
        'WorkerAssignmentRunnerProtocolTests', 'WorkerSlotRunnerTests',
    ),
    'test_worker_runner_handoff_linux.py': ('LinuxWorkerRunnerHandoffTests',),
    'test_worker_contracts.py': ('WorkerContractTests',),
    'test_worker_cli.py': ('WorkerCLITests',),
    'test_worker_local_state.py': ('WorkerLocalStateTests',),
    'test_worker_supervisor.py': ('WorkerSupervisorTests',),
    'test_worker_package.py': ('WorkerPackageTests',),
    'test_multisource_execution_snapshot.py': ('MultisourceExecutionSnapshotTests',),
    'test_remote_direct_credentials.py': (
        'test_direct_child_environment_preserves_operator_provider_settings',
        'test_direct_scanner_entries_reject_credentials_and_skip_docker_pool',
        'test_anonymous_docker_bearer_never_reads_account_pool',
        'test_direct_anonymous_docker_auth_failure_is_permanent',
        'test_huggingface_discovery_errors_never_include_response_body',
        'test_huggingface_missing_repository_is_permanent_and_not_retryable',
        'test_direct_provider_access_failures_are_permanent',
    ),
    'test_remote_worker_db.py': ('RemoteWorkerDBTests',),
    'test_admin_api.py': ('AdminAPITests',),
    'test_edge_deployment.py': (
        'test_updater_persists_canonical_state_and_expires_automatically',
        'test_updater_reuses_persisted_applied_digest_without_reloading',
        'test_updater_rejects_corrupt_state_before_commands_or_snippet_changes',
        'test_updater_rejects_symlinked_snippet_when_supported',
        'test_updater_rejects_non_ip_or_unsafe_addresses_without_commands',
        'test_updater_renders_deterministically_and_bounds_matcher_lines',
        'test_updater_rolls_back_state_and_snippet_when_reload_fails',
        'test_updater_rolls_back_without_reload_when_validation_fails',
        'test_fail2ban_filter_counts_only_redacted_supplied_bad_credentials',
        'test_fail2ban_jail_is_persistent_bounded_and_admin_only',
        'test_caddy_routes_and_failure_log_are_closed_and_redacted',
        'test_edge_compose_only_opts_runtime_namespace_into_https_publication',
        'test_edge_image_and_startup_require_pinned_caddy_hash_and_random_prefix',
        'test_edge_e2e_image_inputs_are_exact_and_host_orchestrator_stays_host_only',
        'test_production_server_omits_trufflehog_but_worker_and_test_retain_it',
        'test_production_runbook_requires_fixed_host_agent_and_runtime_paths',
        'test_real_caddy_e2e_crawls_all_admin_routes_and_detail',
    ),
}

HELPER_SOURCES = ('owned_process_helper.py', 'container_e2e.py', 'container_import_stop_e2e.py',
                  'container_projection_recovery_e2e.py', 'parity_helpers.py')


def selected_nodes():
    """Resolve only the named declarations, without importing any test or app."""
    result = []
    for filename, selectors in SELECTION.items():
        path = TESTS / filename
        if path.resolve().parent != TESTS or not path.is_file():
            raise SystemExit('container-unit: missing or nonlocal selected source: ' + filename)
        tree = ast.parse(path.read_text(encoding='utf-8'), filename=str(path))
        compile(tree, str(path), 'exec')
        for selector in selectors:
            body = tree.body
            declaration = None
            for name in selector.split('::'):
                matches = [node for node in body
                           if isinstance(node, (ast.ClassDef, ast.FunctionDef)) and node.name == name]
                if len(matches) != 1:
                    raise SystemExit('container-unit: missing or ambiguous selection: ' + filename + '::' + selector)
                declaration = matches[0]
                body = declaration.body
            if isinstance(declaration, ast.ClassDef):
                members = [node.name for node in body
                           if isinstance(node, ast.FunctionDef) and node.name.startswith('test_')]
                nodes = [filename + '::' + selector + '::' + member for member in members]
            elif isinstance(declaration, ast.FunctionDef) and declaration.name.startswith('test_'):
                nodes = [filename + '::' + selector]
            else:
                nodes = []
            if not nodes:
                raise SystemExit('container-unit: empty test selection: ' + filename + '::' + selector)
            result.extend(nodes)
    if not result or len(result) != len(set(result)):
        raise SystemExit('container-unit: empty or duplicate selection')
    for filename in HELPER_SOURCES:
        path = TESTS / filename
        if path.resolve().parent != TESTS or not path.is_file():
            raise SystemExit('container-unit: missing or nonlocal helper source: ' + filename)
        compile(ast.parse(path.read_text(encoding='utf-8'), filename=str(path)), str(path), 'exec')
    return result


def main(argv=None):
    if not (sys.flags.isolated and sys.flags.no_site and sys.flags.dont_write_bytecode) or sys.flags.optimize:
        raise SystemExit('container-unit requires python -I -S -B without optimization')
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False, exit_on_error=False)
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument('--list', action='store_true', help='print exact leaf selectors without importing tests')
    modes.add_argument('--check-selection', action='store_true', help='validate selected source AST and node IDs only')
    parser.add_argument('-k', dest='keyword', metavar='EXPRESSION', help='narrow the allowlist using pytest -k')
    try:
        args, extra = parser.parse_known_args(argv)
    except argparse.ArgumentError:
        parser.error('invalid runner arguments')
    if extra or (args.keyword is not None and (args.list or args.check_selection)):
        parser.error('only --list, --check-selection, or -k EXPRESSION is supported')
    if args.keyword is not None and (not args.keyword.strip() or len(args.keyword) > 512
                                     or any(char in args.keyword for char in '\x00\r\n')):
        parser.error('-k requires a nonempty single-line expression of at most 512 characters')

    nodes = selected_nodes()
    if args.list:
        for node in nodes:
            print('tests/' + node)
        return 0
    if args.check_selection:
        print(f'container-unit: AST OK; {len(SELECTION)} modules, {len(nodes)} test definitions, '
              'no runner skips (parametrizations expand in pytest)')
        return 0

    if sys.platform != 'linux' or os.geteuid() != 10001 or TESTS != Path('/opt/truf/tests'):
        raise SystemExit('container-unit execution requires Linux UID 10001 at /opt/truf/tests')
    if not APP.is_dir() or os.access(APP, os.W_OK):
        raise SystemExit('container-unit requires readable, read-only /opt/truf/app')

    # A fresh allowlist also removes mixed-case/future credentials, not just known
    # TRUF_/SCANNER_/SCAN_/TRUFFLEHOG_/KEYCHECK_/PG*/DATABASE_URL/proxy names.
    os.environ.clear()
    os.environ.update({
        'PATH': '/usr/local/bin:/usr/bin:/bin',
        'LANG': 'C.UTF-8', 'LC_ALL': 'C.UTF-8',
        'PYTHONDONTWRITEBYTECODE': '1', 'PYTHONNOUSERSITE': '1', 'PYTHONPATH': '',
        'PYTEST_DISABLE_PLUGIN_AUTOLOAD': '1', 'PYTEST_ADDOPTS': '', 'PYTEST_PLUGINS': '',
    })
    os.umask(0o077)
    with tempfile.TemporaryDirectory(prefix='container-unit-', dir='/tmp') as temporary:
        private = Path(temporary).resolve()
        details = private.stat()
        if private.parent != Path('/tmp') or details.st_uid != 10001 or stat.S_IMODE(details.st_mode) != 0o700:
            raise SystemExit('container-unit temporary directory is not private beneath /tmp')
        for name in ('ALLTEMP', 'TEMP', 'TMP', 'TMPDIR', 'HOME'):
            os.environ[name] = str(private)
        for name, child in (('XDG_CONFIG_HOME', 'config'), ('XDG_CACHE_HOME', 'cache'), ('XDG_DATA_HOME', 'data')):
            os.environ[name] = str(private / child)
        tempfile.tempdir = None
        with tempfile.NamedTemporaryFile(prefix='probe-') as probe:
            actual = Path(probe.name).resolve()
            if (Path(tempfile.gettempdir()).resolve() != private or actual.parent != private
                    or stat.S_IMODE(actual.stat().st_mode) != 0o600):
                raise SystemExit('container-unit tempfile escaped its private directory')

        bootstrap = runpy.run_path(str(APP / 'child_bootstrap.py'))
        bootstrap['_enable_dependency_paths']('supervisor')
        sys.path.insert(0, str(APP))
        import pytest
        import psycopg
        import requests
        import paths

        # Only scanner's import-time defaults are relocated. Restore paths before
        # collection so portable-path and security checks exercise real policy.
        blocked_import = AssertionError('container-unit forbids import-time runtime I/O')
        with ExitStack() as imports:
            imports.enter_context(mock.patch.multiple(paths, APP_DIR=str(private / 'app'), CANONICAL_ROOT=str(private)))
            for target in ('builtins.open', 'os.open', 'os.mkdir', 'os.makedirs', 'sqlite3.connect',
                           'subprocess.Popen.__init__', 'threading.Thread.start', 'socket.socket.__init__',
                           'requests.Session.request', 'psycopg.connect'):
                imports.enter_context(mock.patch(target, side_effect=blocked_import))
            import scanner

        # Filesystem/process mocks have ended: tests exercise real ACLs and locks.
        os.chdir(ROOT)
        sys.path.insert(1, str(TESTS))

        def loopback_only(event, arguments):
            if event in ('socket.getaddrinfo', 'socket.gethostbyname', 'socket.gethostbyaddr'):
                host = arguments[0]
            elif event in ('socket.connect', 'socket.bind', 'socket.sendto', 'socket.getnameinfo'):
                address = arguments[0] if event == 'socket.getnameinfo' else arguments[-1]
                host = address[0] if isinstance(address, tuple) and address else None
            else:
                return
            if host not in ('127.0.0.1', '::1', 'localhost'):
                raise AssertionError('container-unit permits loopback sockets only')

        sys.addaudithook(loopback_only)

        class Scope:
            def pytest_collection_modifyitems(self, items):
                for item in items:
                    node = item.nodeid.removeprefix('tests/').split('[', 1)[0]
                    if node not in nodes:
                        raise pytest.UsageError('container-unit collected a test outside its allowlist')

        options = [
            '-c', '/dev/null', '--rootdir', str(ROOT), '--noconftest',
            '-p', 'no:cacheprovider', '-o', 'addopts=', '--basetemp', str(private / 'pytest'),
            '--tb=short', '--show-capture=no', '-ra',
        ]
        if args.keyword is not None:
            options.append('-k=' + args.keyword)
        options.extend(str(TESTS / node) for node in nodes)
        # Tests may replace these fences with their own scoped mocks, never real
        # PostgreSQL or provider transports. Native local processes stay real.
        with mock.patch.object(psycopg, 'connect', side_effect=AssertionError('external PostgreSQL is forbidden')), \
                mock.patch.object(requests.Session, 'request', side_effect=AssertionError('provider HTTP is forbidden')):
            return int(pytest.main(options, plugins=[Scope()]))


if __name__ == '__main__':
    try:
        code = main()
    except Exception as error:
        print('container-unit bootstrap failed (' + type(error).__name__ + '); details withheld', file=sys.stderr)
        code = 1
    raise SystemExit(code)