diff --git a/tests/net/test_wg_iface_lifecycle.py b/tests/net/test_wg_iface_lifecycle.py index f9c78b79..78c5f542 100644 --- a/tests/net/test_wg_iface_lifecycle.py +++ b/tests/net/test_wg_iface_lifecycle.py @@ -245,6 +245,8 @@ def test_open_wg_iface_shields_cancelled_cleanup( def test_open_wg_bindspace_nests_resource_lifetimes( + tmp_path: Path, + request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch, ) -> None: ''' @@ -270,14 +272,16 @@ def test_open_wg_bindspace_nests_resource_lifetimes( bindspace_spec: BindspaceSpec = BindspaceSpec( kind='netns', ) + token = (tmp_path / 'netns').open('w+b') + request.addfinalizer(token.close) bindspace: Bindspace = Bindspace( spec=bindspace_spec, ref=BindspaceRef( kind='netns', key=None, - inode=1, + inode=os.fstat(token.fileno()).st_ino, ), - namespace_fd=None, + namespace_fd=token.fileno(), ownership='borrowed', ) outer_spec: WGTunnelSpec = WGTunnelSpec( @@ -384,18 +388,16 @@ def test_open_wg_bindspace_nests_resource_lifetimes( sys.platform != 'linux', reason='network namespaces are Linux-only', ) -def test_public_wg_bindspace_scopes_root_actor( +def test_public_wg_bindspace_scopes_child_actor( monkeypatch: pytest.MonkeyPatch, tpt_proto: str, ) -> None: ''' - Public network contexts must fully enclose the root runtime. + Public network contexts must fully enclose supervised children. Attach the real current netns through `tractor.net`, fake only WG - interface provisioning, and open a real root actor with the yielded - `Bindspace`. The trace and inode checks prove interface setup wraps - actor startup, the runtime occupies the realized bindspace, and root - restoration finishes before network-resource teardown. + interface provisioning, and supervise a child in the yielded + `Bindspace`. The nursery reaps it before network-resource teardown. ''' events: list[str] = [] @@ -450,10 +452,18 @@ def test_public_wg_bindspace_scopes_root_actor( ) as bindspace: events.append('bindspace-open') async with tractor.open_root_actor( - bindspace=bindspace, + start_method='trio', enable_transports=[tpt_proto], ): events.append('root-open') + async with tractor.open_nursery() as nursery: + portal = await nursery.start_actor( + 'scoped-worker', + bindspace=bindspace, + parent_addr=tractor.current_actor().accept_addr, + registry_addrs=tractor.current_actor().reg_addrs, + ) + await portal.cancel_actor() assert bindspace.namespace_fd is not None assert os.fstat( bindspace.namespace_fd, diff --git a/tests/test_netns_spawn.py b/tests/test_netns_spawn.py index 6c62f7cb..1e1da178 100644 --- a/tests/test_netns_spawn.py +++ b/tests/test_netns_spawn.py @@ -24,7 +24,6 @@ import trio import tractor from tractor import ( _child, - _root, ) from tractor.devx import _proctitle from tractor.net._bindspace import ( @@ -185,6 +184,7 @@ def _run_in_unshared_netns( '-x', '--tb=short', '--no-header', + '--show-capture=all', '--timeout=30', ], env=nested_env, @@ -220,7 +220,7 @@ class _MockIpcServer: def _netns_bootstrap_from_cmd( command: list[str], -) -> tuple[int, int]: +) -> tuple[int, int, int]: ''' Parse the namespace tuple that the exec child would receive. @@ -320,6 +320,7 @@ def test_root_netns_same_namespace_skips_setns( def fail_enter_netns( inherited_fd: int, + device: int, inode: int, ) -> int: ''' @@ -372,6 +373,7 @@ def test_root_netns_restores_after_body_error( def fake_enter_netns( inherited_fd: int, + device: int, inode: int, ) -> int: ''' @@ -411,10 +413,10 @@ def test_root_netns_restores_on_trio_cancellation( Trio cancellation must not interrupt root-netns restoration. Use deterministic namespace stand-ins and cancel the task inside - `_enter_root_bindspace()` immediately before an explicit Trio + `_enter_netns_temporarily()` immediately before an explicit Trio checkpoint. The enclosing `CancelScope` catches cancellation only - after async-context exit. Two recorded sync transitions and exact - FD state then prove restoration and close completed first. + after the fake synchronous scope exits. Two recorded transitions + and exact FD state prove restoration and close completed first. ''' original_path: Path = tmp_path / 'cancel-original-netns' target_path: Path = tmp_path / 'cancel-target-netns' @@ -427,6 +429,7 @@ def test_root_netns_restores_on_trio_cancellation( def fake_enter_netns( inherited_fd: int, + device: int, inode: int, ) -> int: ''' @@ -446,7 +449,7 @@ def test_root_netns_restores_on_trio_cancellation( ''' with trio.CancelScope() as cancel_scope: - async with _root._enter_root_bindspace(bindspace): + with _netns._enter_netns_temporarily(bindspace): cancel_scope.cancel() await trio.lowlevel.checkpoint() @@ -456,6 +459,7 @@ def test_root_netns_restores_on_trio_cancellation( trio.run(main) # Target entry is recorded first; original-netns restoration # is recorded when `_enter_root_bindspace()` exits. + # The removed adapter is now exercised through its sync primitive. assert transitions == [ bindspace.ref.inode, original_path.stat().st_ino, @@ -496,6 +500,7 @@ def test_root_netns_restore_error_precedence( def fail_restore( inherited_fd: int, + device: int, inode: int, ) -> int: ''' @@ -538,33 +543,26 @@ def test_root_netns_requires_live_bindspace_fd() -> None: ''' Root entry cannot use `BindspaceRef.inode` without a live FD. - Construct a valid ref-only `Bindspace` and enter the real root - namespace scope directly. The concrete live-FD error must occur - before namespace capture, probes, sockets, or actor runtime work. + Reject a ref-only `Bindspace` at construction, before namespace + capture, probes, sockets, or actor runtime work. ''' key: str = 'missing-root-netns' - bindspace = Bindspace( - spec=BindspaceSpec( - kind='netns', - key=key, - ), - ref=BindspaceRef( - kind='netns', - key=key, - inode=1, - ), - # A stored inode cannot authorize namespace entry. - namespace_fd=None, - ownership='borrowed', - ) - - with pytest.raises( - ValueError, - match='bindspace.namespace_fd.*live netns FD', - ): + with pytest.raises(ValueError, match='namespace_fd.*non-negative'): + Bindspace( + spec=BindspaceSpec( + kind='netns', + key=key, + ), + ref=BindspaceRef( + kind='netns', + key=key, + inode=1, + ), + # A stored inode cannot authorize namespace entry. + namespace_fd=None, + ownership='borrowed', + ) # Scope entry must reject the missing live handle. - with _netns._enter_netns_temporarily(bindspace): - pytest.fail('root scope accepted a ref-only bindspace') @_linux_netns_only @@ -624,8 +622,8 @@ def test_bound_root_rejects_persistent_forkserver( ''' with pytest.raises( - NotImplementedError, - match='persistent forkserver', + TypeError, + match='unexpected keyword argument.*bindspace', ): async with tractor.open_root_actor( bindspace=bindspace, @@ -668,6 +666,7 @@ def test_enter_netns_rejects_mismatched_inherited_fd( ): _netns.enter_netns( namespace_file.fileno(), + os.fstat(namespace_file.fileno()).st_dev, # Deliberately differ from `token_path`'s inode. inode + 1, ) @@ -697,7 +696,10 @@ def test_enter_netns_verifies_post_entry_inode( def fake_stat(path: Path) -> SimpleNamespace: stat_calls.append(path) - return SimpleNamespace(st_ino=inode) + return SimpleNamespace( + st_dev=os.fstat(namespace_fd).st_dev, + st_ino=inode if len(stat_calls) > 1 else inode + 1, + ) namespace_file: BinaryIO with token_path.open('rb') as namespace_file: @@ -712,13 +714,14 @@ def test_enter_netns_verifies_post_entry_inode( entered_inode: int = _netns.enter_netns( namespace_fd, + os.fstat(namespace_fd).st_dev, inode, ) assert setns_calls == [ (namespace_fd, _netns.os.CLONE_NEWNET), ] - assert stat_calls == [_netns._SELF_NETNS] + assert stat_calls == [_netns._SELF_NETNS, _netns._SELF_NETNS] assert entered_inode == inode @@ -753,7 +756,10 @@ def test_enter_netns_rejects_wrong_post_entry_namespace( inode: int = token_path.stat().st_ino def fake_stat(path: Path) -> SimpleNamespace: - return SimpleNamespace(st_ino=inode + 1) + return SimpleNamespace( + st_dev=os.fstat(namespace_file.fileno()).st_dev, + st_ino=inode + 1, + ) monkeypatch.setattr( type(_netns._SELF_NETNS), @@ -766,6 +772,7 @@ def test_enter_netns_rejects_wrong_post_entry_namespace( ): _netns.enter_netns( namespace_file.fileno(), + os.fstat(namespace_file.fileno()).st_dev, # Deliberately differ from `fake_stat()`'s inode + 1. inode, ) @@ -782,7 +789,7 @@ def test_empty_netns_bootstrap_is_a_noop( syscall or descriptor ownership work for existing spawn callers. ''' - def fail_enter_netns(namespace_fd: int, inode: int) -> int: + def fail_enter_netns(namespace_fd: int, device: int, inode: int) -> int: ''' Reject namespace entry without an explicit capability. @@ -810,7 +817,7 @@ def test_invalid_netns_fd_is_never_closed( ''' entry_error = ValueError('invalid netns capability') - def fail_enter_netns(namespace_fd: int, inode: int) -> int: + def fail_enter_netns(namespace_fd: int, device: int, inode: int) -> int: ''' Raise the primary namespace bootstrap error. @@ -833,7 +840,7 @@ def test_invalid_netns_fd_is_never_closed( with pytest.raises(ValueError) as exc_info: _entry._consume_netns_bootstrap( - (namespace_fd, 1), # type: ignore[arg-type] + (namespace_fd, 0, 1), # type: ignore[arg-type] ) assert exc_info.value is entry_error @@ -854,7 +861,7 @@ def test_netns_entry_error_survives_close_failure( namespace_fd: int = 1 << 100 entry_error = ValueError('invalid netns capability') - def fail_enter_netns(inherited_fd: int, inode: int) -> int: + def fail_enter_netns(inherited_fd: int, device: int, inode: int) -> int: ''' Raise the primary namespace bootstrap error. @@ -865,7 +872,7 @@ def test_netns_entry_error_survives_close_failure( monkeypatch.setattr(_entry, 'enter_netns', fail_enter_netns) with pytest.raises(ValueError) as exc_info: - _entry._consume_netns_bootstrap((namespace_fd, 1)) + _entry._consume_netns_bootstrap((namespace_fd, 0, 1)) assert exc_info.value is entry_error assert entry_error.__notes__ @@ -888,7 +895,7 @@ def test_trio_child_cli_forwards_netns_bootstrap( calls: list[dict[str, object]] = [] uid: tuple[str, str] = ('cli-netns-child', 'test') parent_addr: tuple[str, int] = ('127.0.0.1', 1616) - bootstrap: tuple[int, int] = (12, 3456) + bootstrap: tuple[int, int, int] = (12, 7, 3456) def fake_actor_child_main(**kwargs: object) -> None: ''' @@ -926,51 +933,28 @@ def test_trio_spawn_requires_live_bindspace_fd() -> None: ''' A `BindspaceRef` alone cannot let a child enter its namespace. - Construct a valid `Bindspace` with its required identity metadata - but no open namespace FD. Calling `trio_proc()` must fail before - `open_process()` because an inode identifies a namespace but does - not provide an open handle that the child can inherit. + Attempt construction with identity metadata but no open namespace + FD. An inode identifies a namespace; construction must reject the + missing capability before the spawn backend can receive it. ''' key: str = 'missing-spawn-netns' - bindspace = Bindspace( - spec=BindspaceSpec( - kind='netns', - key=key, - ), - # A realized bindspace always retains identity metadata; this - # test isolates the missing live-FD condition. - ref=BindspaceRef( - kind='netns', - key=key, - inode=1, - ), - namespace_fd=None, - ownership='borrowed', - ) - uid: tuple[str, str] = ('missing-netns-fd', 'test') - - async def main() -> None: - ''' - Reject the ref-only capability before `open_process()`. - - ''' - with pytest.raises( - ValueError, - match='bindspace.namespace_fd.*required', - ): - await _trio.trio_proc( - name=uid[0], - actor_nursery=_SpawnTestNursery(), - subactor=_spawn_test_subactor(uid), - errors={}, - bind_addrs=[], - parent_addr=('127.0.0.1', 1616), - _runtime_vars={}, - bindspace=bindspace, - ) - - trio.run(main) + with pytest.raises(ValueError, match='namespace_fd.*non-negative'): + Bindspace( + spec=BindspaceSpec( + kind='netns', + key=key, + ), + # A realized bindspace always retains identity metadata; this + # test isolates the missing live-FD condition. + ref=BindspaceRef( + kind='netns', + key=key, + inode=1, + ), + namespace_fd=None, + ownership='borrowed', + ) def test_trio_spawn_relays_bindspace_to_child_actor( @@ -1040,6 +1024,8 @@ def test_trio_spawn_relays_bindspace_to_child_actor( async with tractor.open_nursery() as actor_nursery: portal: tractor.Portal = await actor_nursery.start_actor( 'netns-bootstrap-child', + parent_addr=tractor.current_actor().accept_addr, + registry_addrs=tractor.current_actor().reg_addrs, bindspace=bindspace, enable_modules=[__name__], proc_kwargs={ @@ -1075,72 +1061,14 @@ def test_trio_spawn_relays_bindspace_to_child_actor( os.close(inherited_fd) -def test_root_actor_enters_and_restores_bindspace( - tpt_proto: str, -) -> None: +@pytest.mark.parametrize('backend', ('trio', 'mp_spawn', 'mp_forkserver')) +def test_root_actor_rejects_async_bindspace(backend: str) -> None: ''' - `open_root_actor()` must enter its supplied networking bindspace. + Root networking cannot be changed around an async runtime scope. - Re-exec under an unprivileged user/net namespace, retain that - first netns as the target, then move nested pytest into a second. - A real UDS root actor must run its body in the target inode and - keep the source capability open. After full actor teardown, exact - FD and inode assertions prove its duplicate did not leak and the - caller thread returned to the second/original netns. ''' - if _run_in_unshared_netns( - test_name='test_root_actor_enters_and_restores_bindspace', - reexec_var='TRACTOR_TEST_ROOT_NETNS_E2E_REEXEC', - ): - return - - assert tpt_proto == 'uds' - target_netns_fd: int = os.open( - _SELF_NETNS_PATH, - os.O_RDONLY, - ) - target_netns_inode: int = os.fstat(target_netns_fd).st_ino - reffed_tgt_fds: set[int] = _fds_referencing( - target_netns_fd, - ) - bindspace: Bindspace = _bindspace_for_fd(target_netns_fd) - - # Pin the first disposable netns through `target_netns_fd`, then - # move the caller into a distinct second netns. This gives the root - # one real target to enter and one real caller netns to restore. - os.unshare(os.CLONE_NEWNET) - original_netns_fd: int = os.open( - _SELF_NETNS_PATH, - os.O_RDONLY, - ) - original_netns_inode: int = os.fstat(original_netns_fd).st_ino - assert original_netns_inode != target_netns_inode - - async def main() -> None: - ''' - Inspect the real root runtime inside the target netns. - - ''' - async with tractor.open_root_actor( - bindspace=bindspace, - enable_transports=['uds'], - ): - body_netns_inode: int = _SELF_NETNS_PATH.stat().st_ino - assert body_netns_inode == target_netns_inode - assert os.fstat(target_netns_fd).st_ino == ( - target_netns_inode - ) - - try: - trio.run(main) - assert _SELF_NETNS_PATH.stat().st_ino == original_netns_inode - assert _fds_referencing( - target_netns_fd, - ) == reffed_tgt_fds - assert os.fstat(target_netns_fd).st_ino == target_netns_inode - finally: - os.close(original_netns_fd) - os.close(target_netns_fd) + with pytest.raises(TypeError, match='unexpected keyword argument.*bindspace'): + tractor.open_root_actor(bindspace=None, start_method=backend) def test_trio_spawn_failure_closes_child_netns_fd_in_parent( @@ -1174,8 +1102,9 @@ def test_trio_spawn_failure_closes_child_netns_fd_in_parent( Fail after checking the child FD in `pass_fds` and the CLI. ''' - child_fd, expected_inode = _netns_bootstrap_from_cmd(command) + child_fd, expected_device, expected_inode = _netns_bootstrap_from_cmd(command) assert kwargs['pass_fds'] == (child_fd,) + assert os.fstat(child_fd).st_dev == expected_device assert os.fstat(child_fd).st_ino == expected_inode child_fds.append(child_fd) raise open_error @@ -1247,8 +1176,9 @@ def test_trio_spawn_cancel_closes_child_netns_fd_in_parent( Record the child FD, then signal that the open call is parked. ''' - child_fd, expected_inode = _netns_bootstrap_from_cmd(command) + child_fd, expected_device, expected_inode = _netns_bootstrap_from_cmd(command) assert kwargs['pass_fds'] == (child_fd,) + assert os.fstat(child_fd).st_dev == expected_device assert os.fstat(child_fd).st_ino == expected_inode child_fds.append(child_fd) open_called.set() @@ -1372,6 +1302,7 @@ def test_child_entry_consumes_netns_before_runtime( def fake_enter_netns( inherited_fd: int, + device: int, inode: int, ) -> int: ''' @@ -1504,8 +1435,9 @@ def test_child_entry_consumes_netns_before_runtime( ) actor: Any = ActorSpy() - bootstrap: tuple[int, int] = ( + bootstrap: tuple[int, int, int] = ( namespace_fd, + os.fstat(namespace_fd).st_dev, expected_inode, ) if backend == 'mp': @@ -1562,6 +1494,7 @@ def test_child_entry_failure_closes_netns_fd_before_runtime( def fail_enter_netns( inherited_fd: int, + device: int, inode: int, ) -> int: ''' @@ -1631,8 +1564,9 @@ def test_child_entry_failure_closes_netns_fd_before_runtime( ) actor: Any = ActorSpy() - bootstrap: tuple[int, int] = ( + bootstrap: tuple[int, int, int] = ( namespace_fd, + os.fstat(namespace_fd).st_dev, expected_inode, ) with pytest.raises(RuntimeError) as exc_info: @@ -1656,3 +1590,325 @@ def test_child_entry_failure_closes_netns_fd_before_runtime( assert exc_info.value is entry_error assert events == ['enter-netns'] _assert_fd_closed(namespace_fd) + + +@_linux_netns_only +def test_child_same_netns_needs_no_setns( + monkeypatch: pytest.MonkeyPatch, +) -> None: + '''Validate and consume an already-current namespace without privilege.''' + fd = os.open(_SELF_NETNS_PATH, os.O_RDONLY) + target = os.fstat(fd) + # A Python build without setns support can still inherit its netns. + monkeypatch.delattr(_netns.os, 'setns', raising=False) + assert _entry._consume_netns_bootstrap( + (fd, target.st_dev, target.st_ino), + ) == target.st_ino + _assert_fd_closed(fd) + + +@_linux_netns_only +@pytest.mark.parametrize('field', ('device', 'inode')) +def test_child_rejects_identity_mismatch_and_closes_fd( + field: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + '''Neither identity component may disagree with the transferred FD.''' + fd = os.open(_SELF_NETNS_PATH, os.O_RDONLY) + target = os.fstat(fd) + + def fail_setns(*args: object) -> None: + pytest.fail('identity mismatch reached setns') + + monkeypatch.setattr(_netns.os, 'setns', fail_setns, raising=False) + with pytest.raises(ValueError, match='Inherited namespace FD identity'): + _entry._consume_netns_bootstrap(( + fd, + target.st_dev + (field == 'device'), + target.st_ino + (field == 'inode'), + )) + _assert_fd_closed(fd) + + +async def _report_spawn_scope() -> dict: + '''Observe actual child runtime state after registration and entry.''' + from tractor.runtime import _state + + actor = tractor.current_actor() + ns = _SELF_NETNS_PATH.stat() + return { + 'identity': (ns.st_dev, ns.st_ino), + 'registrars': actor.reg_addrs, + 'runtime_registrars': _state._runtime_vars['_registry_addrs'], + 'parent': actor._parent_chan.raddr.unwrap(), + 'source': tractor.__file__, + } + + +@_linux_netns_only +def test_concurrent_children_share_fd_and_isolate_registrars( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + '''Real children borrow one capability with independent spawn routing.''' + from tractor.runtime import _state + + async def main() -> None: + with _netns.pin_current_netns() as bindspace: + fd = bindspace.namespace_fd + initial_fds = _fds_referencing(fd) + target = os.fstat(fd) + addresses = [('unix', str(tmp_path / f'r{i}.sock')) for i in range(3)] + async with tractor.open_root_actor( + registry_addrs=addresses, + enable_transports=['uds'], + start_method='trio', + ) as actor: + original_regs = list(actor.reg_addrs) + original_runtime = list(_state._runtime_vars['_registry_addrs']) + original_regs_obj = actor.reg_addrs + original_runtime_obj = _state._runtime_vars['_registry_addrs'] + selected_parent = next( + addr for addr in actor.accept_addrs + if addr != actor.accept_addr + ) + selections = [[addresses[0]], [addresses[1], addresses[2]]] + reports = [] + caller_tasks = [] + pin_tasks = [] + real_open = os.open + + def record_open(path, flags, *args, **kwargs): + result = real_open(path, flags, *args, **kwargs) + if path == _netns._SELF_NETNS: + pin_tasks.append(trio.lowlevel.current_task()) + return result + + async with tractor.open_nursery() as nursery: + async def start(index: int) -> None: + portal = await nursery.start_actor( + f'shared-{index}', + bindspace=bindspace, + parent_addr=selected_parent, + registry_addrs=selections[index], + enable_modules=[__name__], + ) + report = await portal.run(_report_spawn_scope) + assert list(map(tuple, report['registrars'])) == selections[index] + assert list(map(tuple, report['runtime_registrars'])) == selections[index] + assert tuple(report['parent']) == selected_parent + assert tuple(report['identity']) == (target.st_dev, target.st_ino) + assert report['source'] == tractor.__file__ + reports.append(report) + + async with trio.open_nursery() as tn: + for index in range(2): + tn.start_soon(start, index) + assert len(reports) == 2 + assert actor.reg_addrs is original_regs_obj + assert actor.reg_addrs == original_regs + assert _state._runtime_vars['_registry_addrs'] is original_runtime_obj + assert _state._runtime_vars['_registry_addrs'] == original_runtime + assert _fds_referencing(fd) == initial_fds + + # Default inheritance pins on this task before dispatch. + monkeypatch.setattr(_netns.os, 'open', record_open) + caller_tasks.append(trio.lowlevel.current_task()) + sibling = await nursery.start_actor( + 'default-sibling', + enable_modules=[__name__], + ) + monkeypatch.setattr(_netns.os, 'open', real_open) + assert pin_tasks == caller_tasks + report = await sibling.run(_report_spawn_scope) + assert list(map(tuple, report['registrars'])) == original_regs + assert list(map(tuple, report['runtime_registrars'])) == original_runtime + assert tuple(report['parent']) == actor.accept_addr + assert tuple(report['identity']) == (target.st_dev, target.st_ino) + await nursery.cancel() + assert _fds_referencing(fd) == initial_fds + + trio.run(main) + + +@pytest.mark.parametrize('option,value', [ + ('parent_addr', ('127.0.0.1', 1616)), + ('parent_addr', ['unix', '/tmp/legacy.sock']), + ('parent_addr', ('uds', '/tmp/alias.sock')), + ('parent_addr', ('unix', '/tmp/not-a-listener.sock')), + ('registry_addrs', []), + ('registry_addrs', (('tcp', '127.0.0.1', 1616),)), + ('registry_addrs', [('127.0.0.1', 1616)]), + ('registry_addrs', [('uds', '/tmp/alias.sock')]), + ('registry_addrs', [('tcp', '127.0.0.1', True)]), + ('registry_addrs', [('tcp', '127.0.0.1', 65536)]), + ('registry_addrs', [('tcp', 'invalid-host', 1616)]), + ('registry_addrs', [('unix', '')]), +]) +def test_start_actor_rejects_invalid_routing(option: str, value: object) -> None: + '''Invalid routing must fail before spawning or publishing child state.''' + from tractor.runtime import _state + + async def main() -> None: + async with tractor.open_nursery() as nursery: + original = dict(_state._runtime_vars) + with pytest.raises(ValueError): + await nursery.start_actor('invalid-routing', **{option: value}) + assert _state._runtime_vars == original + assert not nursery._children + + trio.run(main) + + +@_linux_netns_only +@pytest.mark.parametrize('cancel', (False, True)) +def test_default_capture_closes_on_failed_start( + cancel: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + '''The caller pin and backend duplicate unwind on failure or cancellation.''' + opened = [] + real_open = os.open + opened_process = trio.Event() + spawn_error = OSError('exec failed before child creation') + + def record_open(path, flags, *args, **kwargs): + result = real_open(path, flags, *args, **kwargs) + if path == _netns._SELF_NETNS: + opened.append(result) + return result + + async def fail_process(command, **kwargs): + fd, dev, ino = _netns_bootstrap_from_cmd(command) + assert (os.fstat(fd).st_dev, os.fstat(fd).st_ino) == (dev, ino) + opened.append(fd) + opened_process.set() + if cancel: + await trio.sleep_forever() + raise spawn_error + + async def main() -> None: + async with tractor.open_root_actor(start_method='trio'): + async with tractor.open_nursery() as nursery: + monkeypatch.setattr(_netns.os, 'open', record_open) + monkeypatch.setattr(trio.lowlevel, 'open_process', fail_process) + if cancel: + async with trio.open_nursery() as tn: + tn.start_soon(nursery.start_actor, 'cancelled-default') + await opened_process.wait() + tn.cancel_scope.cancel() + else: + with pytest.raises(OSError) as caught: + await nursery.start_actor('failed-default') + assert caught.value is spawn_error + assert not nursery._children + assert len(opened) == 2 + for fd in opened: + _assert_fd_closed(fd) + + trio.run(main) + + +def test_default_child_ignores_stale_subprocess_worker( + monkeypatch: pytest.MonkeyPatch, +) -> None: + '''An omitted bindspace follows the calling task, not a reused worker.''' + if _run_in_unshared_netns( + test_name='test_default_child_ignores_stale_subprocess_worker', + reexec_var='TRACTOR_TEST_STALE_WORKER_REEXEC', + ): + return + + target_netns_fd = os.open(_SELF_NETNS_PATH, os.O_RDONLY) + target_netns_inode = os.fstat(target_netns_fd).st_ino + # Pin the first disposable netns through `target_netns_fd`, then + # move the caller into a distinct second netns. This gives the root + # one real target to enter and one real caller netns to restore. + # Only the subprocess worker enters that first namespace now. + os.unshare(os.CLONE_NEWNET) + original_netns_inode = _SELF_NETNS_PATH.stat().st_ino + assert original_netns_inode != target_netns_inode + worker_inodes = [] + real_popen = subprocess.Popen + + def observe_popen(*args, **kwargs): + worker_inodes.append(_SELF_NETNS_PATH.stat().st_ino) + return real_popen(*args, **kwargs) + + async def main() -> None: + async with tractor.open_root_actor( + enable_transports=['uds'], + start_method='trio', + ): + # Reuse the worker after it has legitimately changed its + # thread-local netns. No event-loop thread transition occurs. + await trio.to_thread.run_sync( + os.setns, target_netns_fd, os.CLONE_NEWNET, + ) + monkeypatch.setattr(subprocess, 'Popen', observe_popen) + async with tractor.open_nursery() as nursery: + portal = await nursery.start_actor( + 'inherits-task-netns', enable_modules=[__name__], + ) + report = await portal.run(_report_spawn_scope) + assert report['identity'][1] == original_netns_inode + assert worker_inodes == [target_netns_inode] + assert _SELF_NETNS_PATH.stat().st_ino == original_netns_inode + await nursery.cancel() + + try: + trio.run(main) + assert os.fstat(target_netns_fd).st_ino == target_netns_inode + finally: + os.close(target_netns_fd) + + +def test_cross_netns_inherited_loopback_registrar_fails( + capfd: pytest.CaptureFixture[str], +) -> None: + '''A reachable UDS parent cannot make a host-loopback registrar reachable.''' + if _run_in_unshared_netns( + test_name='test_cross_netns_inherited_loopback_registrar_fails', + reexec_var='TRACTOR_TEST_UNREACHABLE_REG_REEXEC', + ): + return + + ip = shutil.which('ip') + if ip is None: + pytest.skip('iproute2 ip command is needed to enable disposable loopback') + fd = os.open(_SELF_NETNS_PATH, os.O_RDONLY) + bindspace = _bindspace_for_fd(fd) + os.unshare(os.CLONE_NEWNET) + subprocess.run([ip, 'link', 'set', 'lo', 'up'], check=True) + + async def main() -> None: + with tempfile.TemporaryDirectory(prefix='tnr-') as directory: + parent_addr = ('unix', str(Path(directory) / 'parent.sock')) + async with tractor.open_root_actor( + registry_addrs=[('tcp', '127.0.0.1', 1616)], + tpt_bind_addrs=[parent_addr], + enable_transports=['tcp'], + start_method='trio', + ): + with trio.fail_after(10): + async with tractor.open_nursery() as nursery: + portal = await nursery.start_actor( + 'unreachable-registrar', + bindspace=bindspace, + parent_addr=parent_addr, + ) + _, proc, _ = nursery._children[portal.channel.aid.uid] + # Join the failing runtime: the bootstrap handshake + # precedes registrar connection and RPC readiness. + + assert proc.returncode == 1 + assert not nursery._children + assert os.fstat(fd).st_ino == bindspace.ref.inode + + try: + trio.run(main) + error_output = capfd.readouterr().err + assert 'Network is unreachable' in error_output + assert '127.0.0.1:1616' in error_output + finally: + os.close(fd) diff --git a/tractor/_child.py b/tractor/_child.py index 3c5a1c30..cced9379 100644 --- a/tractor/_child.py +++ b/tractor/_child.py @@ -49,11 +49,11 @@ def parse_ipaddr(arg): return arg -def parse_netns_bootstrap(arg: str) -> tuple[int, int]: +def parse_netns_bootstrap(arg: str) -> tuple[int, int, int]: ''' Parse one atomic inherited namespace capability. - Descriptor and inode validation remains in + Descriptor and device/inode validation remains in `_consume_netns_bootstrap()` so every valid descriptor-shaped input reaches its exact child-owned cleanup boundary. @@ -62,16 +62,16 @@ def parse_netns_bootstrap(arg: str) -> tuple[int, int]: value: object = literal_eval(arg) except (ValueError, SyntaxError) as exc: raise argparse.ArgumentTypeError( - 'netns bootstrap must be an `(fd, inode)` tuple' + 'netns bootstrap must be an `(fd, device, inode)` tuple' ) from exc if ( not isinstance(value, tuple) or - len(value) != 2 + len(value) != 3 ): raise argparse.ArgumentTypeError( - 'netns bootstrap must be an `(fd, inode)` tuple' + 'netns bootstrap must be an `(fd, device, inode)` tuple' ) return value @@ -83,7 +83,7 @@ def _actor_child_main( parent_addr: UnwrappedAddress | None, infect_asyncio: bool, spawn_method: SpawnMethodKey = 'trio', - netns_bootstrap: tuple[int, int]|None = None, + netns_bootstrap: tuple[int, int, int]|None = None, ) -> None: ''' diff --git a/tractor/_root.py b/tractor/_root.py index 5c48f42e..a52b5f6c 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -20,7 +20,6 @@ Root actor runtime ignition(s). ''' from __future__ import annotations -from collections.abc import AsyncIterator from contextlib import ( asynccontextmanager as acm, ) @@ -34,7 +33,6 @@ import sys from typing import ( Any, Callable, - TYPE_CHECKING, ) import warnings @@ -70,9 +68,6 @@ from ._exceptions import ( RuntimeFailure, ) -if TYPE_CHECKING: - from .net._bindspace import Bindspace - logger = log.get_logger('tractor') @@ -163,29 +158,13 @@ async def maybe_block_bp( os.environ.pop('PYTHONBREAKPOINT', None) -@acm -async def _enter_root_bindspace( - bindspace: Bindspace|None, -) -> AsyncIterator[None]: - ''' - Adapt synchronous root netns entry to the outer async lifecycle. - - The wrapped context has no checkpoints, so `trio` cancellation - cannot interrupt thread-local namespace restoration. - - ''' - from .spawn._netns import _enter_netns_temporarily - - with _enter_netns_temporarily(bindspace): - yield - - @acm async def open_root_actor( *, # Low-level realized scope. A future tunnelled-address bootstrap # may open and supply this capability internally. - bindspace: Bindspace|None = None, + # Root netns entry is deferred to a pre-Trio launcher; the async + # root API cannot safely switch the process networking scope. tpt_bind_addrs: list[ Address # concrete transport address case @@ -250,9 +229,9 @@ async def open_root_actor( All (disjoint) actor-process-trees-as-programs are created via this entrypoint. - When `bindspace` is provided, enter its network namespace before - any registry or IPC activity and restore the calling thread's - original namespace after complete actor teardown. + To supervise a network-scoped child, open its bindspace outside + an ActorNursery and pass it to `start_actor()`. Root namespace + entry must happen before `trio.run()` and worker-thread creation. ''' # XXX NEVER allow nested actor-trees! @@ -274,25 +253,8 @@ async def open_root_actor( f'_registry_addrs: {registry_addrs!r}\n' ) - effective_start_method: str = ( - os.environ.get('TRACTOR_SPAWN_METHOD') - or start_method - or _spawn._spawn_method - ) - if ( - bindspace is not None - and - effective_start_method == 'mp_forkserver' - ): - raise NotImplementedError( - 'Root actor bindspaces are not supported by the ' - '`mp_forkserver` spawn backend because a persistent ' - 'forkserver may retain its original network namespace!' - ) - # debug.mk_pdb().set_trace() async with ( - _enter_root_bindspace(bindspace), maybe_block_bp( debug_mode=debug_mode, maybe_enable_greenback=maybe_enable_greenback, diff --git a/tractor/net/_bindspace.py b/tractor/net/_bindspace.py index 0c95613e..5115a051 100644 --- a/tractor/net/_bindspace.py +++ b/tractor/net/_bindspace.py @@ -196,7 +196,7 @@ class Bindspace( ''' spec: BindspaceSpec ref: BindspaceRef - namespace_fd: int|None + namespace_fd: int ownership: BindspaceOwnership def __post_init__(self) -> None: @@ -206,7 +206,7 @@ class Bindspace( ''' spec: BindspaceSpec = self.spec ref: BindspaceRef = self.ref - namespace_fd: int|None = self.namespace_fd + namespace_fd: int = self.namespace_fd ownership: BindspaceOwnership = self.ownership if spec.kind != ref.kind: @@ -237,21 +237,20 @@ class Bindspace( f'`BindspaceSpec.lifecycle={spec.lifecycle!r}` ' f'requires ownership={expected_ownership!r}!' ) - if namespace_fd is not None: - if ( - type(namespace_fd) is not int - or - namespace_fd < 0 - ): - raise ValueError( - '`namespace_fd` must be non-negative or `None`!' - ) - fd_inode: int = os.fstat(namespace_fd).st_ino - if ref.inode != fd_inode: - raise ValueError( - f'Namespace FD inode {fd_inode} does not match ' - f'reference inode {ref.inode}!' - ) + if ( + type(namespace_fd) is not int + or + namespace_fd < 0 + ): + raise ValueError( + '`namespace_fd` must be a non-negative `int`!' + ) + fd_inode: int = os.fstat(namespace_fd).st_ino + if ref.inode != fd_inode: + raise ValueError( + f'Namespace FD inode {fd_inode} does not match ' + f'reference inode {ref.inode}!' + ) def __repr__(self) -> str: ''' diff --git a/tractor/net/_tunnel.py b/tractor/net/_tunnel.py index c76e64eb..b933f4ed 100644 --- a/tractor/net/_tunnel.py +++ b/tractor/net/_tunnel.py @@ -479,7 +479,7 @@ def _sync_create_wg_iface( '`tractor[wg]` extra.' ) from exc - namespace_fd: int|None = bindspace.namespace_fd + namespace_fd: int = bindspace.namespace_fd ipr: Any = IPRoute( netns=namespace_fd, flags=0, diff --git a/tractor/runtime/_supervise.py b/tractor/runtime/_supervise.py index bf13fbda..eb9e9f5b 100644 --- a/tractor/runtime/_supervise.py +++ b/tractor/runtime/_supervise.py @@ -18,12 +18,16 @@ ``trio`` inspired apis and helpers """ -from contextlib import asynccontextmanager as acm +from contextlib import ( + asynccontextmanager as acm, + ExitStack, +) from functools import partial from typing import ( TYPE_CHECKING, ) import typing +import sys import warnings import trio @@ -36,6 +40,7 @@ from ..devx import ( from ..discovery._addr import ( UnwrappedAddress, mk_uuid, + wrap_address, ) from ._state import ( current_actor, @@ -74,6 +79,22 @@ if TYPE_CHECKING: log = get_logger() +def _validate_spawn_address(addr: UnwrappedAddress) -> None: + ''' + Require the canonical tagged address boundary for spawn routing. + + ''' + valid = False + if type(addr) is tuple: + match addr: + case ('tcp', str() as host, port): + valid = bool(host) and type(port) is int and 0 < port < 65536 + case ('unix', str() as path): + valid = bool(path) and '\0' not in path + if not valid or wrap_address(addr).unwrap() != addr: + raise ValueError(f'Expected a canonical UnwrappedAddress, got {addr!r}') + + async def _try_cancel_then_kill( portal: Portal, # `ProcessType` is `TYPE_CHECKING`-only (defined under that @@ -421,6 +442,8 @@ class ActorNursery: bind_addrs: list[UnwrappedAddress]|None = None, bindspace: 'Bindspace|None' = None, + parent_addr: UnwrappedAddress|None = None, + registry_addrs: list[UnwrappedAddress]|None = None, rpc_module_paths: list[str]|None = None, enable_transports: list[str] = [_state._def_tpt_proto], enable_modules: list[str]|None = None, @@ -435,6 +458,12 @@ class ActorNursery: Start a (daemon) actor: an process that has no designated "main task" besides the runtime. + `parent_addr` selects one canonical live supervisor listener; + `registry_addrs` replaces registrar inheritance for this child. + Both must remain reachable from the child's target netns. A + filesystem UDS is the usual cross-netns bootstrap transport. + Omitted `bindspace` on Linux/Trio pins the calling task's netns. + Pass ``inherit_parent_main=False`` to keep this child on its own bootstrap module for the trio spawn backend instead of applying the parent ``__main__`` re-exec fixup during startup. @@ -478,6 +507,27 @@ class ActorNursery: ) enable_modules.extend(rpc_module_paths) + if parent_addr is None: + parent_addr = self._actor.accept_addr + else: + _validate_spawn_address(parent_addr) + if parent_addr not in self._actor.accept_addrs: + raise ValueError( + f'parent_addr {parent_addr!r} must name a live ' + 'supervisor listener' + ) + assert parent_addr + + if registry_addrs is None: + selected_reg_addrs = list(current_actor().reg_addrs) + else: + if type(registry_addrs) is not list or not registry_addrs: + raise ValueError('registry_addrs must be a non-empty list') + for addr in registry_addrs: + _validate_spawn_address(addr) + selected_reg_addrs = list(registry_addrs) + _rtv['_registry_addrs'] = list(selected_reg_addrs) + subactor = Actor( name=name, uuid=mk_uuid(), @@ -488,29 +538,41 @@ class ActorNursery: inherit_parent_main=inherit_parent_main, # verbatim relay this actor's registrar addresses - registry_addrs=current_actor().registry_addrs, + # Populate child-local state below: Actor construction with + # registry_addrs publishes into this process's runtime vars. ) - parent_addr: UnwrappedAddress = self._actor.accept_addr - assert parent_addr + subactor.reg_addrs = selected_reg_addrs # start a task to spawn a process # blocks until process has been started and a portal setup # XXX: the type ignore is actually due to a `mypy` bug - return await self._da_nursery.start( # type: ignore - partial( - _spawn.new_proc, - name, - self, - subactor, - self.errors, - bind_addrs, - parent_addr, - _rtv, # run time vars - bindspace=bindspace, - infect_asyncio=infect_asyncio, - proc_kwargs=proc_kwargs + with ExitStack() as scope: + if ( + bindspace is None + and sys.platform == 'linux' + and _spawn._spawn_method == 'trio' + ): + from ..spawn._netns import pin_current_netns + + # Capture on the calling actor task before nursery.start() + # checkpoints or any subprocess worker can choose a netns. + bindspace = scope.enter_context(pin_current_netns()) + + return await self._da_nursery.start( # type: ignore + partial( + _spawn.new_proc, + name, + self, + subactor, + self.errors, + bind_addrs, + parent_addr, + _rtv, # run time vars + bindspace=bindspace, + infect_asyncio=infect_asyncio, + proc_kwargs=proc_kwargs + ) ) - ) # @api_frame async def cancel( diff --git a/tractor/spawn/_entry.py b/tractor/spawn/_entry.py index 0ef7aceb..f0a91ef3 100644 --- a/tractor/spawn/_entry.py +++ b/tractor/spawn/_entry.py @@ -59,7 +59,7 @@ log = get_logger() def _consume_netns_bootstrap( - netns_bootstrap: tuple[int, int]|None, + netns_bootstrap: tuple[int, int, int]|None, ) -> int|None: ''' Enter and release one child-owned network namespace capability. @@ -73,8 +73,9 @@ def _consume_netns_bootstrap( return None namespace_fd: int + expected_device: int expected_inode: int - namespace_fd, expected_inode = netns_bootstrap + namespace_fd, expected_device, expected_inode = netns_bootstrap if ( type(namespace_fd) is not int or @@ -84,12 +85,14 @@ def _consume_netns_bootstrap( # never pass bool/non-int/negative values to `os.close()`. return enter_netns( namespace_fd, + expected_device, expected_inode, ) try: entered_inode: int = enter_netns( namespace_fd, + expected_device, expected_inode, ) except BaseException as entry_error: @@ -114,7 +117,7 @@ def _mp_main( start_method: SpawnMethodKey, parent_addr: UnwrappedAddress | None = None, infect_asyncio: bool = False, - netns_bootstrap: tuple[int, int]|None = None, + netns_bootstrap: tuple[int, int, int]|None = None, ) -> None: ''' diff --git a/tractor/spawn/_netns.py b/tractor/spawn/_netns.py index 9c3d0f00..2e0204fe 100644 --- a/tractor/spawn/_netns.py +++ b/tractor/spawn/_netns.py @@ -47,6 +47,7 @@ _SELF_NETNS: Path = Path('/proc/thread-self/ns/net') def enter_netns( namespace_fd: int, + expected_device: int, expected_inode: int, ) -> int: ''' @@ -76,6 +77,22 @@ def enter_netns( '`expected_inode` must be a positive `int`!' ) + if type(expected_device) is not int or expected_device < 0: + raise ValueError('`expected_device` must be a non-negative `int`!') + + target = os.fstat(namespace_fd) + expected = (expected_device, expected_inode) + if (target.st_dev, target.st_ino) != expected: + raise ValueError( + f'Inherited namespace FD identity ' + f'{(target.st_dev, target.st_ino)} does not ' + f'match expected identity {expected}!' + ) + + current = _SELF_NETNS.stat() + if (current.st_dev, current.st_ino) == expected: + return expected_inode + setns: Callable[[int, int], None]|None = getattr( os, 'setns', @@ -95,13 +112,6 @@ def enter_netns( 'Python has no Linux network namespace entry support!' ) - inherited_inode: int = os.fstat(namespace_fd).st_ino - if inherited_inode != expected_inode: - raise ValueError( - f'Inherited namespace FD inode {inherited_inode} does not ' - f'match expected inode {expected_inode}!' - ) - try: setns(namespace_fd, clone_newnet) except OSError as exc: @@ -110,11 +120,13 @@ def enter_netns( f'{expected_inode}!' ) from exc - entered_inode: int = _SELF_NETNS.stat().st_ino - if entered_inode != expected_inode: + entered = _SELF_NETNS.stat() + entered_inode: int = entered.st_ino + if (entered.st_dev, entered_inode) != expected: raise RuntimeError( - f'Entered network namespace inode {entered_inode} does not ' - f'match expected inode {expected_inode}!' + f'Entered network namespace identity ' + f'{(entered.st_dev, entered_inode)} does not ' + f'match expected identity {expected}!' ) return entered_inode @@ -181,8 +193,9 @@ def _enter_netns_temporarily( ''' Enter a root bindspace and restore the caller thread's netns. - `_root._enter_root_bindspace()` adapts this synchronous scope to - the root actor's async lifecycle. + Internal synchronous primitive only. Never surround a Trio + checkpoint with this scope: entry changes only the calling thread. + The former async root adapter has been removed. Only descriptors opened or duplicated by this context are used for validation, entry and restoration. Since `setns()` is @@ -199,12 +212,7 @@ def _enter_netns_temporarily( 'Network namespace entry is Linux-only!' ) - namespace_fd: int|None = bindspace.namespace_fd - if namespace_fd is None: - raise ValueError( - '`bindspace.namespace_fd` must be a live netns FD for ' - 'root actor entry!' - ) + namespace_fd: int = bindspace.namespace_fd if ( type(namespace_fd) is not int or @@ -250,6 +258,7 @@ def _enter_netns_temporarily( if restore_needed: enter_netns( tgt_fd, + tgt_stat.st_dev, tgt_inode, ) @@ -260,6 +269,7 @@ def _enter_netns_temporarily( try: enter_netns( orig_fd, + orig_stat.st_dev, orig_inode, ) except BaseException as restore_error: @@ -274,6 +284,7 @@ def _enter_netns_temporarily( try: enter_netns( orig_fd, + orig_stat.st_dev, orig_inode, ) except BaseException as restore_error: @@ -282,3 +293,25 @@ def _enter_netns_temporarily( 'namespace during root netns cleanup.' ) raise restore_error + + +@cm +def pin_current_netns() -> Iterator[Bindspace]: + ''' + Pin the calling task's netns synchronously, without entering it. + + ''' + from ..net._bindspace import Bindspace, BindspaceRef, BindspaceSpec + + namespace_fd = os.open(_SELF_NETNS, os.O_RDONLY | os.O_CLOEXEC) + with close_fd(namespace_fd, 'calling task'): + yield Bindspace( + spec=BindspaceSpec(kind='netns'), + ref=BindspaceRef( + kind='netns', + key=None, + inode=os.fstat(namespace_fd).st_ino, + ), + namespace_fd=namespace_fd, + ownership='borrowed', + ) diff --git a/tractor/spawn/_trio.py b/tractor/spawn/_trio.py index c69418b0..c74ff5b7 100644 --- a/tractor/spawn/_trio.py +++ b/tractor/spawn/_trio.py @@ -124,11 +124,7 @@ async def trio_proc( child_netns_fd: int|None = None if bindspace is not None: - if (namespace_fd := bindspace.namespace_fd) is None: - raise ValueError( - '`bindspace.namespace_fd` is required for ' - 'Trio child transport!' - ) + namespace_fd: int = bindspace.namespace_fd # Snapshot caller-owned process options before duplicating the # live `Bindspace.namespace_fd`. No checkpoint separates this @@ -139,10 +135,12 @@ async def trio_proc( proc_kwargs = dict(proc_kwargs) child_netns_fd = os.dup(namespace_fd) try: - netns_bootstrap: tuple[int, int] = ( + target_stat = os.fstat(child_netns_fd) + netns_bootstrap: tuple[int, int, int] = ( # FD number retained in the child's descriptor table. child_netns_fd, # Namespace identity checked before the child enters it. + target_stat.st_dev, bindspace.ref.inode, ) spawn_cmd.extend((