Publish curated Unraid-AppOps-Gateway source from 98b19114655572fd0244dbd68420e8491a4337d5

backend/tests/test_controller_hub_release_volume.py

@@ -12,6 +12,7 @@ import sys
 import tarfile
 import threading
 import types
+from contextlib import contextmanager, nullcontext
 from pathlib import Path
 
 import pytest
@@ -2534,7 +2535,7 @@ def test_stop_rejects_failed_storage_before_recording_intent_or_stopping(monkeyp
 
 @pytest.mark.parametrize('native_read_only_graph', [False, True])
 def test_stop_allows_healthy_storage_and_preserves_intent_before_stop(monkeypatch, native_read_only_graph):
-    monkeypatch.setattr(module, '_cleanup_promotion_host_probe', lambda *a: None)
+    monkeypatch.setattr(module, '_cleanup_promotion_host_probe', lambda *a: nullcontext((None, lambda: None)))
     receipt = {'phase': 'prepared'}
     events = []
     monkeypatch.setattr(module, '_load_transaction', lambda _: (42, receipt, {}))
@@ -3615,7 +3616,7 @@ def test_legacy_receipt_projection_rejects_unsafe_mount_before_docker(compose, d
 @pytest.mark.parametrize("phase", ["prepared", "stopped", "activated", "verified"])
 @pytest.mark.parametrize("failure", ["read_only", "full"])
 def test_forward_stop_rechecks_storage_after_intent_or_on_resume(monkeypatch, phase, failure):
-    monkeypatch.setattr(module, "_cleanup_promotion_host_probe", lambda *a: None)
+    monkeypatch.setattr(module, "_cleanup_promotion_host_probe", lambda *a: nullcontext((None, lambda: None)))
     receipt = {"phase": phase, "imageId": IMAGE_ID,
                "previousRevision": OLD_REVISION, "sourceRevision": REVISION}
     persisted = phase != "prepared"
@@ -3660,7 +3661,7 @@ def test_probe_cleanup_failure_precedes_stop_intent_and_controller_stop(monkeypa
 
 
 def test_promotion_cleanup_uses_incoming_code_but_only_attested_previous_source(monkeypatch, tmp_path):
-    receipt = {"imageId": IMAGE_ID, "previousRevision": OLD_REVISION,
+    receipt = {"phase": "prepared", "imageId": IMAGE_ID, "previousRevision": OLD_REVISION,
                "sourceRevision": REVISION, "sourceTree": "d" * 40}
     old_source = module._expected_revision_path(OLD_REVISION).as_posix()
     events = []
@@ -3671,20 +3672,24 @@ def test_promotion_cleanup_uses_incoming_code_but_only_attested_previous_source(
         return {"Id": "c" * 64, "Mounts": [{"Destination": module.EXECUTABLE_TARGET,
                                                "Source": old_source}]}
     monkeypatch.setattr(module, "_controller_image_inspect", inspect)
-    monkeypatch.setattr(module, "_read_tar_tree", lambda _: ({}, {"": 0o755}))
-    monkeypatch.setattr(module, "_release_identity", lambda _: {
-        "sourceRevision": REVISION, "sourceTree": "d" * 40})
-    monkeypatch.setattr(module, "_write_private_tree", lambda *a: None)
-    def run(args, **kw):
-        assert args[0] == "python3"
-        assert Path(args[1]).name == "runner_resource_envelope.py"
-        assert args[2:] == ["--cleanup-stopped-host-probe", IMAGE_ID, old_source]
-        assert kw == {"max_output": 4096, "diagnostic_action": "probe_cleanup"}
+    @contextmanager
+    def lock(*, native_cleanup):
+        assert native_cleanup is True
+        events.append("lock")
+        yield lambda: None
+        events.append("unlock")
+    def cleanup(**kw):
+        assert kw["image_id"] == IMAGE_ID and kw["code_source"] == old_source
+        kw["assert_lock"]()
         events.append("cleanup")
-        return b""
-    monkeypatch.setattr(module, "_run", run)
-    module._cleanup_promotion_host_probe(receipt, {"controller-after.tar": b"qualified"})
-    assert events == ["attest", "cleanup", "attest"]
+    validator = types.SimpleNamespace(_host_probe_lifecycle_lock=lock, _cleanup_stopped_host_probe_locked=cleanup)
+    monkeypatch.setattr(module, "_promotion_probe_validator", lambda *args: nullcontext(validator))
+    monkeypatch.setattr(module, "_run", lambda *args, **kwargs: pytest.fail("No nested child lock"))
+    with module._cleanup_promotion_host_probe(receipt, {"controller-after.tar": b"qualified"}) as (identifier, assert_lock):
+        assert identifier == "c" * 64
+        assert_lock()
+        events.append("stop-window")
+    assert events == ["attest", "lock", "cleanup", "attest", "stop-window", "unlock"]
 
 
 @pytest.mark.parametrize("present", [False, True])
@@ -3698,11 +3703,15 @@ def test_legacy_promotion_allows_only_no_probe_without_adopting_legacy_source(mo
         calls.append(args)
         return b"d" * 64 if present else b""
     monkeypatch.setattr(module, "_run", run)
+    validator = types.SimpleNamespace(_host_probe_lifecycle_lock=lambda **kw: nullcontext(lambda: None))
+    monkeypatch.setattr(module, "_promotion_probe_validator", lambda *args: nullcontext(validator))
     if present:
         with pytest.raises(module.VolumeTransitionError, match="controller_volume_probe_cleanup_identity_invalid"):
-            module._cleanup_promotion_host_probe(receipt, {})
+            with module._cleanup_promotion_host_probe(receipt, {}):
+                pytest.fail("Unknown legacy probe admitted")
     else:
-        module._cleanup_promotion_host_probe(receipt, {})
+        with module._cleanup_promotion_host_probe(receipt, {}):
+            pass
     assert calls == [["docker", "ps", "-a", "--no-trunc", "--filter",
                      "name=^itworx-runner-cgroup-probe$", "--format", "{{.ID}}"]]
 
@@ -3713,7 +3722,7 @@ def test_resumed_forward_stop_of_stopped_controller_needs_no_storage(monkeypatch
     monkeypatch.setattr(module, "_load_transaction", lambda _: (42, receipt, {}))
     monkeypatch.setattr(module.os, "close", lambda _: None)
     monkeypatch.setattr(module, "_controller_image_inspect", lambda *_a, **_kw: {
-        "State": {"Running": False}, "Mounts": [{"Destination": module.EXECUTABLE_TARGET,
+        "Id": "c" * 64, "State": {"Running": False}, "Mounts": [{"Destination": module.EXECUTABLE_TARGET,
         "Source": module._expected_revision_path(OLD_REVISION).as_posix()}]})
     monkeypatch.setattr(module, "_require_writable_promotion_storage",
                         lambda: pytest.fail("Already stopped recovery must remain possible"))
@@ -3721,6 +3730,166 @@ def test_resumed_forward_stop_of_stopped_controller_needs_no_storage(monkeypatch
     module.stop_promotion(Path("/unused-backup"))
 
 
+def test_qualified_probe_modules_use_private_imports_and_leave_bare_modules_untouched(monkeypatch):
+    names = ('mcp_hub_release_store', 'runner_profiles', 'runner_resource_envelope')
+    entries = {f'{name}.py': ((ROOT / 'ops/gitea-deploy-broker' / f'{name}.py').read_bytes(), 0o644) for name in names}
+    monkeypatch.setattr(module, '_read_tar_tree', lambda _: (entries, {'': 0o755}))
+    monkeypatch.setattr(module, '_release_identity', lambda _: {'sourceRevision': REVISION, 'sourceTree': 'd' * 40})
+    sentinels = {name: types.ModuleType(f'foreign_{name}') for name in names}
+    for name, value in sentinels.items():
+        monkeypatch.setitem(sys.modules, name, value)
+    before = set(sys.modules)
+    before_path = list(sys.path)
+    receipt = {'sourceRevision': REVISION, 'sourceTree': 'd' * 40}
+    with module._promotion_probe_validator(receipt, {'controller-after.tar': b'qualified'}) as validator:
+        assert validator._HOST_PROBE_NAME == 'itworx-runner-cgroup-probe'
+        assert validator.PinnedDirectory.__module__.startswith('_itworx_promotion_probe_')
+        assert len(validator.RUNNER_PROFILES) == 6
+        assert all(sys.modules[name] is value for name, value in sentinels.items())
+        assert sys.path == before_path
+    assert not {name for name in set(sys.modules) - before if name.startswith('_itworx_promotion_probe_')}
+    assert all(sys.modules[name] is value for name, value in sentinels.items())
+
+
+@pytest.fixture
+def probe_stop_window(monkeypatch):
+    receipt = {'phase': 'prepared', 'imageId': IMAGE_ID, 'previousRevision': OLD_REVISION,
+               'sourceRevision': REVISION, 'sourceTree': 'd' * 40}
+    state = {'running': True, 'held': False, 'inspections': 0, 'events': [], 'real_close': os.close}
+    old_source = module._expected_revision_path(OLD_REVISION).as_posix()
+    monkeypatch.setattr(module, '_load_transaction', lambda _: (42, receipt, {
+        'controller-after.tar': b'qualified', 'compose-rollback.json': b'old', 'compose-after.json': b'new'}))
+    monkeypatch.setattr(module.os, 'close', lambda _: None)
+    monkeypatch.setattr(module, '_require_writable_promotion_storage', lambda: None)
+    def inspect(*args, **kwargs):
+        state['inspections'] += 1
+        identifier = 'c' * 64
+        if state.get('changed_on') == state['inspections']:
+            identifier = 'f' * 64
+        return {'Id': identifier, 'State': {'Running': state['running']},
+                'Mounts': [{'Destination': module.EXECUTABLE_TARGET, 'Source': state.get('source', old_source)}]}
+    monkeypatch.setattr(module, '_controller_image_inspect', inspect)
+    def assert_lock():
+        assert state['held']
+    @contextmanager
+    def lock(*, native_cleanup):
+        assert native_cleanup is True and not state['held']
+        state['held'] = True
+        state['events'].append('lock')
+        try:
+            yield assert_lock
+        finally:
+            state['held'] = False
+            state['events'].append('unlock')
+    def cleanup(**kwargs):
+        kwargs['assert_lock']()
+        state['events'].append('cleanup')
+        if state.get('cleanup_failure'):
+            error = RuntimeError('private-secret-canary')
+            error.condition = state['cleanup_failure']
+            raise error
+    validator = types.SimpleNamespace(_host_probe_lifecycle_lock=lock, _cleanup_stopped_host_probe_locked=cleanup)
+    monkeypatch.setattr(module, '_promotion_probe_validator', lambda *args: nullcontext(validator))
+    def intent(_fd, value, phase):
+        assert_lock()
+        state['events'].append('intent')
+        value['phase'] = phase
+        return value
+    monkeypatch.setattr(module, '_set_promotion_phase', intent)
+    def compose(*args):
+        assert_lock()
+        state['events'].append('compose-stop')
+        if not state.get('still_running'):
+            state['running'] = False
+    monkeypatch.setattr(module, 'compose_action', compose)
+    return receipt, state, validator
+
+
+@pytest.mark.parametrize('phase', ['prepared', 'stopped', 'activated', 'verified'])
+def test_probe_lifecycle_owner_covers_cleanup_intent_and_exact_confirmed_controller_stop(probe_stop_window, phase):
+    receipt, state, _validator = probe_stop_window
+    receipt['phase'] = phase
+    if phase in {'activated', 'verified'}:
+        state['source'] = module._expected_revision_path(REVISION).as_posix()
+    module.stop_promotion(Path('/unused-backup'))
+    assert state['events'] == ['lock', 'cleanup', *(['intent'] if phase == 'prepared' else []), 'compose-stop', 'unlock']
+    assert state['inspections'] == 4 and state['running'] is False and state['held'] is False
+
+
+@pytest.mark.parametrize('failure', ['identity_invalid', 'identity_changed', 'remove_busy', 'foreign-unapproved-condition'])
+def test_untrusted_or_active_probe_prevents_intent_and_stop_with_finite_redaction(probe_stop_window, failure):
+    receipt, state, _validator = probe_stop_window
+    state['cleanup_failure'] = failure
+    with pytest.raises(module.VolumeTransitionError) as error:
+        module.stop_promotion(Path('/unused-backup'))
+    assert receipt['phase'] == 'prepared' and state['running'] is True
+    assert state['events'] == ['lock', 'cleanup', 'unlock']
+    assert 'private-secret-canary' not in str(error.value)
+    expected = 'probe_' + failure
+    assert error.value.command_failure == ('probe_cleanup', 'exit_1', expected if expected in module._COMMAND_DIAGNOSTIC_CONDITIONS else None)
+
+
+@pytest.mark.parametrize('failure', ['changed_before_stop', 'changed_after_stop', 'still_running'])
+def test_stop_never_claims_success_without_same_controller_and_stopped_state(probe_stop_window, failure):
+    receipt, state, _validator = probe_stop_window
+    if failure == 'still_running':
+        state['still_running'] = True
+    else:
+        state['changed_on'] = 3 if failure == 'changed_before_stop' else 4
+    with pytest.raises(module.VolumeTransitionError):
+        module.stop_promotion(Path('/unused-backup'))
+    assert receipt['phase'] == 'stopped' and state['held'] is False
+    if failure == 'changed_before_stop':
+        assert 'compose-stop' not in state['events']
+
+
+@pytest.mark.skipif(os.name != 'posix', reason='native independent process lifecycle exclusion')
+def test_late_independent_probe_is_excluded_through_persist_and_confirmed_stop(probe_stop_window, monkeypatch, tmp_path):
+    import fcntl
+    receipt, state, validator = probe_stop_window
+    lock_path = tmp_path / 'probe.lock'
+    lock_path.write_bytes(b'')
+    close = state['real_close']
+    # The fixture suppresses closing its fake transaction descriptor only.
+    monkeypatch.setattr(module.os, 'close', close)
+    monkeypatch.setattr(module, '_load_transaction', lambda _: (os.open(lock_path, os.O_RDONLY), receipt, {
+        'controller-after.tar': b'qualified', 'compose-rollback.json': b'old', 'compose-after.json': b'new'}))
+    child = ('import fcntl,os,sys; fd=os.open(sys.argv[1],os.O_RDWR);\n'
+             'try: fcntl.flock(fd,fcntl.LOCK_EX|fcntl.LOCK_NB)\n'
+             'except BlockingIOError: sys.exit(0)\n'
+             'sys.exit(1)')
+    def excluded():
+        result = subprocess.run([sys.executable, '-c', child, str(lock_path)], capture_output=True, timeout=5, check=False)
+        assert result.returncode == 0 and not result.stdout and not result.stderr
+    @contextmanager
+    def native_lock(*, native_cleanup):
+        assert native_cleanup is True
+        with lock_path.open('r+b') as stream:
+            fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            state['held'] = True
+            try:
+                yield excluded
+            finally:
+                state['held'] = False
+                fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
+    validator._host_probe_lifecycle_lock = native_lock
+    cleanup = validator._cleanup_stopped_host_probe_locked
+    def checked_cleanup(**kwargs):
+        excluded()
+        cleanup(**kwargs)
+    validator._cleanup_stopped_host_probe_locked = checked_cleanup
+    stop = module._stop_current_controller
+    def checked_stop(*args, **kwargs):
+        excluded()
+        stop(*args, **kwargs)
+        excluded()
+    monkeypatch.setattr(module, '_stop_current_controller', checked_stop)
+    module.stop_promotion(Path('/unused-backup'))
+    assert state['running'] is False
+    result = subprocess.run([sys.executable, '-c', child, str(lock_path)], capture_output=True, timeout=5, check=False)
+    assert result.returncode == 1
+
+
 def test_rollback_stop_remains_available_without_forward_storage_gate(monkeypatch):
     receipt = {"phase": "stopped", "imageId": IMAGE_ID,
                "previousRevision": OLD_REVISION, "sourceRevision": REVISION}

backend/tests/test_runner_resource_envelope.py

@@ -979,6 +979,40 @@ def test_probe_cleanup_refuses_foreign_running_or_changed_identity(monkeypatch,
         envelope.cleanup_stopped_host_probe(image_id=image_id, code_source=code_source)
 
 
+@pytest.mark.parametrize("drift", [None, "extra_mount", "changed_mount_field", "changed_unknown_mount_field",
+                                   "changed_top_level_field", "changed_host_mounts", "changed_state"])
+def test_probe_cleanup_cas_ignores_only_complete_mount_collection_order(monkeypatch, drift):
+    envelope = _module()
+    probe, image_id, code_source = _host_probe_fixture(envelope)
+    fresh = copy.deepcopy(probe)
+    fresh["Mounts"].reverse()
+    if drift == "extra_mount":
+        fresh["Mounts"].append(copy.deepcopy(fresh["Mounts"][0]))
+    elif drift == "changed_mount_field":
+        fresh["Mounts"][0]["RW"] = True
+    elif drift == "changed_unknown_mount_field":
+        fresh["Mounts"][0]["unexpectedMetadata"] = "changed"
+    elif drift == "changed_top_level_field":
+        fresh["unexpectedMetadata"] = "changed"
+    elif drift == "changed_host_mounts":
+        fresh["HostConfig"]["Mounts"].reverse()
+    elif drift == "changed_state":
+        fresh["State"]["unexpectedMetadata"] = "changed"
+    monkeypatch.setattr(envelope, "_host_probe_image_labels", lambda _: {})
+    inspected = iter([probe, fresh, None])
+    monkeypatch.setattr(envelope, "_inspect_host_probe_by_name", lambda: next(inspected))
+    removed = []
+    monkeypatch.setattr(envelope, "_remove_cleanup_probe", lambda identifier: removed.append(identifier))
+    if drift is None:
+        assert envelope.cleanup_stopped_host_probe(image_id=image_id, code_source=code_source)
+        assert removed == [probe["Id"]]
+    else:
+        with pytest.raises(RuntimeError, match="resource_cgroup_probe_cleanup_identity_changed"):
+            envelope.cleanup_stopped_host_probe(image_id=image_id, code_source=code_source)
+        assert removed == []
+    assert fresh != probe
+
+
 @pytest.mark.parametrize("stderr,condition", [
     (b"private-canary: no space left on device", "remove_storage_full"),
     (b"private-canary: read-only file system", "remove_read_only"),