Files
NODEDC_MISSION_CORE/tests/test_xgrids_macos_wifi.py
T

607 lines
20 KiB
Python

from __future__ import annotations
import hashlib
import json
import os
import subprocess
from pathlib import Path
import pytest
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, *, seed_compiled_cache: bool = True) -> Path:
helper = tmp_path / "associate_wifi.swift"
helper.write_text("// offline fixture\n", encoding="utf-8")
if seed_compiled_cache:
executable = wifi._compiled_macos_helper_path(helper)
executable.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
executable.write_bytes(b"offline compiled fixture\n")
executable.chmod(0o700)
return helper
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":"system-wifi-keychain"}'
),
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": "system-wifi-keychain",
}
assert len(calls) == 1
call = calls[0]
assert call["argv"] == [str(wifi._compiled_macos_helper_path(helper))]
request = json.loads(bytes(call["input"]).decode("utf-8"))
assert request == {
"action": "associate",
"profile_id": TEST_PROFILE_ID,
"ssid": "XGR-OFFLINE",
"scan_timeout_seconds": 15.0,
}
assert call["check"] is False
assert call["timeout"] == 180.0
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"])
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"])
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"}"
),
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,
}
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
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_cold_helper_build_uses_source_hash_and_separate_timeout(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setattr(wifi.sys, "platform", "darwin")
helper = _helper(tmp_path, seed_compiled_cache=False)
executable = wifi._compiled_macos_helper_path(helper)
calls: list[dict[str, object]] = []
def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
calls.append({"argv": argv, **kwargs})
if argv[:2] == ["/usr/bin/xcrun", "swiftc"]:
assert argv[2] == str(helper.resolve())
assert argv[3] == "-o"
assert kwargs["stdin"] == subprocess.DEVNULL
assert "input" not in kwargs
assert 0 < float(kwargs["timeout"]) <= wifi.DEFAULT_HELPER_BUILD_TIMEOUT_SECONDS
staging = Path(argv[4])
staging.write_bytes(b"compiled fixture\n")
staging.chmod(0o700)
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
assert argv == [str(executable)]
assert kwargs["timeout"] == 30.0
return subprocess.CompletedProcess(
argv,
0,
stdout=(
b'{"ok":true,"adapter":"macOS Keychain",'
b'"profile_available":true}'
),
stderr=b"",
)
result = wifi.check_wifi_profile(
helper,
TEST_PROFILE_ID,
"XGR-OFFLINE",
runner=fake_runner,
)
assert result["available"] is True
assert len(calls) == 2
assert calls[0]["argv"][:2] == ["/usr/bin/xcrun", "swiftc"]
assert calls[1]["argv"] == [str(executable)]
assert executable.is_file()
assert executable.stat().st_mode & 0o111
assert executable.name.endswith(hashlib.sha256(helper.read_bytes()).hexdigest())
assert not list(executable.parent.glob(f".{executable.name}.*.tmp"))
def test_plugin_helper_cache_is_stable_under_repository_runtime(tmp_path: Path) -> None:
helper = tmp_path / "repo" / "plugins" / "xgrids-k1" / "macos" / "associate_wifi.swift"
helper.parent.mkdir(parents=True)
helper.write_text("// offline fixture\n", encoding="utf-8")
assert wifi._helper_cache_directory(helper) == (
tmp_path / "repo" / ".runtime" / "mission-core" / "helpers"
)
def test_helper_lock_and_compile_share_one_build_deadline(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setattr(wifi.sys, "platform", "darwin")
helper = _helper(tmp_path, seed_compiled_cache=False)
monotonic_values = iter((100.0, 100.0, 104.0))
monkeypatch.setattr(wifi.time, "monotonic", lambda: next(monotonic_values))
monkeypatch.setattr(wifi.time, "monotonic_ns", lambda: 0)
lock_timeouts: list[float] = []
compile_timeouts: list[float] = []
class FakeLock:
def __enter__(self) -> None:
return None
def __exit__(self, *_args: object) -> None:
return None
def fake_lock(_path: Path, *, timeout_seconds: float) -> FakeLock:
lock_timeouts.append(timeout_seconds)
return FakeLock()
def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
if argv[:2] == ["/usr/bin/xcrun", "swiftc"]:
compile_timeouts.append(float(kwargs["timeout"]))
staging = Path(argv[4])
staging.write_bytes(b"compiled fixture\n")
staging.chmod(0o700)
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
return subprocess.CompletedProcess(
argv,
0,
stdout=b'{"ok":true,"adapter":"macOS Keychain","profile_available":true}',
stderr=b"",
)
monkeypatch.setattr(wifi, "_exclusive_helper_build_lock", fake_lock)
result = wifi._run_macos_helper(
helper,
{"action": "check-profile"},
timeout_seconds=3.0,
build_timeout_seconds=10.0,
runner=fake_runner,
)
assert result["ok"] is True
assert lock_timeouts == [10.0]
assert compile_timeouts == [6.0]
def test_compiled_helper_is_reused_without_recompiling(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setattr(wifi.sys, "platform", "darwin")
helper = _helper(tmp_path, seed_compiled_cache=False)
executable = wifi._compiled_macos_helper_path(helper)
compile_count = 0
runtime_count = 0
def fake_runner(argv: list[str], **_: object) -> subprocess.CompletedProcess[bytes]:
nonlocal compile_count, runtime_count
if argv[:2] == ["/usr/bin/xcrun", "swiftc"]:
compile_count += 1
staging = Path(argv[4])
staging.write_bytes(b"compiled fixture\n")
staging.chmod(0o700)
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
runtime_count += 1
assert argv == [str(executable)]
return subprocess.CompletedProcess(
argv,
0,
stdout=(
b'{"ok":true,"adapter":"macOS Keychain",'
b'"profile_available":true}'
),
stderr=b"",
)
for _ in range(2):
wifi.check_wifi_profile(
helper,
TEST_PROFILE_ID,
"XGR-OFFLINE",
runner=fake_runner,
)
assert compile_count == 1
assert runtime_count == 2
def test_helper_build_timeout_is_separate_from_operation_timeout(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setattr(wifi.sys, "platform", "darwin")
helper = _helper(tmp_path, seed_compiled_cache=False)
seen_timeout: float | None = None
monotonic_values = iter((1_000_000_000, 1_123_000_000))
monkeypatch.setattr(wifi.time, "monotonic_ns", lambda: next(monotonic_values))
def timed_out_compiler(
argv: list[str], **kwargs: object
) -> subprocess.CompletedProcess[bytes]:
nonlocal seen_timeout
assert argv[:2] == ["/usr/bin/xcrun", "swiftc"]
seen_timeout = float(kwargs["timeout"])
raise subprocess.TimeoutExpired(argv, timeout=seen_timeout)
with pytest.raises(
wifi.HostWifiProfileError,
match="host-wifi-helper-build-timeout",
) as raised:
wifi._run_macos_helper(
helper,
{"action": "check-profile"},
timeout_seconds=3.0,
build_timeout_seconds=7.5,
runner=timed_out_compiler,
)
assert raised.value.reason_code == "host-wifi-helper-build-timeout"
assert raised.value.helper_stage == "compile"
assert raised.value.helper_elapsed_ms == 123
assert seen_timeout is not None
assert 0 < seen_timeout <= 7.5
assert not wifi._compiled_macos_helper_path(helper).exists()
def test_helper_build_lock_contention_is_bounded(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fcntl = pytest.importorskip("fcntl")
lock_path = tmp_path / "helper.lock"
descriptor = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
fcntl.flock(descriptor, fcntl.LOCK_EX)
monotonic_values = iter((100.0, 100.02))
monkeypatch.setattr(wifi.time, "monotonic", lambda: next(monotonic_values))
try:
with (
pytest.raises(wifi.HostWifiProfileError) as raised,
wifi._exclusive_helper_build_lock(
lock_path,
timeout_seconds=0.01,
),
):
raise AssertionError("contended lock must not be acquired")
finally:
fcntl.flock(descriptor, fcntl.LOCK_UN)
os.close(descriptor)
assert raised.value.reason_code == "host-wifi-helper-build-timeout"
assert raised.value.helper_stage == "compile-lock"
assert raised.value.helper_elapsed_ms == 19
@pytest.mark.parametrize(
("compiler_failure", "expected_reason_code"),
[
("exit", "host-wifi-helper-build-failed"),
("unavailable", "host-wifi-helper-compiler-unavailable"),
],
)
def test_helper_build_errors_have_sanitized_build_taxonomy(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
compiler_failure: str,
expected_reason_code: str,
) -> None:
monkeypatch.setattr(wifi.sys, "platform", "darwin")
helper = _helper(tmp_path, seed_compiled_cache=False)
def failing_compiler(
argv: list[str], **_: object
) -> subprocess.CompletedProcess[bytes]:
assert argv[:2] == ["/usr/bin/xcrun", "swiftc"]
if compiler_failure == "unavailable":
raise OSError(f"private diagnostic {TEST_PASSWORD}")
return subprocess.CompletedProcess(
argv,
1,
stdout=b"",
stderr=f"private diagnostic {TEST_PASSWORD}".encode(),
)
with pytest.raises(wifi.HostWifiProfileError) as raised:
wifi.check_wifi_profile(
helper,
TEST_PROFILE_ID,
"XGR-OFFLINE",
runner=failing_compiler,
)
assert raised.value.reason_code == expected_reason_code
assert raised.value.helper_stage == "compile"
assert isinstance(raised.value.helper_elapsed_ms, int)
assert raised.value.helper_elapsed_ms >= 0
assert TEST_PASSWORD not in str(raised.value)
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",
)