from __future__ import annotations import json import os import subprocess import sys import threading import time from pathlib import Path import pytest from k1link.device_plugins.xgrids_k1.application_control_process_lease import ( ApplicationControlProcessLease, ApplicationControlProcessLeaseUnavailable, ) from k1link.host_network import wifi TEST_PASSWORD = "fixture-only-network-secret" TEST_PROFILE_ID = "fixture.quick-connect.v1" TEST_CREDENTIAL_SOURCE_ID = "fixture.firmware-provider.v1" def _helper(tmp_path: Path) -> Path: helper = tmp_path / "associate_wifi.swift" helper.write_text("// offline fixture\n", encoding="utf-8") return helper def _private_process_lease( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> tuple[Path, ApplicationControlProcessLease]: repository_root = tmp_path / "repository" repository_root.mkdir() monkeypatch.setenv("MISSIONCORE_DATA_DIR", str(tmp_path / "private-data")) return repository_root, ApplicationControlProcessLease.acquire(repository_root) def test_association_identity_is_opaque_process_scoped_evidence( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") captured: dict[str, object] = {} live_inputs: list[bytearray] = [] def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: assert isinstance(kwargs["input"], bytearray) live_inputs.append(kwargs["input"]) captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])}) return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":true,' b'"association_identity":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' b'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","association_evidence":"ssid+bssid"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=fake_runner, continuity_key=bytes(range(32)), ) result = probe.observe("en0") assert result == { "schema_version": 1, "adapter": "CoreWLAN", "wifi_interface": True, "association_state": "associated", "evidence_quality": "ssid+bssid", "continuity_proven": True, "continuity_token": "a" * 64, "reason_code": None, } request = json.loads(bytes(captured["input"]).decode("utf-8")) assert request == { "action": "inspect-association", "profile_id": "host-association-inspection.v1", "interface_name": "en0", "continuity_key_hex": bytes(range(32)).hex(), } assert "continuity_key_hex" not in " ".join(captured["argv"]) assert "ssid" not in request assert "bssid" not in request assert live_inputs and not any(live_inputs[0]) def test_association_identity_changes_an_otherwise_identical_route_fingerprint() -> None: base_route = "c" * 64 first: wifi.HostWifiAssociationIdentityResult = { "schema_version": 1, "adapter": "CoreWLAN", "wifi_interface": True, "association_state": "associated", "evidence_quality": "bssid-only", "continuity_proven": True, "continuity_token": "a" * 64, "reason_code": None, } second: wifi.HostWifiAssociationIdentityResult = { **first, "continuity_token": "b" * 64, } assert wifi.bind_route_fingerprint_to_wifi_association( base_route, first ) != wifi.bind_route_fingerprint_to_wifi_association(base_route, second) def test_association_identity_reuses_only_a_subsecond_read_only_observation( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") clocks = {"monotonic": 100.0, "wall": 1_000.0} calls = 0 def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 identity = "a" * 64 if calls == 1 else "b" * 64 return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":true,' + f'"association_identity":"{identity}",'.encode() + b'"association_evidence":"ssid+bssid"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=fake_runner, continuity_key=b"c" * 32, monotonic_clock=lambda: clocks["monotonic"], wall_clock=lambda: clocks["wall"], ) first = probe.observe("en0") first["continuity_token"] = "f" * 64 clocks["monotonic"] += 0.74 clocks["wall"] += 0.74 cached = probe.observe("en0") assert calls == 1 assert cached["continuity_token"] == "a" * 64 clocks["monotonic"] += 0.02 clocks["wall"] += 0.02 refreshed = probe.observe("en0") assert calls == 2 assert refreshed["continuity_token"] == "b" * 64 def test_association_identity_timeout_includes_waiting_for_an_inflight_observer( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") runner_entered = threading.Event() release_runner = threading.Event() first_result: list[wifi.HostWifiAssociationIdentityResult] = [] runner_calls = 0 def blocked_runner( argv: list[str], **_: object, ) -> subprocess.CompletedProcess[bytes]: nonlocal runner_calls runner_calls += 1 runner_entered.set() assert release_runner.wait(timeout=2.0) return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":true,' b'"association_identity":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' b'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","association_evidence":"bssid-only"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=blocked_runner, continuity_key=b"t" * 32, max_cache_age_seconds=0, ) owner = threading.Thread( target=lambda: first_result.append( probe.observe("en0", timeout_seconds=1.0) ) ) owner.start() assert runner_entered.wait(timeout=1.0) started = time.monotonic() timed_out = probe.observe("en0", timeout_seconds=0.05) elapsed = time.monotonic() - started assert elapsed < 0.25 assert timed_out["continuity_proven"] is False assert timed_out["reason_code"] == "host-wifi-operation-timeout" assert runner_calls == 1 release_runner.set() owner.join(timeout=2.0) assert owner.is_alive() is False assert first_result[0]["continuity_proven"] is True def test_association_identity_interface_transition_is_an_immediate_cache_barrier( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") calls = 0 def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 identity = format(calls, "x") * 64 return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":true,' + f'"association_identity":"{identity}",'.encode() + b'"association_evidence":"ssid+bssid"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=fake_runner, continuity_key=b"d" * 32, monotonic_clock=lambda: 100.0, wall_clock=lambda: 1_000.0, ) first = probe.observe("en0") second = probe.observe("en1") third = probe.observe("en0") assert calls == 3 assert first["continuity_token"] == "1" * 64 assert second["continuity_token"] == "2" * 64 assert third["continuity_token"] == "3" * 64 def test_association_identity_wall_clock_sleep_expires_cache_fail_closed( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") clocks = {"monotonic": 100.0, "wall": 1_000.0} calls = 0 def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":true,' b'"association_identity":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' b'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","association_evidence":"bssid-only"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=fake_runner, continuity_key=b"e" * 32, monotonic_clock=lambda: clocks["monotonic"], wall_clock=lambda: clocks["wall"], ) probe.observe("en0") clocks["wall"] += 10.0 probe.observe("en0") assert calls == 2 @pytest.mark.parametrize("max_cache_age_seconds", [-0.01, 0.751, float("nan")]) def test_association_identity_cache_age_cannot_exceed_safety_bound( tmp_path: Path, max_cache_age_seconds: float, ) -> None: with pytest.raises(ValueError, match="between 0 and 0.75 seconds"): wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), max_cache_age_seconds=max_cache_age_seconds, ) def test_unavailable_association_identity_keeps_one_process_scoped_route_token( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]: return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":true,' b'"association_evidence":"unavailable",' b'"reason_code":"association-identity-unavailable"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=fake_runner, continuity_key=b"x" * 32, ) first = probe.observe("en0") second = probe.observe("en0") assert first["continuity_proven"] is False assert first["association_state"] == "unavailable" assert first["reason_code"] == "association-identity-unavailable" assert first["continuity_token"] == second["continuity_token"] assert wifi.bind_route_fingerprint_to_wifi_association( "route", first ) == wifi.bind_route_fingerprint_to_wifi_association("route", second) def test_unavailable_association_identity_changes_with_interface_scope( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]: return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":true,' b'"association_evidence":"unavailable",' b'"reason_code":"association-identity-unavailable"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=fake_runner, continuity_key=b"z" * 32, ) first = probe.observe("en0") second = probe.observe("en1") assert first["continuity_proven"] is False assert second["continuity_proven"] is False assert first["continuity_token"] != second["continuity_token"] def test_non_wifi_interface_has_stable_secret_free_continuity_evidence( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]: return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","wifi_interface":false,' b'"association_identity":"dddddddddddddddddddddddddddddddd' b'dddddddddddddddddddddddddddddddd","association_evidence":"not-wifi"}' ), stderr=b"", ) probe = wifi.HostWifiAssociationIdentityProbe( _helper(tmp_path), runner=fake_runner, continuity_key=b"y" * 32, ) first = probe.observe("en7") second = probe.observe("en7") assert first == second assert first["wifi_interface"] is False assert first["association_state"] == "not-wifi" assert first["continuity_proven"] is True def test_association_exposes_only_profile_id_and_expected_ssid_to_platform_helper( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") calls: list[dict[str, object]] = [] live_inputs: list[bytearray] = [] def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: assert isinstance(kwargs["input"], bytearray) live_inputs.append(kwargs["input"]) calls.append({"argv": argv, **kwargs, "input": bytes(kwargs["input"])}) return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","already_associated":false,' b'"profile_enrolled":true,"scan_attempt_count":3,' b'"scan_elapsed_ms":1840,"credential_source":"exact-firmware-profile"}' ), stderr=b"", ) helper = _helper(tmp_path) result = wifi.associate_with_wifi_profile_once( helper, TEST_PROFILE_ID, "XGR-OFFLINE", runner=fake_runner, ) assert result == { "schema_version": 1, "adapter": "CoreWLAN", "outcome": "associated", "already_associated": False, "profile_enrolled": True, "scan_attempt_count": 3, "scan_elapsed_ms": 1840, "credential_source": "exact-firmware-profile", } assert len(calls) == 1 call = calls[0] assert call["argv"] == [ "/usr/bin/xcrun", "swift", str(helper.resolve()), ] assert "swiftc" not in call["argv"] request = json.loads(bytes(call["input"]).decode("utf-8")) assert request == { "action": "associate", "profile_id": TEST_PROFILE_ID, "ssid": "XGR-OFFLINE", "scan_timeout_seconds": 30.0, } assert call["check"] is False assert call["timeout"] == 180.0 assert live_inputs and not any(live_inputs[0]) def test_ephemeral_bridge_association_uses_stdin_and_does_not_enroll_profile( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") captured: dict[str, object] = {} live_inputs: list[bytearray] = [] def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: assert isinstance(kwargs["input"], bytearray) live_inputs.append(kwargs["input"]) captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])}) return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"CoreWLAN","already_associated":false,' b'"profile_enrolled":false,"scan_attempt_count":2,' b'"scan_elapsed_ms":910,"credential_source":"operation-memory"}' ), stderr=b"", ) result = wifi.associate_with_ephemeral_wifi_credentials_once( _helper(tmp_path), "DCEXPRESS", TEST_PASSWORD, runner=fake_runner, ) assert result == { "schema_version": 1, "adapter": "CoreWLAN", "outcome": "associated", "already_associated": False, "profile_enrolled": False, "scan_attempt_count": 2, "scan_elapsed_ms": 910, "credential_source": "operation-memory", } request = json.loads(bytes(captured["input"]).decode("utf-8")) assert request == { "action": "associate-ephemeral", "profile_id": "bridge-operation-memory.v1", "ssid": "DCEXPRESS", "password": TEST_PASSWORD, "scan_timeout_seconds": 30.0, } assert TEST_PASSWORD not in " ".join(captured["argv"]) assert "env" not in captured assert live_inputs and not any(live_inputs[0]) def test_profile_store_passes_secret_only_through_stdin( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") captured: dict[str, object] = {} live_inputs: list[bytearray] = [] def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: assert isinstance(kwargs["input"], bytearray) live_inputs.append(kwargs["input"]) captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])}) return subprocess.CompletedProcess( argv, 0, stdout=b'{"ok":true,"adapter":"macOS Keychain","stored":true}', stderr=b"", ) result = wifi.store_wifi_profile_once( _helper(tmp_path), TEST_PROFILE_ID, "XGR-OFFLINE", TEST_PASSWORD, runner=fake_runner, ) assert result["outcome"] == "stored" assert TEST_PASSWORD not in " ".join(captured["argv"]) assert "env" not in captured request = json.loads(bytes(captured["input"]).decode("utf-8")) assert request == { "action": "store-profile", "profile_id": TEST_PROFILE_ID, "ssid": "XGR-OFFLINE", "password": TEST_PASSWORD, } assert live_inputs and not any(live_inputs[0]) def test_firmware_material_store_passes_secret_only_through_stdin( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") captured: dict[str, object] = {} live_inputs: list[bytearray] = [] def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: assert isinstance(kwargs["input"], bytearray) live_inputs.append(kwargs["input"]) captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])}) return subprocess.CompletedProcess( argv, 0, stdout=b'{"ok":true,"adapter":"macOS Keychain","stored":true}', stderr=b"", ) result = wifi.store_wifi_credential_material( _helper(tmp_path), TEST_CREDENTIAL_SOURCE_ID, TEST_PASSWORD, runner=fake_runner, ) assert result["outcome"] == "stored" assert TEST_PASSWORD not in " ".join(captured["argv"]) assert "env" not in captured request = json.loads(bytes(captured["input"]).decode("utf-8")) assert request == { "action": "store-credential-material", "profile_id": TEST_CREDENTIAL_SOURCE_ID, "password": TEST_PASSWORD, } assert live_inputs and not any(live_inputs[0]) def test_profile_is_materialized_from_opaque_firmware_source( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") captured: dict[str, object] = {} def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])}) return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"macOS Keychain","profile_available":true,' b'"profile_enrolled":true,"credential_source":"exact-firmware-profile"}' ), stderr=b"", ) result = wifi.ensure_wifi_profile_from_credential_source( _helper(tmp_path), TEST_PROFILE_ID, "XGR-OFFLINE", TEST_CREDENTIAL_SOURCE_ID, runner=fake_runner, ) assert result == { "schema_version": 1, "adapter": "macOS Keychain", "available": True, "profile_enrolled": True, "credential_source": "exact-firmware-profile", } request = json.loads(bytes(captured["input"]).decode("utf-8")) assert request == { "action": "ensure-profile", "profile_id": TEST_PROFILE_ID, "ssid": "XGR-OFFLINE", "credential_source_id": TEST_CREDENTIAL_SOURCE_ID, } assert TEST_PASSWORD not in bytes(captured["input"]).decode("utf-8") @pytest.mark.parametrize("available", [True, False]) def test_profile_preflight_checks_only_the_expected_keychain_item( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, available: bool, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") captured: dict[str, object] = {} def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: captured.update({"argv": argv, **kwargs, "input": bytes(kwargs["input"])}) encoded_available = b"true" if available else b"false" return subprocess.CompletedProcess( argv, 0, stdout=( b'{"ok":true,"adapter":"macOS Keychain","profile_available":' + encoded_available + ( b',"credential_source":"exact-firmware-profile"}' if available else b"}" ) ), stderr=b"", ) result = wifi.check_wifi_profile( _helper(tmp_path), TEST_PROFILE_ID, "XGR-OFFLINE", runner=fake_runner, ) assert result == { "schema_version": 1, "adapter": "macOS Keychain", "available": available, "credential_source": ( "exact-firmware-profile" if available else None ), } request = json.loads(bytes(captured["input"]).decode("utf-8")) assert request == { "action": "check-profile", "profile_id": TEST_PROFILE_ID, "ssid": "XGR-OFFLINE", } def test_association_reports_only_sanitized_helper_reason( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]: return subprocess.CompletedProcess( argv, 1, stdout=( b'{"ok":false,"reason_code":"network-not-found",' b'"scan_attempt_count":4,"scan_elapsed_ms":15021}' ), stderr=f"private diagnostic {TEST_PASSWORD}".encode(), ) with pytest.raises(wifi.HostWifiProfileError, match="network-not-found") as raised: wifi.associate_with_wifi_profile_once( _helper(tmp_path), TEST_PROFILE_ID, "XGR-OFFLINE", runner=fake_runner, ) assert TEST_PASSWORD not in str(raised.value) assert raised.value.scan_attempt_count == 4 assert raised.value.scan_elapsed_ms == 15021 def test_association_reports_operator_timeout_separately_from_missing_helper( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") def timed_out_runner( argv: list[str], **_: object ) -> subprocess.CompletedProcess[bytes]: raise subprocess.TimeoutExpired(argv, timeout=180.0) with pytest.raises( wifi.HostWifiProfileError, match="host-wifi-operation-timeout", ) as raised: wifi.associate_with_wifi_profile_once( _helper(tmp_path), TEST_PROFILE_ID, "XGR-OFFLINE", runner=timed_out_runner, ) assert raised.value.reason_code == "host-wifi-operation-timeout" assert raised.value.helper_stage is None assert raised.value.helper_elapsed_ms is None @pytest.mark.skipif(os.name != "posix", reason="helper flock inheritance is POSIX-only") def test_fenced_host_helper_child_keeps_lifecycle_lock_after_parent_crash( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: repository_root, owner = _private_process_lease(tmp_path, monkeypatch) ready_path = tmp_path / "host-helper-ready" release_path = tmp_path / "host-helper-release" outcome: list[subprocess.CompletedProcess[bytes] | BaseException] = [] parent_descriptor_closed = False def run_helper() -> None: try: outcome.append( wifi._run_macos_helper_with_process_fence( # noqa: SLF001 [ sys.executable, "-c", ( "import pathlib,sys,time\n" "ready=pathlib.Path(sys.argv[1])\n" "release=pathlib.Path(sys.argv[2])\n" "ready.write_text('ready')\n" "while not release.exists(): time.sleep(0.01)\n" ), str(ready_path), str(release_path), ], request_bytes=bytearray(b"{}"), timeout_seconds=5.0, process_fence_descriptor_factory=( owner.duplicate_descriptor_for_child ), ) ) except BaseException as exc: # pragma: no cover - asserted below outcome.append(exc) worker = threading.Thread(target=run_helper, daemon=True) worker.start() try: deadline = time.monotonic() + 5.0 while not ready_path.exists() and time.monotonic() < deadline: time.sleep(0.01) assert ready_path.exists() os.close(owner._descriptor) # noqa: SLF001 owner._released = True # noqa: SLF001 parent_descriptor_closed = True with pytest.raises(ApplicationControlProcessLeaseUnavailable): ApplicationControlProcessLease.acquire(repository_root) release_path.touch() worker.join(timeout=5.0) assert not worker.is_alive() assert len(outcome) == 1 assert isinstance(outcome[0], subprocess.CompletedProcess) assert outcome[0].returncode == 0 with ApplicationControlProcessLease.acquire(repository_root): pass finally: release_path.touch(exist_ok=True) worker.join(timeout=5.0) if not parent_descriptor_closed: owner.release() @pytest.mark.skipif(os.name != "posix", reason="helper process groups are POSIX-only") def test_fenced_host_helper_timeout_kills_descendants_before_lock_release( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: repository_root, owner = _private_process_lease(tmp_path, monkeypatch) descendant_code = "import time; time.sleep(30)" group_leader_code = ( "import os,subprocess,sys,time\n" "fds=[]\n" "for fd in range(3,256):\n" " try: os.fstat(fd)\n" " except OSError: continue\n" " fds.append(fd)\n" "subprocess.Popen([sys.executable,'-c',sys.argv[1]],pass_fds=tuple(fds))\n" "time.sleep(30)\n" ) try: with pytest.raises(subprocess.TimeoutExpired): wifi._run_macos_helper_with_process_fence( # noqa: SLF001 [sys.executable, "-c", group_leader_code, descendant_code], request_bytes=bytearray(b"{}"), timeout_seconds=0.2, process_fence_descriptor_factory=( owner.duplicate_descriptor_for_child ), ) finally: owner.release() # A surviving descendant inherited the lifecycle descriptor deliberately. # Successful reacquisition therefore proves killpg + reap completed before # the mutating helper returned its timeout. with ApplicationControlProcessLease.acquire(repository_root): pass def test_association_reports_an_unavailable_helper_separately_from_timeout( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") def unavailable_runner( _: list[str], **__: object ) -> subprocess.CompletedProcess[bytes]: raise OSError("offline fixture") with pytest.raises( wifi.HostWifiProfileError, match="host-wifi-helper-unavailable", ) as raised: wifi.associate_with_wifi_profile_once( _helper(tmp_path), TEST_PROFILE_ID, "XGR-OFFLINE", runner=unavailable_runner, ) assert raised.value.reason_code == "host-wifi-helper-unavailable" def test_missing_helper_is_rejected_before_starting_a_process( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "darwin") missing_helper = tmp_path / "missing-associate-wifi.swift" runner_called = False def unexpected_runner( _: list[str], **__: object ) -> subprocess.CompletedProcess[bytes]: nonlocal runner_called runner_called = True raise AssertionError("missing helper must fail before subprocess startup") with pytest.raises( wifi.HostWifiProfileError, match="host-wifi-helper-missing", ) as raised: wifi.check_wifi_profile( missing_helper, TEST_PROFILE_ID, "XGR-OFFLINE", runner=unexpected_runner, ) assert raised.value.reason_code == "host-wifi-helper-missing" assert runner_called is False def test_swift_helper_source_forbids_interactive_runtime_password_fallbacks() -> None: source_path = ( Path(__file__).resolve().parents[1] / "plugins" / "xgrids-k1" / "macos" / "associate_wifi.swift" ) source = source_path.read_text(encoding="utf-8") for forbidden in ( "import AppKit", "CWKeychainFindWiFiPassword", "loadSystemWiFiProfile", "promptForDevicePassword", "NSApplication.shared", "NSAlert", '"native-secure-prompt"', '"system-wifi-keychain"', ): assert forbidden not in source metadata_check = source[ source.index("private func keychainItemExists") : source.index( "private func keychainReasonCode" ) ] assert "kSecReturnAttributes as String] = true" in metadata_check assert "kSecReturnData" not in metadata_check assert "nonInteractiveAuthenticationContext()" in metadata_check assert "context.interactionNotAllowed = true" in source assert "interactionAllowed: false" in source assert "coreWLANReasonCode(error)" in source assert 'return "corewlan-authorization-denied"' in source assert 'return "host-wifi-operation-timeout"' in source ensure_profile = source[ source.index('if request.action == "ensure-profile"') : source.index( 'if request.action == "check-profile"' ) ] assert "profile.ssid == ssid" in ensure_profile assert 'profile.credentialSource == "exact-firmware-profile"' in ensure_profile assert "interactionAllowed: false" in ensure_profile check_profile = source[ source.index('if request.action == "check-profile"') : source.index( 'guard request.action == "associate"' ) ] assert "profile.ssid == expectedSSID" in check_profile assert 'profile.credentialSource == "exact-firmware-profile"' in check_profile assert "interactionAllowed: false" in check_profile association_inspection = source[ source.index('if request.action == "inspect-association"') : source.index( 'if request.action == "store-profile"' ) ] assert "interface.ssid()" in association_inspection assert "interface.bssid()" in association_inspection assert "associationIdentity(" in association_inspection assert "ssid: currentSSID" not in association_inspection assert "bssid: currentBSSID" in association_inspection # Raw network identifiers remain local variables inside the helper. The # response contract exports only an opaque keyed identity and quality. response_contract = source[ source.index("private struct HostWifiResponse") : source.index( "private func emit" ) ] assert "let ssid" not in response_contract assert "let bssid" not in response_contract assert "associationIdentity" in response_contract assert "associationEvidence" in response_contract ephemeral_start = source.index('if request.action == "associate-ephemeral"') ephemeral_association = source[ ephemeral_start : source.index(" } else {", ephemeral_start) ] assert 'credentialSource = "operation-memory"' in ephemeral_association assert "storeProfile(" not in ephemeral_association assert "loadProfile(" not in ephemeral_association def test_swift_association_identity_uses_bssid_not_ssid_visibility() -> None: source_path = ( Path(__file__).resolve().parents[1] / "plugins" / "xgrids-k1" / "macos" / "associate_wifi.swift" ) source = source_path.read_text(encoding="utf-8") identity_function = source[ source.index("private func associationIdentity(") : source.index( "private let input" ) ] association_inspection = source[ source.index('if request.action == "inspect-association"') : source.index( 'if request.action == "store-profile"' ) ] # With the same interface+BSSID, switching between ssid+bssid and # bssid-only evidence feeds identical material to the HMAC. SSID remains a # local evidence-quality signal and never changes the continuity token. assert "mission-core/host-wifi-association/v2" in identity_function assert " ssid: String" not in identity_function assert "appendLengthPrefixed(ssid" not in identity_function assert "appendLengthPrefixed(interfaceName" in identity_function assert "appendLengthPrefixed(bssid.lowercased()" in identity_function assert "interface.ssid()" in association_inspection assert '"bssid-only"' in association_inspection assert '"ssid+bssid"' in association_inspection assert "ssid: currentSSID" not in association_inspection assert "bssid: currentBSSID" in association_inspection def test_association_rejects_an_uninstalled_platform( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(wifi.sys, "platform", "linux") with pytest.raises(wifi.HostWifiProfileError, match="unsupported-platform"): wifi.associate_with_wifi_profile_once( _helper(tmp_path), TEST_PROFILE_ID, "XGR-OFFLINE", )