Package onboard K1 separately with a private portable installer
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,224 @@
|
||||
"""Private installer and import-boundary checks with synthetic material only."""
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
PACKAGING = ROOT / "plugins/xgrids-k1/packaging"
|
||||
SYNTHETIC_KEY = b"11111111-2222-3333-4444-555555555555"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def modules(monkeypatch):
|
||||
monkeypatch.syspath_prepend(str(PACKAGING))
|
||||
from importlib import import_module
|
||||
|
||||
credential = import_module("credential_install")
|
||||
runtime = import_module("runtime_payload")
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"k1_package_builder_test", PACKAGING / "build_deb.py"
|
||||
)
|
||||
builder = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(builder)
|
||||
return credential, runtime, builder
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def root_metadata(monkeypatch, tmp_path, modules):
|
||||
credential, _, _ = modules
|
||||
original_stat = Path.stat
|
||||
original_fstat = os.fstat
|
||||
|
||||
def owned(info):
|
||||
values = list(info)
|
||||
values[4] = 0
|
||||
return os.stat_result(values)
|
||||
|
||||
def file_stat(path, *args, **kwargs):
|
||||
info = original_stat(path, *args, **kwargs)
|
||||
return owned(info) if path.is_relative_to(tmp_path) else info
|
||||
|
||||
monkeypatch.setattr(Path, "stat", file_stat)
|
||||
monkeypatch.setattr(credential.os, "fstat", lambda fd: owned(original_fstat(fd)))
|
||||
monkeypatch.setattr(credential.os, "geteuid", lambda: 0)
|
||||
return credential
|
||||
|
||||
|
||||
def test_private_material_is_encrypted_on_each_host_and_reinstall_is_idempotent(
|
||||
root_metadata,
|
||||
tmp_path,
|
||||
):
|
||||
credential = root_metadata
|
||||
calls = []
|
||||
|
||||
def runner(args, **kwargs):
|
||||
calls.append(args)
|
||||
assert SYNTHETIC_KEY.decode() not in str(args)
|
||||
if args[1] == "encrypt":
|
||||
assert kwargs["input"] == SYNTHETIC_KEY
|
||||
Path(args[-1]).write_bytes(b"synthetic-host-encrypted-material")
|
||||
return subprocess.CompletedProcess(args, 0)
|
||||
return subprocess.CompletedProcess(args, 0, stdout=SYNTHETIC_KEY)
|
||||
|
||||
store = tmp_path / "credentials"
|
||||
assert credential.install(bytearray(SYNTHETIC_KEY), root=store, runner=runner) == "installed"
|
||||
before = (store / "k1-application").stat().st_ino
|
||||
assert credential.install(bytearray(SYNTHETIC_KEY), root=store, runner=runner) == "unchanged"
|
||||
assert (store / "k1-application").stat().st_ino == before
|
||||
assert (store / "k1-application").stat().st_mode & 0o777 == 0o600
|
||||
assert [args[1] for args in calls] == ["encrypt", "decrypt"]
|
||||
|
||||
|
||||
def test_different_material_never_rotates_an_existing_installation(root_metadata, tmp_path):
|
||||
store = tmp_path / "credentials"
|
||||
store.mkdir(mode=0o700)
|
||||
target = store / "k1-application"
|
||||
target.write_bytes(b"old-encrypted-credential")
|
||||
target.chmod(0o600)
|
||||
|
||||
def runner(args, **kwargs):
|
||||
assert args[1] == "decrypt"
|
||||
return subprocess.CompletedProcess(args, 0, stdout=b"a-different-private-application-value")
|
||||
|
||||
with pytest.raises(ValueError, match="explicit rotation") as error:
|
||||
root_metadata.install(bytearray(SYNTHETIC_KEY), root=store, runner=runner)
|
||||
assert SYNTHETIC_KEY.decode() not in str(error.value)
|
||||
assert target.read_bytes() == b"old-encrypted-credential"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", [b"", b"short", b"x" * 1025, b"\x00" * 36, b" " * 36])
|
||||
def test_bad_material_fails_before_system_tools(modules, tmp_path, bad):
|
||||
credential, _, _ = modules
|
||||
|
||||
def forbidden(*args, **kwargs):
|
||||
pytest.fail("Invalid material reached an operating-system command")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
credential.install(bytearray(bad), root=tmp_path / "absent", runner=forbidden)
|
||||
assert not (tmp_path / "absent").exists()
|
||||
|
||||
|
||||
def test_bundle_rejects_symlink_and_public_permissions(root_metadata, tmp_path):
|
||||
target = tmp_path / "profile"
|
||||
target.write_bytes(SYNTHETIC_KEY)
|
||||
target.chmod(0o644)
|
||||
with pytest.raises(ValueError):
|
||||
root_metadata.read_bundle(target)
|
||||
target.chmod(0o600)
|
||||
link = tmp_path / "link"
|
||||
link.symlink_to(target)
|
||||
with pytest.raises(OSError):
|
||||
root_metadata.read_bundle(link)
|
||||
assert root_metadata.read_bundle(target) == SYNTHETIC_KEY
|
||||
|
||||
|
||||
def test_private_release_contains_material_only_in_root_private_member(
|
||||
modules, monkeypatch, tmp_path
|
||||
):
|
||||
_, _, builder = modules
|
||||
packaging = tmp_path / "packaging"
|
||||
shutil.copytree(PACKAGING, packaging, ignore=shutil.ignore_patterns("__pycache__"))
|
||||
(packaging / "k1-bundle.json").write_text('{"wheels": []}')
|
||||
monkeypatch.setattr(builder, "PACKAGING", packaging)
|
||||
public = builder.payload(tmp_path)
|
||||
private = builder.payload(tmp_path, authority=bytearray(SYNTHETIC_KEY))
|
||||
assert all(SYNTHETIC_KEY not in data for _, data, _ in public)
|
||||
secret_members = [(name, mode) for name, data, mode in private if SYNTHETIC_KEY in data]
|
||||
assert secret_members == [("usr/share/mission-core-node/k1/private-application-key", 0o600)]
|
||||
provenance = next(data for name, data, _ in private if name.endswith("provenance.json"))
|
||||
metadata = json.loads(provenance)
|
||||
assert metadata["application_material_included"] is True
|
||||
assert not any("private-application-key" in key for key in metadata["files"])
|
||||
assert metadata["acceptance_scope"] == "owner-controlled-k1-fw-3.0.2-ubuntu-24.04-amd64"
|
||||
|
||||
destination = tmp_path / "private.deb"
|
||||
builder.build(destination, tmp_path, authority=bytearray(SYNTHETIC_KEY))
|
||||
assert destination.stat().st_mode & 0o777 == 0o600
|
||||
original = destination.read_bytes()
|
||||
with pytest.raises(FileExistsError):
|
||||
builder.build(destination, tmp_path, authority=bytearray(SYNTHETIC_KEY))
|
||||
assert destination.read_bytes() == original
|
||||
position = 8
|
||||
members = {}
|
||||
while position < len(original):
|
||||
header = original[position : position + 60]
|
||||
length = int(header[48:58])
|
||||
members[header[:16].decode().strip().rstrip("/")] = original[
|
||||
position + 60 : position + 60 + length
|
||||
]
|
||||
position += 60 + length + length % 2
|
||||
with tarfile.open(fileobj=io.BytesIO(members["control.tar.gz"]), mode="r:gz") as archive:
|
||||
control = archive.extractfile("control").read().decode()
|
||||
assert "Depends: mission-core-node (>= 0.8.0)" in control
|
||||
assert "Replaces: mission-core-node (<< 0.8.0)" in control
|
||||
|
||||
|
||||
def test_onboard_imports_run_from_declared_payload_without_core_checkout(modules, tmp_path):
|
||||
_, runtime, builder = modules
|
||||
stage = tmp_path / "installed"
|
||||
for path in runtime.files():
|
||||
target = stage / path.relative_to(ROOT)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copyfile(path, target)
|
||||
for relative in builder.RESOURCES:
|
||||
target = stage / relative
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copyfile(ROOT / relative, target)
|
||||
script = """
|
||||
import asyncio, pathlib, sys
|
||||
stage = pathlib.Path(sys.argv[1])
|
||||
sys.path.insert(0, str(stage / "src"))
|
||||
from k1link.device_plugins.xgrids_k1.node_bridge import NodeBridge
|
||||
assert "k1link.laboratory.execution" not in sys.modules
|
||||
assert "k1link.device_plugins.xgrids_k1.legacy_api" not in sys.modules
|
||||
assert "k1link.compute.jobs" not in sys.modules
|
||||
import k1link
|
||||
assert pathlib.Path(k1link.__file__).is_relative_to(stage)
|
||||
bridge = NodeBridge(stage)
|
||||
async def check():
|
||||
try:
|
||||
value = await bridge.state()
|
||||
assert value["phase"] == "idle"
|
||||
assert value["candidates"] == []
|
||||
for name, module in list(sys.modules.items()):
|
||||
if name.startswith("k1link.") and getattr(module, "__file__", None):
|
||||
assert pathlib.Path(module.__file__).is_relative_to(stage), name
|
||||
finally:
|
||||
bridge.service.close()
|
||||
asyncio.run(check())
|
||||
"""
|
||||
env = dict(os.environ, MISSIONCORE_DATA_DIR=str(tmp_path / "state"))
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-I", "-c", script, str(stage)],
|
||||
cwd=tmp_path,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=40,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_lab_lazy_exports_keep_the_public_objects_and_do_not_load_on_leaf_import(tmp_path):
|
||||
script = """
|
||||
import importlib, sys
|
||||
import k1link.laboratory as lab
|
||||
from k1link.laboratory.canonical_recorded_migration import matches_historical_recorded_projection
|
||||
assert "k1link.laboratory.execution" not in sys.modules
|
||||
assert set(lab.__all__) == set(lab._EXPORTS)
|
||||
for name in lab.__all__:
|
||||
assert getattr(lab, name) is getattr(importlib.import_module(lab._EXPORTS[name]), name)
|
||||
"""
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", script], cwd=tmp_path, capture_output=True, text=True, timeout=30
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
Reference in New Issue
Block a user