Accept systemd credential delivery and clarify manual K1 control

tests/test_linux_application_authority.py

@@ -0,0 +1,77 @@
+import os
+import secrets
+import stat
+
+import pytest
+
+from k1link.device_plugins.xgrids_k1.linux_host import LinuxApplicationAuthorityLoader
+from k1link.device_plugins.xgrids_k1.protocol.application_authority import (
+    ApplicationAuthorityLoadError,
+)
+
+
+@pytest.mark.parametrize("owner,mode,allowed", [
+    ("root", 0o440, True),  # systemd v255: root-owned file and service UID read ACL
+    ("root", 0o400, True),
+    ("service", 0o400, True),
+    ("service", 0o600, True),
+    ("root", 0o444, False),
+    ("root", 0o460, False),
+    ("service", 0o440, False),
+    ("other", 0o400, False),
+])
+def test_systemd_credential_ownership_contract(tmp_path, monkeypatch, owner, mode, allowed):
+    secret = secrets.token_hex(18)
+    path = tmp_path / "k1-application"
+    path.write_text(secret)
+    path.chmod(0o600)
+    file_inode, directory_inode = path.stat().st_ino, tmp_path.stat().st_ino
+    original = os.fstat
+
+    def delivered_metadata(fd):
+        result = original(fd)
+        fields = list(result)
+        if result.st_ino == file_inode:
+            fields[0] = stat.S_IFREG | mode
+            fields[4] = {"root": 0, "service": os.geteuid(), "other": os.geteuid()+1}[owner]
+        elif result.st_ino == directory_inode:
+            fields[0] = stat.S_IFDIR | 0o550
+            fields[4] = 0
+        return os.stat_result(fields)
+
+    # Read actual bounded file bytes; replace only kernel metadata unavailable
+    # on macOS. This is the documented systemd credential ACL representation.
+    monkeypatch.setattr(os, "fstat", delivered_metadata)
+    loader = LinuxApplicationAuthorityLoader(tmp_path)
+    if allowed:
+        assert loader.load().openapi_key == secret
+    else:
+        with pytest.raises(ApplicationAuthorityLoadError) as failure:
+            loader.load()
+        assert secret not in str(failure.value)
+
+
+@pytest.mark.parametrize(
+    "kind", ["missing", "file-symlink", "directory-symlink", "fifo", "oversized"],
+)
+def test_credential_loader_rejects_nonregular_or_unbounded_input(tmp_path, kind):
+    directory = tmp_path / "credentials"
+    directory.mkdir(mode=0o700)
+    path = directory / "k1-application"
+    if kind == "file-symlink":
+        target = tmp_path / "target"
+        target.write_text(secrets.token_hex(18))
+        path.symlink_to(target)
+    elif kind == "directory-symlink":
+        path.write_text(secrets.token_hex(18))
+        path.chmod(0o600)
+        link = tmp_path / "link"
+        link.symlink_to(directory, target_is_directory=True)
+        directory = link
+    elif kind == "fifo":
+        os.mkfifo(path, 0o600)
+    elif kind == "oversized":
+        path.write_text(secrets.token_hex(1025))
+        path.chmod(0o600)
+    with pytest.raises(ApplicationAuthorityLoadError):
+        LinuxApplicationAuthorityLoader(directory).load()

tests/test_node_k1_bridge.py

@@ -70,6 +70,39 @@ def bridge():
     return result
 
 
+@pytest.mark.parametrize("available", [False, True])
+def test_startup_credential_check_does_not_authorize_control(monkeypatch, available):
+    from k1link.device_plugins.xgrids_k1 import node_bridge
+    from k1link.device_plugins.xgrids_k1.protocol.application_authority import (
+        ApplicationAuthorityLoadError,
+    )
+
+    calls = []
+
+    class Loader:
+        def load(self):
+            calls.append("load")
+            if not available:
+                raise ApplicationAuthorityLoadError("Unavailable")
+            return object()
+
+    monkeypatch.setattr(node_bridge, "_validate_installed_compatibility_profile", lambda _: None)
+    monkeypatch.setattr(node_bridge, "LinuxApplicationAuthorityLoader", Loader)
+    monkeypatch.setattr(node_bridge, "XgridsK1CompatibilityService", lambda *a, **kw: object())
+    device = NodeBridge(Path.cwd())
+    device.facade = Facade()
+    device.facade.current["connection_lifecycle"] = {"connection_ready": False}
+
+    async def read():
+        for _ in range(2):
+            snapshot = await device.state()
+            assert snapshot["application_authority_available"] is available
+            assert snapshot["connected"] is False
+
+    asyncio.run(read())
+    assert calls == ["load"]
+
+
 def test_journalled_failure_logs_source_without_secret_and_does_not_repeat(caplog):
     from bleak.exc import BleakError
 
@@ -352,6 +385,27 @@ def test_sensor_projection_binds_native_sdk_to_board():
     assert value["kind"] == "k1"
 
 
+def test_applied_wifi_does_not_grant_control_or_start_authority():
+    current = state()
+    current["connection_lifecycle"] = {
+        "connection_ready": False, "ready_to_start": False,
+        "allowed_actions": ["verify-control-read-only"],
+    }
+    current["connection_attempt"] = {
+        "phase": "network_applied", "public_error_code": "application_authority_unavailable",
+    }
+    item = project_sensor(current, "node-test")
+    assert not item["online"] and not item["verified"]
+    assert item["control"]["network_applied"]
+    assert item["control"]["reason_code"] == "application_authority_unavailable"
+    assert not item["control"]["can_start"]
+    assert item["control"]["can_verify"]
+    current["connection_lifecycle"].update(connection_ready=True, ready_to_start=True)
+    item = project_sensor(current, "node-test")
+    assert item["online"] and item["verified"] and item["control"]["can_start"]
+    assert item["control"]["reason_code"] is None
+
+
 def test_one_start_intent_preserves_canonical_enter_prepare_start_sequence():
     async def run():
         device = bridge()