DCCONSTRUCTIONS
0ca7316a24
Capture the current unreleased K1 connection, recovery, lifecycle, viewer, and test work as a single known-bad baseline for subsequent fixes.
1043 lines
35 KiB
Python
1043 lines
35 KiB
Python
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",
|
|
)
|