DCCONSTRUCTIONS
b53d6d5a45
Add calibrated K1 projection, recorded and near-live perception qualification, unified Rerun operator layers, bounded replay admission, audited viewer controls, worker experiments, and lab evidence.
646 lines
26 KiB
Python
646 lines
26 KiB
Python
"""Strict loader for the local XGRIDS K1 compatibility profile.
|
|
|
|
The profile is descriptive and fail-closed. Loading it never performs device
|
|
I/O and never grants write authority to a transport implementation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
PROFILE_SCHEMA_VERSION = 1
|
|
DEFAULT_PROFILE_ID = "xgrids.lixelkity-k1.fw-3.0.2.local-network.v2"
|
|
DEFAULT_PROFILE_PATH = (
|
|
Path(__file__).parent / "profiles" / "fw-3.0.2" / "local-network.v2.json"
|
|
)
|
|
EVIDENCE_FLAGS = (
|
|
"observed",
|
|
"decoded",
|
|
"replay_verified",
|
|
"physical_verified",
|
|
"write_enabled",
|
|
)
|
|
|
|
|
|
class CompatibilityProfileError(ValueError):
|
|
"""The compatibility profile is malformed or exceeds its reviewed scope."""
|
|
|
|
|
|
def _object(value: Any, path: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise CompatibilityProfileError(f"{path} must be an object")
|
|
return value
|
|
|
|
|
|
def _array(value: Any, path: str) -> list[Any]:
|
|
if not isinstance(value, list):
|
|
raise CompatibilityProfileError(f"{path} must be an array")
|
|
return value
|
|
|
|
|
|
def _string(value: Any, path: str) -> str:
|
|
if not isinstance(value, str) or not value:
|
|
raise CompatibilityProfileError(f"{path} must be a non-empty string")
|
|
return value
|
|
|
|
|
|
def _unique_index(items: list[Any], path: str) -> dict[str, dict[str, Any]]:
|
|
index: dict[str, dict[str, Any]] = {}
|
|
for position, value in enumerate(items):
|
|
item_path = f"{path}[{position}]"
|
|
item = _object(value, item_path)
|
|
item_id = _string(item.get("id"), f"{item_path}.id")
|
|
if item_id in index:
|
|
raise CompatibilityProfileError(f"{path} contains duplicate id {item_id!r}")
|
|
index[item_id] = item
|
|
return index
|
|
|
|
|
|
def _validate_evidence(value: Any, path: str) -> dict[str, bool]:
|
|
evidence = _object(value, path)
|
|
if set(evidence) != set(EVIDENCE_FLAGS):
|
|
expected = ", ".join(EVIDENCE_FLAGS)
|
|
raise CompatibilityProfileError(f"{path} must contain exactly: {expected}")
|
|
validated: dict[str, bool] = {}
|
|
for flag in EVIDENCE_FLAGS:
|
|
flag_value = evidence[flag]
|
|
if not isinstance(flag_value, bool):
|
|
raise CompatibilityProfileError(f"{path}.{flag} must be a boolean")
|
|
validated[flag] = flag_value
|
|
if evidence["write_enabled"]:
|
|
raise CompatibilityProfileError(f"{path}.write_enabled must remain false in v1")
|
|
return validated
|
|
|
|
|
|
def _walk_and_validate_evidence(value: Any, path: str = "$") -> None:
|
|
if isinstance(value, dict):
|
|
if "evidence" in value:
|
|
_validate_evidence(value["evidence"], f"{path}.evidence")
|
|
for key, child in value.items():
|
|
child_path = f"{path}.{key}"
|
|
if key == "write_enabled" and path != "$.evidence_vocabulary":
|
|
if not isinstance(child, bool):
|
|
raise CompatibilityProfileError(f"{child_path} must be a boolean")
|
|
if child:
|
|
raise CompatibilityProfileError(f"{child_path} must remain false in v1")
|
|
_walk_and_validate_evidence(child, child_path)
|
|
elif isinstance(value, list):
|
|
for position, child in enumerate(value):
|
|
_walk_and_validate_evidence(child, f"{path}[{position}]")
|
|
|
|
|
|
def _expect_evidence(
|
|
item: dict[str, Any],
|
|
path: str,
|
|
*,
|
|
observed: bool,
|
|
decoded: bool,
|
|
replay_verified: bool,
|
|
physical_verified: bool,
|
|
) -> None:
|
|
actual = _validate_evidence(item.get("evidence"), f"{path}.evidence")
|
|
expected = {
|
|
"observed": observed,
|
|
"decoded": decoded,
|
|
"replay_verified": replay_verified,
|
|
"physical_verified": physical_verified,
|
|
"write_enabled": False,
|
|
}
|
|
if actual != expected:
|
|
raise CompatibilityProfileError(
|
|
f"{path}.evidence exceeds or contradicts the reviewed v1 evidence"
|
|
)
|
|
|
|
|
|
def _validate_sources(profile: dict[str, Any]) -> set[str]:
|
|
sources = _unique_index(
|
|
_array(profile.get("evidence_sources"), "$.evidence_sources"),
|
|
"$.evidence_sources",
|
|
)
|
|
for source_id, source in sources.items():
|
|
_string(source.get("kind"), f"$.evidence_sources[{source_id!r}].kind")
|
|
path = _string(source.get("path"), f"$.evidence_sources[{source_id!r}].path")
|
|
if path.startswith("/") or ".." in Path(path).parts:
|
|
raise CompatibilityProfileError(
|
|
f"$.evidence_sources[{source_id!r}].path must be repository-relative"
|
|
)
|
|
_string(source.get("scope"), f"$.evidence_sources[{source_id!r}].scope")
|
|
return set(sources)
|
|
|
|
|
|
def _validate_connection_modes(scope: dict[str, Any]) -> None:
|
|
modes = _unique_index(
|
|
_array(scope.get("connection_modes"), "$.scope.connection_modes"),
|
|
"$.scope.connection_modes",
|
|
)
|
|
expected = {
|
|
"bridge": {
|
|
"topology": "direct-lan",
|
|
"direction": "k1-joins-existing-network",
|
|
"device_network_action": "single-reviewed-ble-provisioning-write",
|
|
"host_network_action": "none",
|
|
"target_address_policy": "ble-status-non-ap-private-ipv4",
|
|
"acceptance": "mission-core-physical-accepted",
|
|
},
|
|
"direct-connect": {
|
|
"topology": "controller-hotspot",
|
|
"direction": "k1-joins-controller-network",
|
|
"device_network_action": "single-reviewed-ble-provisioning-write",
|
|
"host_network_action": "operator-prepared-hotspot",
|
|
"target_address_policy": "ble-status-non-ap-private-ipv4",
|
|
"acceptance": "implementation-ready-physical-acceptance-pending",
|
|
},
|
|
"quick-connect": {
|
|
"topology": "device-ap",
|
|
"direction": "controller-joins-k1-network",
|
|
"device_network_action": "single-reviewed-ble-ap-activation-write",
|
|
"host_network_action": (
|
|
"exact-firmware-provider-and-single-corewlan-association"
|
|
),
|
|
"target_address_policy": "reviewed-fixed-k1-ap-private-ipv4",
|
|
"acceptance": "prepared-macos-host-physical-accepted",
|
|
},
|
|
}
|
|
if set(modes) != set(expected):
|
|
raise CompatibilityProfileError("connection mode set differs from reviewed evidence")
|
|
for mode_id, expected_fields in expected.items():
|
|
actual = {key: value for key, value in modes[mode_id].items() if key != "id"}
|
|
if actual != expected_fields:
|
|
raise CompatibilityProfileError(
|
|
f"connection mode {mode_id!r} differs from reviewed evidence"
|
|
)
|
|
|
|
|
|
def _validate_source_references(value: Any, source_ids: set[str], path: str = "$") -> None:
|
|
if isinstance(value, dict):
|
|
if "source_ids" in value:
|
|
references = _array(value["source_ids"], f"{path}.source_ids")
|
|
if len(references) != len(set(references)):
|
|
raise CompatibilityProfileError(f"{path}.source_ids contains duplicates")
|
|
for position, source_id in enumerate(references):
|
|
source_id = _string(source_id, f"{path}.source_ids[{position}]")
|
|
if source_id not in source_ids:
|
|
raise CompatibilityProfileError(
|
|
f"{path}.source_ids[{position}] references unknown evidence source"
|
|
)
|
|
for key, child in value.items():
|
|
_validate_source_references(child, source_ids, f"{path}.{key}")
|
|
elif isinstance(value, list):
|
|
for position, child in enumerate(value):
|
|
_validate_source_references(child, source_ids, f"{path}[{position}]")
|
|
|
|
|
|
def _validate_transports(profile: dict[str, Any]) -> None:
|
|
transports = _unique_index(
|
|
_array(profile.get("transports"), "$.transports"),
|
|
"$.transports",
|
|
)
|
|
if set(transports) != {
|
|
"ble.wifi-bootstrap.fw3.v1",
|
|
"mqtt.local-ipv4.fw3.v1",
|
|
"rtsp.camera-preview.fw3.v1",
|
|
}:
|
|
raise CompatibilityProfileError("$.transports must contain only the reviewed v1 transports")
|
|
|
|
ble = transports["ble.wifi-bootstrap.fw3.v1"]
|
|
if ble.get("service_uuid") != "00007f00-0000-1000-8000-00805f9b34fb":
|
|
raise CompatibilityProfileError("BLE service UUID differs from reviewed evidence")
|
|
characteristics = _object(ble.get("characteristics"), "$.transports[ble].characteristics")
|
|
if characteristics != {
|
|
"wifi_request": "00007f01-0000-1000-8000-00805f9b34fb",
|
|
"wifi_status": "00007f02-0000-1000-8000-00805f9b34fb",
|
|
}:
|
|
raise CompatibilityProfileError("BLE characteristic UUIDs differ from reviewed evidence")
|
|
if ble.get("request_frame_bytes") != 99:
|
|
raise CompatibilityProfileError("BLE provisioning frame must remain exactly 99 bytes")
|
|
_expect_evidence(
|
|
ble,
|
|
"$.transports[ble.wifi-bootstrap.fw3.v1]",
|
|
observed=True,
|
|
decoded=True,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
mqtt = transports["mqtt.local-ipv4.fw3.v1"]
|
|
if mqtt.get("protocol") != "MQTT 3.1.1":
|
|
raise CompatibilityProfileError("local application protocol must remain MQTT 3.1.1")
|
|
network = _object(mqtt.get("network"), "$.transports[mqtt].network")
|
|
if network.get("transport") != "TCP" or network.get("port") != 1883:
|
|
raise CompatibilityProfileError("local MQTT endpoint must remain TCP 1883")
|
|
if network.get("tls") is not False or network.get("authentication") != "none-observed":
|
|
raise CompatibilityProfileError("local MQTT security claim differs from observation")
|
|
allowlist = _array(
|
|
mqtt.get("subscription_allowlist"),
|
|
"$.transports[mqtt].subscription_allowlist",
|
|
)
|
|
if "lixel/application/report/#" not in allowlist:
|
|
raise CompatibilityProfileError("MQTT report-topic allowlist is missing")
|
|
for topic in allowlist:
|
|
topic = _string(topic, "$.transports[mqtt].subscription_allowlist[]")
|
|
if "/request/" in topic:
|
|
raise CompatibilityProfileError(
|
|
"request topics cannot enter the subscribe-only profile"
|
|
)
|
|
_expect_evidence(
|
|
mqtt,
|
|
"$.transports[mqtt.local-ipv4.fw3.v1]",
|
|
observed=True,
|
|
decoded=False,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
rtsp = transports["rtsp.camera-preview.fw3.v1"]
|
|
if rtsp.get("protocol") != "RTSP 1.0 with interleaved RTP over TCP":
|
|
raise CompatibilityProfileError("camera-preview protocol differs from observation")
|
|
rtsp_network = _object(rtsp.get("network"), "$.transports[rtsp].network")
|
|
if rtsp_network.get("transport") != "TCP" or rtsp_network.get("port") != 8554:
|
|
raise CompatibilityProfileError("camera-preview endpoint must remain TCP 8554")
|
|
if (
|
|
rtsp_network.get("tls") is not False
|
|
or rtsp_network.get("authentication") != "none-observed"
|
|
):
|
|
raise CompatibilityProfileError("camera-preview security differs from observation")
|
|
media = _object(rtsp.get("media"), "$.transports[rtsp].media")
|
|
if media != {
|
|
"codec": "H.264",
|
|
"rtp_payload_type": 96,
|
|
"clock_hz": 90000,
|
|
"framing": "RTP/AVP/TCP interleaved channels 0-1",
|
|
}:
|
|
raise CompatibilityProfileError("camera-preview media contract differs from observation")
|
|
_expect_evidence(
|
|
rtsp,
|
|
"$.transports[rtsp.camera-preview.fw3.v1]",
|
|
observed=True,
|
|
decoded=False,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
|
|
def _validate_channels(profile: dict[str, Any]) -> None:
|
|
channels = _unique_index(_array(profile.get("channels"), "$.channels"), "$.channels")
|
|
expected_ids = {
|
|
"spatial.point-cloud.live",
|
|
"spatial.pose.live",
|
|
"device.modeling.live",
|
|
"device.status.live",
|
|
"device.heartbeat.live",
|
|
"camera.preview.live",
|
|
}
|
|
if set(channels) != expected_ids:
|
|
raise CompatibilityProfileError("$.channels differs from the reviewed v1 channel set")
|
|
|
|
point = channels["spatial.point-cloud.live"]
|
|
if point.get("topic") != "lixel/application/report/lio_pcl":
|
|
raise CompatibilityProfileError("point-cloud topic differs from reviewed evidence")
|
|
_expect_evidence(
|
|
point,
|
|
"$.channels[spatial.point-cloud.live]",
|
|
observed=True,
|
|
decoded=True,
|
|
replay_verified=True,
|
|
physical_verified=True,
|
|
)
|
|
|
|
pose = channels["spatial.pose.live"]
|
|
if pose.get("topic") != "lixel/application/report/lio_pose":
|
|
raise CompatibilityProfileError("pose topic differs from reviewed evidence")
|
|
_expect_evidence(
|
|
pose,
|
|
"$.channels[spatial.pose.live]",
|
|
observed=True,
|
|
decoded=True,
|
|
replay_verified=True,
|
|
physical_verified=True,
|
|
)
|
|
|
|
modeling = channels["device.modeling.live"]
|
|
if (
|
|
modeling.get("topic") != "lixel/application/report/modeling"
|
|
or modeling.get("wire_format") != "protobuf ModelingReport acquisition telemetry subset"
|
|
or modeling.get("semantic_payload")
|
|
!= (
|
|
"nonnegative MoveDistance metres, MoveSpeed metres per second, "
|
|
"int64 ScanTime at two ticks per second, and int32 PgoProgress"
|
|
)
|
|
or modeling.get("bounds")
|
|
!= {
|
|
"max_mqtt_payload_bytes": 65_536,
|
|
"max_fields": 64,
|
|
"max_nested_fields": 16,
|
|
}
|
|
):
|
|
raise CompatibilityProfileError("modeling telemetry channel differs from evidence")
|
|
_expect_evidence(
|
|
modeling,
|
|
"$.channels[device.modeling.live]",
|
|
observed=True,
|
|
decoded=True,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
status = channels["device.status.live"]
|
|
if (
|
|
status.get("topic") != "lixel/application/report/device_status"
|
|
or status.get("wire_format") != "protobuf DeviceStatusReport acquisition lifecycle subset"
|
|
or status.get("semantic_payload")
|
|
!= (
|
|
"bounded modeling-state base-offset mapping, init-ready flag, "
|
|
"project presence and redacted identity fields"
|
|
)
|
|
):
|
|
raise CompatibilityProfileError("device-status channel differs from evidence")
|
|
_expect_evidence(
|
|
status,
|
|
"$.channels[device.status.live]",
|
|
observed=True,
|
|
decoded=True,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
heartbeat = channels["device.heartbeat.live"]
|
|
if (
|
|
heartbeat.get("topic") != "lixel/application/report/heartbeat"
|
|
or heartbeat.get("semantic_payload") is not None
|
|
):
|
|
raise CompatibilityProfileError("device.heartbeat.live must remain raw-only in v1")
|
|
_expect_evidence(
|
|
heartbeat,
|
|
"$.channels[device.heartbeat.live]",
|
|
observed=True,
|
|
decoded=False,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
camera = channels["camera.preview.live"]
|
|
if camera.get("discovery_status") != "observed":
|
|
raise CompatibilityProfileError("camera discovery must match the owner-controlled run")
|
|
if camera.get("topic") is not None:
|
|
raise CompatibilityProfileError("camera preview is not an MQTT topic")
|
|
expected_endpoints = {
|
|
"rtsp://{confirmed-device-private-ipv4}:8554/live/chn_left_main",
|
|
"rtsp://{confirmed-device-private-ipv4}:8554/live/chn_right_main",
|
|
}
|
|
endpoints = _array(camera.get("endpoint_templates"), "$.channels[camera].endpoint_templates")
|
|
if set(endpoints) != expected_endpoints or len(endpoints) != len(expected_endpoints):
|
|
raise CompatibilityProfileError("camera endpoint templates differ from observation")
|
|
if camera.get("wire_format") != "RTSP 1.0, interleaved RTP/TCP, H.264 PT96 at 90000 Hz":
|
|
raise CompatibilityProfileError("camera wire format differs from observation")
|
|
_string(camera.get("semantic_payload"), "$.channels[camera].semantic_payload")
|
|
_expect_evidence(
|
|
camera,
|
|
"$.channels[camera.preview.live]",
|
|
observed=True,
|
|
decoded=False,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
|
|
def _validate_acquisition_control(profile: dict[str, Any]) -> None:
|
|
control = _object(profile.get("acquisition_control"), "$.acquisition_control")
|
|
if control.get("mode") != "operator-manual" or control.get("write_enabled") is not False:
|
|
raise CompatibilityProfileError("acquisition control must remain operator-manual")
|
|
|
|
acceptance_transport = _object(
|
|
control.get("software_acceptance_transport"),
|
|
"$.acquisition_control.software_acceptance_transport",
|
|
)
|
|
if acceptance_transport != {
|
|
"status": "installed-operator-present",
|
|
"default_authority": "disabled",
|
|
"profile_gate": "live-device-info-exact-match",
|
|
"dialogue": "single-socket-canonical-start-to-stop",
|
|
"automatic_retry": False,
|
|
"supported_mount_type": "handheld",
|
|
"supported_gnss_mode": "none",
|
|
}:
|
|
raise CompatibilityProfileError(
|
|
"software acceptance transport differs from the reviewed operator-present contract"
|
|
)
|
|
|
|
device_control = _object(
|
|
control.get("verified_device_control"),
|
|
"$.acquisition_control.verified_device_control",
|
|
)
|
|
if device_control.get("gesture") != "physical-double-click":
|
|
raise CompatibilityProfileError("physical acquisition gesture differs from lab evidence")
|
|
_expect_evidence(
|
|
device_control,
|
|
"$.acquisition_control.verified_device_control",
|
|
observed=True,
|
|
decoded=False,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
actions = _unique_index(
|
|
_array(control.get("semantic_actions"), "$.acquisition_control.semantic_actions"),
|
|
"$.acquisition_control.semantic_actions",
|
|
)
|
|
if set(actions) != {
|
|
"acquisition.start",
|
|
"acquisition.stop",
|
|
"calibration.device.start",
|
|
}:
|
|
raise CompatibilityProfileError("semantic action set differs from reviewed v1")
|
|
|
|
for action_id, action_value in (("acquisition.start", 1), ("acquisition.stop", 2)):
|
|
action = actions[action_id]
|
|
if action.get("execution") != "operator-manual":
|
|
raise CompatibilityProfileError(f"{action_id} must remain operator-manual")
|
|
mapping = _object(
|
|
action.get("vendor_request_mapping"),
|
|
f"$.acquisition_control.semantic_actions[{action_id}].vendor_request_mapping",
|
|
)
|
|
if mapping.get("evidence_kind") != "owner-controlled-wire-observation":
|
|
raise CompatibilityProfileError(f"{action_id} vendor mapping differs from evidence")
|
|
if mapping.get("transport") != "MQTT 3.1.1":
|
|
raise CompatibilityProfileError(f"{action_id} transport differs from evidence")
|
|
if mapping.get("message_type") != "ModelingRequest":
|
|
raise CompatibilityProfileError(f"{action_id} message type differs from evidence")
|
|
if mapping.get("write_enabled") is not False:
|
|
raise CompatibilityProfileError(
|
|
f"{action_id} vendor mapping must explicitly remain write-disabled"
|
|
)
|
|
if mapping.get("topic") != "lixel/application/request/modeling":
|
|
raise CompatibilityProfileError(
|
|
f"{action_id} vendor topic differs from static evidence"
|
|
)
|
|
if (
|
|
mapping.get("qos") != 2
|
|
or mapping.get("retain") is not False
|
|
or mapping.get("action_field_value") != action_value
|
|
):
|
|
raise CompatibilityProfileError(
|
|
f"{action_id} vendor mapping differs from static evidence"
|
|
)
|
|
if mapping.get("header_contract") != {
|
|
"device_id": "explicit-observed-identity",
|
|
"session_id": "{device_id}:ModelingRequest",
|
|
"openapi_key": "private-application-level-runtime-authority",
|
|
}:
|
|
raise CompatibilityProfileError(
|
|
f"{action_id} header contract differs from retained evidence"
|
|
)
|
|
expected_request_fields: dict[str, object] = (
|
|
{
|
|
"project_name": "required-operator-value",
|
|
"record_mode": 2,
|
|
"scan_mode": 1,
|
|
"mount_type": 0,
|
|
"pre_project_id": "omitted-in-retained-request",
|
|
}
|
|
if action_id == "acquisition.start"
|
|
else {}
|
|
)
|
|
if mapping.get("request_fields") != expected_request_fields:
|
|
raise CompatibilityProfileError(
|
|
f"{action_id} request fields differ from retained evidence"
|
|
)
|
|
if mapping.get("success_result_code") != 302_252_033:
|
|
raise CompatibilityProfileError(
|
|
f"{action_id} success result differs from retained evidence"
|
|
)
|
|
if not _array(
|
|
mapping.get("required_unresolved_context"),
|
|
f"$.acquisition_control.semantic_actions[{action_id}].required_unresolved_context",
|
|
):
|
|
raise CompatibilityProfileError(f"{action_id} must declare unresolved request context")
|
|
_expect_evidence(
|
|
mapping,
|
|
f"$.acquisition_control.semantic_actions[{action_id}].vendor_request_mapping",
|
|
observed=True,
|
|
decoded=True,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
_expect_evidence(
|
|
action,
|
|
f"$.acquisition_control.semantic_actions[{action_id}]",
|
|
observed=True,
|
|
decoded=True,
|
|
replay_verified=False,
|
|
physical_verified=True,
|
|
)
|
|
|
|
calibration = actions["calibration.device.start"]
|
|
if calibration.get("execution") != "unavailable":
|
|
raise CompatibilityProfileError("calibration must remain unavailable in v1")
|
|
if calibration.get("vendor_request_mapping") is not None:
|
|
raise CompatibilityProfileError("calibration vendor request is not evidenced")
|
|
_expect_evidence(
|
|
calibration,
|
|
"$.acquisition_control.semantic_actions[calibration.device.start]",
|
|
observed=False,
|
|
decoded=False,
|
|
replay_verified=False,
|
|
physical_verified=False,
|
|
)
|
|
|
|
|
|
def validate_compatibility_profile(profile: Any) -> dict[str, Any]:
|
|
"""Validate the read-only firmware-3 local connection matrix."""
|
|
root = _object(profile, "$")
|
|
if root.get("schema_version") != PROFILE_SCHEMA_VERSION:
|
|
raise CompatibilityProfileError("unsupported compatibility profile schema_version")
|
|
if root.get("profile_id") != DEFAULT_PROFILE_ID:
|
|
raise CompatibilityProfileError("unexpected compatibility profile_id")
|
|
|
|
scope = _object(root.get("scope"), "$.scope")
|
|
if scope.get("vendor") != "XGRIDS" or scope.get("model") != "LixelKity K1":
|
|
raise CompatibilityProfileError("profile vendor/model must remain XGRIDS LixelKity K1")
|
|
if scope.get("platform_type") != "A4":
|
|
raise CompatibilityProfileError("profile platform type must remain the observed A4")
|
|
firmware = _object(scope.get("firmware"), "$.scope.firmware")
|
|
if firmware != {"match": "exact", "version": "3.0.2"}:
|
|
raise CompatibilityProfileError("profile must match firmware 3.0.2 exactly")
|
|
if scope.get("topology") != "local-network-matrix":
|
|
raise CompatibilityProfileError("profile topology must remain local-network-matrix")
|
|
_validate_connection_modes(scope)
|
|
|
|
vocabulary = _object(root.get("evidence_vocabulary"), "$.evidence_vocabulary")
|
|
if set(vocabulary) != set(EVIDENCE_FLAGS):
|
|
raise CompatibilityProfileError("evidence vocabulary differs from schema v1")
|
|
for flag in EVIDENCE_FLAGS:
|
|
_string(vocabulary[flag], f"$.evidence_vocabulary.{flag}")
|
|
|
|
safety = _object(root.get("safety"), "$.safety")
|
|
if safety.get("default_mode") != "read-only":
|
|
raise CompatibilityProfileError("profile default mode must remain read-only")
|
|
if safety.get("vendor_writes_enabled") is not False:
|
|
raise CompatibilityProfileError("profile must not enable vendor writes")
|
|
if safety.get("unknown_firmware_policy") != "reject-profile":
|
|
raise CompatibilityProfileError("unknown firmware must fail closed")
|
|
if safety.get("request_topic_subscription_enabled") is not False:
|
|
raise CompatibilityProfileError("request-topic subscription must remain disabled")
|
|
|
|
source_ids = _validate_sources(root)
|
|
_validate_source_references(root, source_ids)
|
|
_walk_and_validate_evidence(root)
|
|
_validate_transports(root)
|
|
_validate_channels(root)
|
|
_validate_acquisition_control(root)
|
|
return root
|
|
|
|
|
|
def load_compatibility_profile(path: Path | str = DEFAULT_PROFILE_PATH) -> dict[str, Any]:
|
|
"""Load a JSON profile, rejecting duplicate keys and unreviewed claims."""
|
|
resolved = Path(path).expanduser().resolve()
|
|
|
|
def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
result: dict[str, Any] = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise CompatibilityProfileError(f"duplicate JSON key {key!r}")
|
|
result[key] = value
|
|
return result
|
|
|
|
try:
|
|
raw = resolved.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise CompatibilityProfileError(f"cannot read compatibility profile: {resolved}") from exc
|
|
try:
|
|
profile = json.loads(raw, object_pairs_hook=reject_duplicate_keys)
|
|
except json.JSONDecodeError as exc:
|
|
raise CompatibilityProfileError(f"invalid compatibility profile JSON: {exc}") from exc
|
|
return validate_compatibility_profile(profile)
|
|
|
|
|
|
def matches_target(
|
|
profile: dict[str, Any],
|
|
*,
|
|
firmware: str,
|
|
topology: str,
|
|
) -> bool:
|
|
"""Return whether an already validated exact-match profile covers the target."""
|
|
validated = validate_compatibility_profile(profile)
|
|
scope = validated["scope"]
|
|
firmware_scope = _object(scope.get("firmware"), "$.scope.firmware")
|
|
modes = _array(scope.get("connection_modes"), "$.scope.connection_modes")
|
|
return firmware_scope.get("version") == firmware and any(
|
|
isinstance(mode, dict) and mode.get("topology") == topology for mode in modes
|
|
)
|
|
|
|
|
|
def _main() -> int:
|
|
parser = argparse.ArgumentParser(description="Validate an XGRIDS K1 compatibility profile")
|
|
parser.add_argument("path", nargs="?", type=Path, default=DEFAULT_PROFILE_PATH)
|
|
args = parser.parse_args()
|
|
profile = load_compatibility_profile(args.path)
|
|
print(profile["profile_id"])
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(_main())
|