diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index e3524ef0..488202ce 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -127,7 +127,7 @@ jobs: - name: shellcheck run: | sudo apt-get update -qq && sudo apt-get install -y -qq shellcheck - shellcheck -S warning install.sh scripts/install-docker.sh \ + shellcheck -S warning install.sh scripts/install-docker.sh scripts/hardware-id.sh \ windows/provision-msys2.sh tests/integration/harness.sh # The Windows installer ships a compiled runtime inside an MSYS2 tree, so diff --git a/CLAUDE.md b/CLAUDE.md index 6fc1ce41..8900d3c1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -160,6 +160,8 @@ State management: `core/src/plc_app/plc_state_manager.cpp` the WebSocket side is `webserver/debug_websocket.py` - **Watchdog**: `core/src/plc_app/utils/watchdog.c` - health monitoring - **Image tables**: `core/src/plc_app/image_tables.cpp` - I/O buffer management +- **Hardware ID v2**: `scripts/hardware-id.sh` (canonical bash reader; installers embed its marked + block verbatim), `webserver/hardware_id.py`, `GET /api/hardware-id`. Spec: `docs/HARDWARE_ID.md` ## Code Style diff --git a/docs/HARDWARE_ID.md b/docs/HARDWARE_ID.md new file mode 100644 index 00000000..04df7996 --- /dev/null +++ b/docs/HARDWARE_ID.md @@ -0,0 +1,109 @@ +# Hardware ID v2 + +The hardware ID identifies the physical machine an OpenPLC runtime licence was bought for, so a +lost licence can be recovered from the same hardware after a wipe or reinstall. Nothing is enforced +on the device: the ID only has to be the same every time it is computed on the same hardware. + +The reader is `scripts/hardware-id.sh`, plain bash with no dependency beyond coreutils. The block +between its `BEGIN hardware-id v2` and `END hardware-id v2` markers is the canonical copy: the +orchestrator-agent and autonomy-node installers embed it verbatim, and their checks compare it byte +for byte with this repository. + +Version 1 (`openplc-dev-v1`, used by VPP licences and debug function code 0x48) is unchanged. + +## Format + +``` +hw2:<64 lowercase hex characters> +``` + +## Derivation + +1. Read each element below. Every value has NUL bytes removed, surrounding whitespace trimmed and is + lower-cased. +2. Hash each element on its own: `sha256("openplc-hw-v2||")`, as lowercase hex. +3. Sort the element hashes, join them with a newline after each one, and hash the result: + `ID = "hw2:" + sha256(h1 "\n" h2 "\n" ... hn "\n")`. + +| Kind | Source | Accepted when | Strong | +| --- | --- | --- | --- | +| `serial` | `/proc/device-tree/serial-number` | Not empty | Yes | +| `uuid` | `/sys/class/dmi/id/product_uuid` | Hex with dashes; not all `0`, not all `f`, not `03000200-0400-0500-0006-000700080009` | Yes | +| `board` | `/sys/class/dmi/id/board_serial` | Not empty, `to be filled by o.e.m.`, `default string`, `not specified` or `0123456789` | Yes | +| `mac` | `/sys/class/net/*/address`, one element per NIC | See NIC rules; never inside a VM | Yes | +| `cpu` | `/proc/cpuinfo`, first match of each field | x86: `\|`; ARM: `\|` | No | + +NIC rules, all of which must hold: + +- the interface has a `device` link (rules out bridges, veth, tun, bonds and other virtual links); +- it is not wireless (no `wireless` or `phy80211` entry); +- `addr_assign_type` is `0` (factory address); +- the address is a well-formed MAC, not all zeros, with the locally administered bit clear; +- no directory on its device path has `removable` set to `removable`; +- it is not on USB, unless a USB port on its device path reports `connect_type` = `hardwired`. + +Other rules: + +- **No software state.** No element comes from the OS, kernel, architecture string, hostname or + installed packages. The same hardware gives the same ID under a 32-bit or 64-bit OS. +- **VMs.** When `/proc/cpuinfo` lists the `hypervisor` flag, or the DMI `sys_vendor` or + `product_name` names VMware, QEMU, KVM, VirtualBox (innotek), Xen, Amazon EC2, Google Compute + Engine, Parallels or Microsoft Corporation "Virtual Machine", NICs are left out. +- **Refused hosts.** When `/proc/sys/kernel/osrelease` or `/proc/version` mentions `microsoft`, + `wsl` or `linuxkit` (WSL2, Docker Desktop), no ID is produced. +- **No strong element, no ID.** The CPU alone never yields an ID. +- **Frozen.** Every rule and placeholder above is part of v2. Any change after licences are issued is + a new version, v3. + +## Usage + +``` +scripts/hardware-id.sh # prints the ID; exit status 2 when none can be derived +scripts/hardware-id.sh --elements # prints " " per element +scripts/hardware-id.sh --record # prints the record (below) +scripts/hardware-id.sh --store FILE # writes the record to FILE unless FILE already holds one +``` + +Run it as root. On x86 the DMI `product_uuid` and `board_serial` files are readable by root only, +so an unprivileged run sees only the CPU and derives no ID. Every installer runs as root. + +`OPENPLC_HWID_ROOT` replaces `/` for every read. It exists for tests only. + +## Record and storage + +The record is the ID line followed by one ` ` line per element. It is computed +once, at installation, and never replaced: `--store` (`hw2_store` in the embedded block) leaves an +existing record alone, so a reinstall on the same hardware keeps it and a wiped device recomputes +the same one. + +| Install | Who computes it | Record | +| --- | --- | --- | +| Native (`install.sh --native`) | `install.sh` | `/var/lib/openplc-runtime/hardware-id` | +| Container (`scripts/install-docker.sh`) | `install-docker.sh`, on the host | `/hardware-id`, default `/var/lib/openplc-runtime/hardware-id` | +| vPLC | orchestrator-agent's installer, on the host | `/etc/openplc-runtime/hardware-id` inside the container, read-only | + +A runtime installed before the record existed computes it on start when none is stored, but only +where it sees the host hardware: natively, or in the bootloader's privileged container. A vPLC never +computes one. + +The runtime advertises the record on `GET /api/hardware-id` (any authenticated user): + +```json +{"hardwareId": "hw2:…", "source": "device", "elements": [{"kind": "serial", "hash": "…"}]} +``` + +`source` is `orchestrator` in a vPLC. Without an ID, `hardwareId` and `source` are null and +`elements` is empty. + +## Test vectors + +`tests/pytest/hardware_id/vectors.json` holds fake hardware trees with their element values and the +expected ID. The tests check the reader against them and recompute every hash independently in Python. + +| Vector | Elements | ID | +| --- | --- | --- | +| `raspberry-pi` | serial, mac, cpu | `hw2:363598c3230281d8cc2cfcdd1b772a6957aa6b8fbd631712ba1b195be5d22318` | +| `x86-industrial-pc` | uuid, board, 2 mac, cpu | `hw2:5337b315679912a71e5f004a718544e1282e004f13acb0ac17c4c8142587a14c` | +| `vm` | uuid, cpu | `hw2:5e26377d20855444f09a7b40f13d8c4e00550166e98efbc6d1d01c05b51a5eb1` | +| `placeholders-only` | cpu | none | +| `wsl2` | none | none | diff --git a/install.sh b/install.sh index 6e047e48..4322df54 100755 --- a/install.sh +++ b/install.sh @@ -641,6 +641,11 @@ else mkdir -p /var/lib/openplc-runtime chmod 755 /var/lib/openplc-runtime log_info "Created persistent data directory at /var/lib/openplc-runtime" + if bash "$OPENPLC_DIR/scripts/hardware-id.sh" --store /var/lib/openplc-runtime/hardware-id; then + log_info "Hardware ID: $(head -n 1 /var/lib/openplc-runtime/hardware-id)" + else + log_warning "No hardware ID can be derived on this device; runtime licences cannot be bound to it." + fi fi fi diff --git a/scripts/hardware-id.sh b/scripts/hardware-id.sh new file mode 100755 index 00000000..3546938f --- /dev/null +++ b/scripts/hardware-id.sh @@ -0,0 +1,214 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: GPL-3.0-only +# Copyright (c) 2026 Autonomy® + +# Hardware ID v2 reader (spec: docs/HARDWARE_ID.md). Canonical copy: installers embed the +# block between the BEGIN/END markers verbatim, and their parity checks compare it byte for byte. +# +# Usage: hardware-id.sh print the ID (hw2:<64 hex>), exit 2 when none can be derived +# hardware-id.sh --elements print " " lines +# hardware-id.sh --record print the ID line followed by the element lines +# hardware-id.sh --store FILE write the record to FILE unless FILE already holds one +# OPENPLC_HWID_ROOT replaces / for every read (tests only). + +# BEGIN hardware-id v2 +hw2_root() { printf '%s' "${OPENPLC_HWID_ROOT:-}"; } + +hw2_sha256() { + if command -v sha256sum >/dev/null 2>&1; then + sha256sum | cut -d' ' -f1 + else + shasum -a 256 | cut -d' ' -f1 + fi +} + +hw2_lower() { tr '[:upper:]' '[:lower:]'; } + +# Reads a file, drops NUL bytes, trims surrounding whitespace and lower-cases it. +hw2_read() { + [ -r "$1" ] || return 1 + tr -d '\000' <"$1" | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//' | hw2_lower +} + +hw2_cpuinfo_field() { + sed -n "s/^$1[[:space:]]*:[[:space:]]*//p" "$(hw2_root)/proc/cpuinfo" 2>/dev/null | + head -n 1 | sed 's/[[:space:]]*$//' | hw2_lower +} + +hw2_refused_host() { + local r f + r="$(hw2_root)" + for f in "$r/proc/sys/kernel/osrelease" "$r/proc/version"; do + if [ -r "$f" ] && grep -qi -e microsoft -e wsl -e linuxkit "$f"; then + return 0 + fi + done + return 1 +} + +hw2_is_vm() { + local r vendor product + r="$(hw2_root)" + if grep -qiw hypervisor "$r/proc/cpuinfo" 2>/dev/null; then + return 0 + fi + vendor="$(hw2_read "$r/sys/class/dmi/id/sys_vendor")" + product="$(hw2_read "$r/sys/class/dmi/id/product_name")" + case "$vendor" in + *vmware* | *qemu* | *kvm* | *innotek* | *virtualbox* | *xen* | *"amazon ec2"* | *google* | *parallels*) return 0 ;; + esac + case "$product" in + *vmware* | *qemu* | *kvm* | *virtualbox* | *xen* | *"google compute engine"* | *parallels*) return 0 ;; + esac + if [ "$vendor" = "microsoft corporation" ] && [ "$product" = "virtual machine" ]; then + return 0 + fi + return 1 +} + +hw2_uuid_ok() { + local bare + bare="$(printf '%s' "$1" | tr -d '-')" + case "$bare" in + "" | *[!0-9a-f]*) return 1 ;; + esac + [ -n "$(printf '%s' "$bare" | tr -d '0')" ] || return 1 + [ -n "$(printf '%s' "$bare" | tr -d 'f')" ] || return 1 + [ "$1" != "03000200-0400-0500-0006-000700080009" ] +} + +hw2_board_serial_ok() { + case "$1" in + "" | "to be filled by o.e.m." | "default string" | "not specified" | "0123456789") return 1 ;; + esac + return 0 +} + +# Succeeds when the NIC's device path or an ancestor is marked removable, or sits on USB +# without a hardwired port. +hw2_nic_detachable() { + local dev on_usb hardwired + dev="$(cd -P "$1/device" 2>/dev/null && pwd)" || return 0 + on_usb=0 + hardwired=0 + while [ -n "$dev" ] && [ "$dev" != "/" ] && [ "$dev" != "$(hw2_root)" ]; do + if [ "$(hw2_read "$dev/removable")" = "removable" ]; then + return 0 + fi + case "${dev##*/}" in + usb[0-9]*) on_usb=1 ;; + esac + if [ "$(hw2_read "$dev/port/connect_type")" = "hardwired" ]; then + hardwired=1 + fi + dev="${dev%/*}" + done + [ "$on_usb" -eq 1 ] && [ "$hardwired" -eq 0 ] +} + +hw2_macs() { + local r nic mac first + r="$(hw2_root)" + for nic in "$r"/sys/class/net/*; do + [ -e "$nic/device" ] || continue + [ -e "$nic/wireless" ] && continue + [ -e "$nic/phy80211" ] && continue + [ "$(hw2_read "$nic/addr_assign_type")" = "0" ] || continue + mac="$(hw2_read "$nic/address")" + case "$mac" in + [0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]) ;; + *) continue ;; + esac + [ "$mac" = "00:00:00:00:00:00" ] && continue + first=$((16#${mac%%:*})) + [ $((first & 2)) -eq 0 ] || continue + hw2_nic_detachable "$nic" && continue + printf '%s\n' "$mac" + done | sort -u +} + +hw2_element() { printf '%s %s\n' "$1" "$(printf 'openplc-hw-v2|%s|%s' "$1" "$2" | hw2_sha256)"; } + +# Prints " " per element; returns 1 when the host is refused or has no strong element. +hw2_elements() { + local r v strong=0 mac vendor model + r="$(hw2_root)" + if hw2_refused_host; then + return 1 + fi + v="$(hw2_read "$r/proc/device-tree/serial-number")" + if [ -n "$v" ]; then + hw2_element serial "$v" + strong=1 + fi + v="$(hw2_read "$r/sys/class/dmi/id/product_uuid")" + if hw2_uuid_ok "$v"; then + hw2_element uuid "$v" + strong=1 + fi + v="$(hw2_read "$r/sys/class/dmi/id/board_serial")" + if hw2_board_serial_ok "$v"; then + hw2_element board "$v" + strong=1 + fi + if ! hw2_is_vm; then + for mac in $(hw2_macs); do + hw2_element mac "$mac" + strong=1 + done + fi + vendor="$(hw2_cpuinfo_field vendor_id)" + model="$(hw2_cpuinfo_field 'model name')" + if [ -z "$vendor" ]; then + vendor="$(hw2_cpuinfo_field 'CPU implementer')" + model="$(hw2_cpuinfo_field 'CPU part')" + fi + if [ -n "$vendor$model" ]; then + hw2_element cpu "$vendor|$model" + fi + [ "$strong" -eq 1 ] +} + +hw2_id_of() { printf 'hw2:%s\n' "$(printf '%s\n' "$1" | cut -d' ' -f2 | sort | hw2_sha256)"; } + +# Prints hw2:<64 hex>; returns 1 when no ID can be derived. +hw2_id() { + local elements + elements="$(hw2_elements)" || return 1 + hw2_id_of "$elements" +} + +# Prints the ID line followed by the element lines; returns 1 when no ID can be derived. +hw2_record() { + local elements + elements="$(hw2_elements)" || return 1 + hw2_id_of "$elements" + printf '%s\n' "$elements" +} + +# Writes the record to $1 atomically; an existing record is never replaced. +hw2_store() { + local tmp + [ -s "$1" ] && return 0 + tmp="$1.tmp.$$" + if hw2_record >"$tmp"; then + chmod 644 "$tmp" && mv -f "$tmp" "$1" + else + rm -f "$tmp" + return 1 + fi +} +# END hardware-id v2 + +if [ "${BASH_SOURCE[0]}" = "$0" ]; then + set -u + case "${1:-}" in + --elements) hw2_elements || exit 2 ;; + --record) hw2_record || exit 2 ;; + --store) + [ -n "${2:-}" ] || { echo "usage: $0 --store FILE" >&2; exit 64; } + hw2_store "$2" || exit 2 + ;; + *) hw2_id || exit 2 ;; + esac +fi diff --git a/scripts/install-docker.sh b/scripts/install-docker.sh index 2c037592..15c44ebb 100755 --- a/scripts/install-docker.sh +++ b/scripts/install-docker.sh @@ -490,11 +490,206 @@ json_array() { printf ']' } +# Verbatim copy of the block in scripts/hardware-id.sh; a test keeps them identical. +# BEGIN hardware-id v2 +hw2_root() { printf '%s' "${OPENPLC_HWID_ROOT:-}"; } + +hw2_sha256() { + if command -v sha256sum >/dev/null 2>&1; then + sha256sum | cut -d' ' -f1 + else + shasum -a 256 | cut -d' ' -f1 + fi +} + +hw2_lower() { tr '[:upper:]' '[:lower:]'; } + +# Reads a file, drops NUL bytes, trims surrounding whitespace and lower-cases it. +hw2_read() { + [ -r "$1" ] || return 1 + tr -d '\000' <"$1" | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//' | hw2_lower +} + +hw2_cpuinfo_field() { + sed -n "s/^$1[[:space:]]*:[[:space:]]*//p" "$(hw2_root)/proc/cpuinfo" 2>/dev/null | + head -n 1 | sed 's/[[:space:]]*$//' | hw2_lower +} + +hw2_refused_host() { + local r f + r="$(hw2_root)" + for f in "$r/proc/sys/kernel/osrelease" "$r/proc/version"; do + if [ -r "$f" ] && grep -qi -e microsoft -e wsl -e linuxkit "$f"; then + return 0 + fi + done + return 1 +} + +hw2_is_vm() { + local r vendor product + r="$(hw2_root)" + if grep -qiw hypervisor "$r/proc/cpuinfo" 2>/dev/null; then + return 0 + fi + vendor="$(hw2_read "$r/sys/class/dmi/id/sys_vendor")" + product="$(hw2_read "$r/sys/class/dmi/id/product_name")" + case "$vendor" in + *vmware* | *qemu* | *kvm* | *innotek* | *virtualbox* | *xen* | *"amazon ec2"* | *google* | *parallels*) return 0 ;; + esac + case "$product" in + *vmware* | *qemu* | *kvm* | *virtualbox* | *xen* | *"google compute engine"* | *parallels*) return 0 ;; + esac + if [ "$vendor" = "microsoft corporation" ] && [ "$product" = "virtual machine" ]; then + return 0 + fi + return 1 +} + +hw2_uuid_ok() { + local bare + bare="$(printf '%s' "$1" | tr -d '-')" + case "$bare" in + "" | *[!0-9a-f]*) return 1 ;; + esac + [ -n "$(printf '%s' "$bare" | tr -d '0')" ] || return 1 + [ -n "$(printf '%s' "$bare" | tr -d 'f')" ] || return 1 + [ "$1" != "03000200-0400-0500-0006-000700080009" ] +} + +hw2_board_serial_ok() { + case "$1" in + "" | "to be filled by o.e.m." | "default string" | "not specified" | "0123456789") return 1 ;; + esac + return 0 +} + +# Succeeds when the NIC's device path or an ancestor is marked removable, or sits on USB +# without a hardwired port. +hw2_nic_detachable() { + local dev on_usb hardwired + dev="$(cd -P "$1/device" 2>/dev/null && pwd)" || return 0 + on_usb=0 + hardwired=0 + while [ -n "$dev" ] && [ "$dev" != "/" ] && [ "$dev" != "$(hw2_root)" ]; do + if [ "$(hw2_read "$dev/removable")" = "removable" ]; then + return 0 + fi + case "${dev##*/}" in + usb[0-9]*) on_usb=1 ;; + esac + if [ "$(hw2_read "$dev/port/connect_type")" = "hardwired" ]; then + hardwired=1 + fi + dev="${dev%/*}" + done + [ "$on_usb" -eq 1 ] && [ "$hardwired" -eq 0 ] +} + +hw2_macs() { + local r nic mac first + r="$(hw2_root)" + for nic in "$r"/sys/class/net/*; do + [ -e "$nic/device" ] || continue + [ -e "$nic/wireless" ] && continue + [ -e "$nic/phy80211" ] && continue + [ "$(hw2_read "$nic/addr_assign_type")" = "0" ] || continue + mac="$(hw2_read "$nic/address")" + case "$mac" in + [0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]) ;; + *) continue ;; + esac + [ "$mac" = "00:00:00:00:00:00" ] && continue + first=$((16#${mac%%:*})) + [ $((first & 2)) -eq 0 ] || continue + hw2_nic_detachable "$nic" && continue + printf '%s\n' "$mac" + done | sort -u +} + +hw2_element() { printf '%s %s\n' "$1" "$(printf 'openplc-hw-v2|%s|%s' "$1" "$2" | hw2_sha256)"; } + +# Prints " " per element; returns 1 when the host is refused or has no strong element. +hw2_elements() { + local r v strong=0 mac vendor model + r="$(hw2_root)" + if hw2_refused_host; then + return 1 + fi + v="$(hw2_read "$r/proc/device-tree/serial-number")" + if [ -n "$v" ]; then + hw2_element serial "$v" + strong=1 + fi + v="$(hw2_read "$r/sys/class/dmi/id/product_uuid")" + if hw2_uuid_ok "$v"; then + hw2_element uuid "$v" + strong=1 + fi + v="$(hw2_read "$r/sys/class/dmi/id/board_serial")" + if hw2_board_serial_ok "$v"; then + hw2_element board "$v" + strong=1 + fi + if ! hw2_is_vm; then + for mac in $(hw2_macs); do + hw2_element mac "$mac" + strong=1 + done + fi + vendor="$(hw2_cpuinfo_field vendor_id)" + model="$(hw2_cpuinfo_field 'model name')" + if [ -z "$vendor" ]; then + vendor="$(hw2_cpuinfo_field 'CPU implementer')" + model="$(hw2_cpuinfo_field 'CPU part')" + fi + if [ -n "$vendor$model" ]; then + hw2_element cpu "$vendor|$model" + fi + [ "$strong" -eq 1 ] +} + +hw2_id_of() { printf 'hw2:%s\n' "$(printf '%s\n' "$1" | cut -d' ' -f2 | sort | hw2_sha256)"; } + +# Prints hw2:<64 hex>; returns 1 when no ID can be derived. +hw2_id() { + local elements + elements="$(hw2_elements)" || return 1 + hw2_id_of "$elements" +} + +# Prints the ID line followed by the element lines; returns 1 when no ID can be derived. +hw2_record() { + local elements + elements="$(hw2_elements)" || return 1 + hw2_id_of "$elements" + printf '%s\n' "$elements" +} + +# Writes the record to $1 atomically; an existing record is never replaced. +hw2_store() { + local tmp + [ -s "$1" ] && return 0 + tmp="$1.tmp.$$" + if hw2_record >"$tmp"; then + chmod 644 "$tmp" && mv -f "$tmp" "$1" + else + rm -f "$tmp" + return 1 + fi +} +# END hardware-id v2 + write_spec() { mkdir -p "$BOOTLOADER_STATE_DIR" chmod 750 "$BOOTLOADER_STATE_DIR" mkdir -p "$RUNTIME_DATA_DIR" chmod 755 "$RUNTIME_DATA_DIR" + if hw2_store "$RUNTIME_DATA_DIR/hardware-id"; then + log_info "Hardware ID: $(head -n 1 "$RUNTIME_DATA_DIR/hardware-id")" + else + log_warning "No hardware ID can be derived on this device; runtime licences cannot be bound to it." + fi local spec="$BOOTLOADER_STATE_DIR/runtime-spec.json" local tmp="$spec.tmp" diff --git a/tests/pytest/hardware_id/test_hardware_id.py b/tests/pytest/hardware_id/test_hardware_id.py new file mode 100644 index 00000000..9b6cf021 --- /dev/null +++ b/tests/pytest/hardware_id/test_hardware_id.py @@ -0,0 +1,309 @@ +# SPDX-License-Identifier: GPL-3.0-only +# Copyright (c) 2026 Autonomy® + +"""Hardware ID v2 reader (scripts/hardware-id.sh) against fake hardware trees.""" + +import copy +import hashlib +import json +import os +import pathlib +import subprocess +from dataclasses import dataclass + +import pytest + +REPO = pathlib.Path(__file__).resolve().parents[3] +SCRIPT = REPO / "scripts" / "hardware-id.sh" +VECTORS = json.loads((pathlib.Path(__file__).parent / "vectors.json").read_text())["vectors"] + +Tree = dict[str, dict[str, str]] + + +@dataclass(frozen=True) +class Result: + code: int + out: str + + +def element_hash(kind: str, value: str) -> str: + return hashlib.sha256(f"openplc-hw-v2|{kind}|{value}".encode()).hexdigest() + + +def expected_id(elements: list[list[str]]) -> str: + hashes = sorted(element_hash(kind, value) for kind, value in elements) + return "hw2:" + hashlib.sha256("".join(h + "\n" for h in hashes).encode()).hexdigest() + + +def build(root: pathlib.Path, tree: Tree) -> pathlib.Path: + for rel, content in tree.get("files", {}).items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content) + for rel, target in tree.get("links", {}).items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.symlink_to(target) + return root + + +def run(root: pathlib.Path, *args: str) -> Result: + env = dict(os.environ, OPENPLC_HWID_ROOT=str(root)) + proc = subprocess.run( + ["bash", str(SCRIPT), *args], env=env, capture_output=True, text=True, check=False + ) + return Result(proc.returncode, proc.stdout) + + +def vector(name: str) -> Tree: + return copy.deepcopy(next(v for v in VECTORS if v["name"] == name)) + + +def elements_of(root: pathlib.Path) -> list[str]: + return [line.split(" ")[0] for line in run(root, "--elements").out.splitlines()] + + +@pytest.mark.parametrize("vec", VECTORS, ids=[v["name"] for v in VECTORS]) +def test_vector(tmp_path: pathlib.Path, vec: Tree) -> None: + root = build(tmp_path, vec) + result = run(root) + if vec["id"] is None: + assert result.code == 2 + assert result.out == "" + return + assert result.code == 0 + assert result.out == vec["id"] + "\n" + assert vec["id"] == expected_id(vec["elements"]) + listed = sorted(run(root, "--elements").out.splitlines()) + assert listed == sorted(f"{k} {element_hash(k, v)}" for k, v in vec["elements"]) + + +def test_vectors_are_listed_in_the_spec() -> None: + spec = (REPO / "docs" / "HARDWARE_ID.md").read_text() + for vec in VECTORS: + assert f"`{vec['name']}`" in spec + if vec["id"]: + assert vec["id"] in spec + + +def test_same_hardware_same_id_whatever_the_software(tmp_path: pathlib.Path) -> None: + first = build(tmp_path / "a", vector("raspberry-pi")) + other = vector("raspberry-pi") + other["files"]["proc/version"] = "Linux version 6.1.0-rpi7-rpi-v7l (armv7l) #1 SMP\n" + other["files"]["etc/hostname"] = "renamed\n" + second = build(tmp_path / "b", other) + assert run(first).out == run(second).out + + +def test_reads_are_trimmed_and_lowercased(tmp_path: pathlib.Path) -> None: + vec = vector("x86-industrial-pc") + vec["files"]["sys/class/dmi/id/board_serial"] = " CNWS2000900ABC \n" + assert run(build(tmp_path, vec)).out == vector("x86-industrial-pc")["id"] + "\n" + + +@pytest.mark.parametrize( + "uuid", + [ + "00000000-0000-0000-0000-000000000000", + "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF", + "03000200-0400-0500-0006-000700080009", + "Not Settable", + "", + ], +) +def test_uuid_placeholders_rejected(tmp_path: pathlib.Path, uuid: str) -> None: + vec = vector("x86-industrial-pc") + vec["files"]["sys/class/dmi/id/product_uuid"] = uuid + "\n" + assert "uuid" not in elements_of(build(tmp_path, vec)) + + +@pytest.mark.parametrize( + "serial", ["To be filled by O.E.M.", "Default string", "Not Specified", "0123456789", ""] +) +def test_board_serial_placeholders_rejected(tmp_path: pathlib.Path, serial: str) -> None: + vec = vector("x86-industrial-pc") + vec["files"]["sys/class/dmi/id/board_serial"] = serial + "\n" + assert "board" not in elements_of(build(tmp_path, vec)) + + +def _single_nic(name: str, mac: str, aat: str, device: str | None) -> Tree: + files = { + "proc/cpuinfo": "vendor_id\t: GenuineIntel\nmodel name\t: Test CPU\n", + f"sys/class/net/{name}/address": mac + "\n", + f"sys/class/net/{name}/addr_assign_type": aat + "\n", + } + links: dict[str, str] = {} + if device: + files[f"sys/{device}/.present"] = "" + links[f"sys/class/net/{name}/device"] = "../../../" + device + return {"files": files, "links": links} + + +PCI = "devices/pci0000:00/0000:00:1c.0/0000:01:00.0" +USB = "devices/pci0000:00/0000:00:14.0/usb1/1-1/1-1:1.0" + + +def test_wired_pci_nic_counts(tmp_path: pathlib.Path) -> None: + root = build(tmp_path, _single_nic("enp1s0", "00:e0:4c:68:00:01", "0", PCI)) + assert ( + run(root).out + == expected_id([["mac", "00:e0:4c:68:00:01"], ["cpu", "genuineintel|test cpu"]]) + "\n" + ) + + +def test_virtual_nic_without_device_link_ignored(tmp_path: pathlib.Path) -> None: + assert run(build(tmp_path, _single_nic("veth0", "00:e0:4c:68:00:01", "0", None))).code == 2 + + +def test_wireless_nic_ignored(tmp_path: pathlib.Path) -> None: + tree = _single_nic("wlp2s0", "00:e0:4c:68:00:01", "0", PCI) + tree["files"]["sys/class/net/wlp2s0/phy80211/.present"] = "" + assert run(build(tmp_path, tree)).code == 2 + + +def test_changed_mac_ignored(tmp_path: pathlib.Path) -> None: + assert run(build(tmp_path, _single_nic("enp1s0", "00:e0:4c:68:00:01", "3", PCI))).code == 2 + + +def test_locally_administered_mac_ignored(tmp_path: pathlib.Path) -> None: + assert run(build(tmp_path, _single_nic("enp1s0", "02:e0:4c:68:00:01", "0", PCI))).code == 2 + + +def test_removable_nic_ignored(tmp_path: pathlib.Path) -> None: + tree = _single_nic("enp1s0", "00:e0:4c:68:00:01", "0", PCI) + tree["files"]["sys/devices/pci0000:00/0000:00:1c.0/removable"] = "removable\n" + assert run(build(tmp_path, tree)).code == 2 + + +def test_usb_nic_ignored(tmp_path: pathlib.Path) -> None: + tree = _single_nic("enx0", "00:24:9b:11:22:33", "0", USB) + tree["files"]["sys/devices/pci0000:00/0000:00:14.0/usb1/1-1/port/connect_type"] = "hotplug\n" + assert run(build(tmp_path, tree)).code == 2 + + +def test_hardwired_usb_nic_counts(tmp_path: pathlib.Path) -> None: + tree = _single_nic("eth0", "b8:27:eb:11:22:33", "0", USB) + tree["files"]["sys/devices/pci0000:00/0000:00:14.0/usb1/1-1/port/connect_type"] = "hardwired\n" + assert elements_of(build(tmp_path, tree)) == ["mac", "cpu"] + + +def test_plugging_a_usb_nic_keeps_the_id(tmp_path: pathlib.Path) -> None: + before = run(build(tmp_path / "a", vector("raspberry-pi"))).out + vec = vector("raspberry-pi") + vec["files"]["sys/class/net/eth1/address"] = "00:24:9b:aa:bb:cc\n" + vec["files"]["sys/class/net/eth1/addr_assign_type"] = "0\n" + vec["files"]["sys/devices/platform/scb/fd500000.pcie/usb1/1-1/1-1:1.0/.present"] = "" + vec["links"][ + "sys/class/net/eth1/device" + ] = "../../../devices/platform/scb/fd500000.pcie/usb1/1-1/1-1:1.0" + assert run(build(tmp_path / "b", vec)).out == before + + +def test_vm_detected_by_dmi_leaves_nics_out(tmp_path: pathlib.Path) -> None: + tree = _single_nic("ens3", "00:16:3e:11:22:33", "0", PCI) + tree["files"]["sys/class/dmi/id/sys_vendor"] = "VMware, Inc.\n" + assert run(build(tmp_path, tree)).code == 2 + + +def test_vm_detected_by_hyperv_product_leaves_nics_out(tmp_path: pathlib.Path) -> None: + tree = _single_nic("eth0", "00:15:5d:11:22:33", "0", PCI) + tree["files"]["sys/class/dmi/id/sys_vendor"] = "Microsoft Corporation\n" + tree["files"]["sys/class/dmi/id/product_name"] = "Virtual Machine\n" + assert run(build(tmp_path, tree)).code == 2 + + +def test_microsoft_hardware_is_not_a_vm(tmp_path: pathlib.Path) -> None: + tree = _single_nic("eth0", "00:15:5d:11:22:33", "0", PCI) + tree["files"]["sys/class/dmi/id/sys_vendor"] = "Microsoft Corporation\n" + tree["files"]["sys/class/dmi/id/product_name"] = "Surface Pro 9\n" + assert run(build(tmp_path, tree)).code == 0 + + +def test_vm_detected_by_hypervisor_flag(tmp_path: pathlib.Path) -> None: + tree = _single_nic("ens3", "00:16:3e:11:22:33", "0", PCI) + tree["files"]["proc/cpuinfo"] = "vendor_id\t: GenuineIntel\nflags\t\t: fpu sse hypervisor\n" + assert run(build(tmp_path, tree)).code == 2 + + +@pytest.mark.parametrize( + "path,content", + [ + ("proc/sys/kernel/osrelease", "5.15.153.1-microsoft-standard-WSL2\n"), + ("proc/version", "Linux version 6.10.14-linuxkit (root@buildkitsandbox)\n"), + ], +) +def test_refused_hosts_produce_nothing(tmp_path: pathlib.Path, path: str, content: str) -> None: + vec = vector("x86-industrial-pc") + vec["files"][path] = content + root = build(tmp_path, vec) + assert run(root) == Result(2, "") + assert run(root, "--elements") == Result(2, "") + + +def test_cpu_alone_gives_no_id(tmp_path: pathlib.Path) -> None: + root = build(tmp_path, {"files": {"proc/cpuinfo": "vendor_id\t: GenuineIntel\n"}}) + assert run(root).code == 2 + + +def test_empty_root_gives_no_id(tmp_path: pathlib.Path) -> None: + assert run(tmp_path) == Result(2, "") + + +def test_arm_cpu_uses_first_core_registers(tmp_path: pathlib.Path) -> None: + root = build(tmp_path, vector("raspberry-pi")) + cpu = [line for line in run(root, "--elements").out.splitlines() if line.startswith("cpu ")] + assert cpu == [f"cpu {element_hash('cpu', '0x41|0xd08')}"] + + +def _block(path: pathlib.Path) -> str: + text = path.read_text() + start = text.index("# BEGIN hardware-id v2") + end = text.index("# END hardware-id v2") + len("# END hardware-id v2\n") + return text[start:end] + + +def test_docker_installer_embeds_the_canonical_block() -> None: + assert _block(REPO / "scripts" / "install-docker.sh") == _block(SCRIPT) + + +def test_record_is_the_id_then_the_elements(tmp_path: pathlib.Path) -> None: + root = build(tmp_path, vector("raspberry-pi")) + lines = run(root, "--record").out.splitlines() + assert lines[0] == vector("raspberry-pi")["id"] + assert lines[1:] == run(root, "--elements").out.splitlines() + + +def _store(root: pathlib.Path, target: pathlib.Path) -> int: + """Calls hw2_store from a shell with the installers' strict options.""" + script = f'set -euo pipefail; source "{SCRIPT}"; hw2_store "{target}"' + env = dict(os.environ, OPENPLC_HWID_ROOT=str(root)) + return subprocess.run(["bash", "-c", script], env=env, check=False).returncode + + +def test_store_writes_the_record_under_strict_mode(tmp_path: pathlib.Path) -> None: + root = build(tmp_path / "root", vector("x86-industrial-pc")) + target = tmp_path / "hardware-id" + assert _store(root, target) == 0 + assert target.read_text() == run(root, "--record").out + assert oct(target.stat().st_mode & 0o777) == "0o644" + + +def test_store_never_replaces_a_record(tmp_path: pathlib.Path) -> None: + root = build(tmp_path / "root", vector("x86-industrial-pc")) + target = tmp_path / "hardware-id" + target.write_text("hw2:" + "0" * 64 + "\n") + assert _store(root, target) == 0 + assert target.read_text() == "hw2:" + "0" * 64 + "\n" + + +def test_store_leaves_nothing_without_an_id(tmp_path: pathlib.Path) -> None: + target = tmp_path / "hardware-id" + assert _store(tmp_path / "empty", target) == 1 + assert list(tmp_path.iterdir()) == [] + + +def test_store_cli(tmp_path: pathlib.Path) -> None: + root = build(tmp_path / "root", vector("raspberry-pi")) + target = tmp_path / "hardware-id" + assert run(root, "--store", str(target)).code == 0 + assert target.read_text().splitlines()[0] == vector("raspberry-pi")["id"] diff --git a/tests/pytest/hardware_id/vectors.json b/tests/pytest/hardware_id/vectors.json new file mode 100644 index 00000000..5e706bd6 --- /dev/null +++ b/tests/pytest/hardware_id/vectors.json @@ -0,0 +1,151 @@ +{ + "format": "openplc-hardware-id-v2-vectors", + "vectors": [ + { + "name": "raspberry-pi", + "files": { + "sys/class/net/eth0/address": "dc:a6:32:01:02:03\n", + "sys/class/net/eth0/addr_assign_type": "0\n", + "sys/class/net/wlan0/address": "dc:a6:32:01:02:04\n", + "sys/class/net/wlan0/addr_assign_type": "0\n", + "sys/class/net/lo/address": "00:00:00:00:00:00\n", + "sys/class/net/lo/addr_assign_type": "0\n", + "proc/device-tree/serial-number": "10000000a1b2c3d4\u0000", + "proc/cpuinfo": "processor\t: 0\nBogoMIPS\t: 108.00\nFeatures\t: fp asimd evtstrm crc32 cpuid\nCPU implementer\t: 0x41\nCPU architecture: 8\nCPU variant\t: 0x0\nCPU part\t: 0xd08\nCPU revision\t: 3\n\nprocessor\t: 1\nCPU implementer\t: 0x41\nCPU part\t: 0xd08\n\nRevision\t: c03111\nSerial\t\t: 10000000a1b2c3d4\nModel\t\t: Raspberry Pi 4 Model B Rev 1.1\n", + "proc/version": "Linux version 6.1.0-rpi7-rpi-v8 (debian-kernel@lists.debian.org) #1 SMP PREEMPT Debian 1:6.1.63-1+rpt1 aarch64\n", + "sys/class/net/wlan0/wireless/.present": "", + "sys/devices/platform/scb/fd580000.ethernet/.present": "", + "sys/devices/platform/soc/fe300000.mmcnr/mmc_host/mmc1/mmc1:0001/mmc1:0001:1/.present": "" + }, + "links": { + "sys/class/net/eth0/device": "../../../devices/platform/scb/fd580000.ethernet", + "sys/class/net/wlan0/device": "../../../devices/platform/soc/fe300000.mmcnr/mmc_host/mmc1/mmc1:0001/mmc1:0001:1" + }, + "elements": [ + [ + "serial", + "10000000a1b2c3d4" + ], + [ + "mac", + "dc:a6:32:01:02:03" + ], + [ + "cpu", + "0x41|0xd08" + ] + ], + "id": "hw2:363598c3230281d8cc2cfcdd1b772a6957aa6b8fbd631712ba1b195be5d22318" + }, + { + "name": "x86-industrial-pc", + "files": { + "sys/class/net/enp1s0/address": "00:E0:4C:68:00:01\n", + "sys/class/net/enp1s0/addr_assign_type": "0\n", + "sys/class/net/enp2s0/address": "00:e0:4c:68:00:02\n", + "sys/class/net/enp2s0/addr_assign_type": "0\n", + "sys/class/net/enp3s0/address": "00:e0:4c:68:00:03\n", + "sys/class/net/enp3s0/addr_assign_type": "3\n", + "sys/class/net/enx00249b112233/address": "00:24:9b:11:22:33\n", + "sys/class/net/enx00249b112233/addr_assign_type": "0\n", + "sys/class/net/docker0/address": "02:42:ac:11:00:01\n", + "sys/class/net/docker0/addr_assign_type": "3\n", + "sys/class/dmi/id/product_uuid": "4C4C4544-0042-3510-8051-B2C04F4E3132\n", + "sys/class/dmi/id/board_serial": "CNWS2000900ABC\n", + "sys/class/dmi/id/sys_vendor": "Dell Inc.\n", + "sys/class/dmi/id/product_name": "OptiPlex 7090\n", + "proc/cpuinfo": "processor\t: 0\nvendor_id\t: GenuineIntel\ncpu family\t: 6\nmodel\t\t: 165\nmodel name\t: Intel(R) Core(TM) i5-10500T CPU @ 2.30GHz\nflags\t\t: fpu vme de pse tsc msr sse sse2\n\nprocessor\t: 1\nvendor_id\t: GenuineIntel\nmodel name\t: Intel(R) Core(TM) i5-10500T CPU @ 2.30GHz\n", + "proc/version": "Linux version 6.8.0-45-generic (buildd@lcy02-amd64-115) x86_64\n", + "sys/devices/pci0000:00/0000:00:1c.0/0000:01:00.0/removable": "fixed\n", + "sys/devices/pci0000:00/0000:00:1c.1/0000:02:00.0/removable": "unknown\n", + "sys/devices/pci0000:00/0000:00:1c.2/0000:03:00.0/.present": "", + "sys/devices/pci0000:00/0000:00:14.0/usb2/2-1/2-1:1.0/.present": "", + "sys/devices/pci0000:00/0000:00:14.0/usb2/2-1/removable": "removable\n", + "sys/devices/pci0000:00/0000:00:14.0/usb2/2-1/port/connect_type": "hotplug\n" + }, + "links": { + "sys/class/net/enp1s0/device": "../../../devices/pci0000:00/0000:00:1c.0/0000:01:00.0", + "sys/class/net/enp2s0/device": "../../../devices/pci0000:00/0000:00:1c.1/0000:02:00.0", + "sys/class/net/enp3s0/device": "../../../devices/pci0000:00/0000:00:1c.2/0000:03:00.0", + "sys/class/net/enx00249b112233/device": "../../../devices/pci0000:00/0000:00:14.0/usb2/2-1/2-1:1.0" + }, + "elements": [ + [ + "uuid", + "4c4c4544-0042-3510-8051-b2c04f4e3132" + ], + [ + "board", + "cnws2000900abc" + ], + [ + "mac", + "00:e0:4c:68:00:01" + ], + [ + "mac", + "00:e0:4c:68:00:02" + ], + [ + "cpu", + "genuineintel|intel(r) core(tm) i5-10500t cpu @ 2.30ghz" + ] + ], + "id": "hw2:5337b315679912a71e5f004a718544e1282e004f13acb0ac17c4c8142587a14c" + }, + { + "name": "vm", + "files": { + "sys/class/net/ens3/address": "00:16:3e:11:22:33\n", + "sys/class/net/ens3/addr_assign_type": "0\n", + "sys/class/dmi/id/product_uuid": "8f6c2a1e-5b3d-4c7a-9e21-0d4f6b8a1c33\n", + "sys/class/dmi/id/board_serial": "\n", + "sys/class/dmi/id/sys_vendor": "QEMU\n", + "sys/class/dmi/id/product_name": "Standard PC (Q35 + ICH9, 2009)\n", + "proc/cpuinfo": "processor\t: 0\nvendor_id\t: AuthenticAMD\nmodel name\t: AMD EPYC 7763 64-Core Processor\nflags\t\t: fpu vme sse sse2 hypervisor\n", + "sys/devices/pci0000:00/0000:00:03.0/virtio0/.present": "" + }, + "links": { + "sys/class/net/ens3/device": "../../../devices/pci0000:00/0000:00:03.0/virtio0" + }, + "elements": [ + [ + "uuid", + "8f6c2a1e-5b3d-4c7a-9e21-0d4f6b8a1c33" + ], + [ + "cpu", + "authenticamd|amd epyc 7763 64-core processor" + ] + ], + "id": "hw2:5e26377d20855444f09a7b40f13d8c4e00550166e98efbc6d1d01c05b51a5eb1" + }, + { + "name": "placeholders-only", + "files": { + "sys/class/dmi/id/product_uuid": "00000000-0000-0000-0000-000000000000\n", + "sys/class/dmi/id/board_serial": "To be filled by O.E.M.\n", + "proc/cpuinfo": "processor\t: 0\nvendor_id\t: GenuineIntel\nmodel name\t: Intel(R) Celeron(R) J4125 CPU @ 2.00GHz\n" + }, + "links": {}, + "elements": [ + [ + "cpu", + "genuineintel|intel(r) celeron(r) j4125 cpu @ 2.00ghz" + ] + ], + "id": null + }, + { + "name": "wsl2", + "files": { + "proc/sys/kernel/osrelease": "5.15.153.1-microsoft-standard-WSL2\n", + "sys/class/dmi/id/product_uuid": "1d4e3c2b-aa11-4c5d-9f0e-123456789abc\n", + "proc/cpuinfo": "processor\t: 0\nvendor_id\t: GenuineIntel\nmodel name\t: Intel(R) Core(TM) i7-1185G7 @ 3.00GHz\nflags\t\t: fpu hypervisor\n" + }, + "links": {}, + "elements": [], + "id": null + } + ] +} \ No newline at end of file diff --git a/tests/pytest/restapi/test_hardware_id_endpoint.py b/tests/pytest/restapi/test_hardware_id_endpoint.py new file mode 100644 index 00000000..cbe0e013 --- /dev/null +++ b/tests/pytest/restapi/test_hardware_id_endpoint.py @@ -0,0 +1,161 @@ +# SPDX-License-Identifier: GPL-3.0-only +# Copyright (c) 2026 Autonomy® + +"""The hardware ID record: which source the runtime advertises, the endpoint, and +when the runtime computes the record itself.""" + +import pathlib +import subprocess + +import pytest +from conftest import auth, create_user, token_for + +from webserver import hardware_id + +ID = "hw2:" + "a" * 64 +OTHER_ID = "hw2:" + "b" * 64 +RECORD = f"{ID}\nserial {'1' * 64}\nmac {'2' * 64}\ncpu {'3' * 64}\n" + + +@pytest.fixture() +def paths(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, pathlib.Path]: + stored = tmp_path / "data" / "hardware-id" + injected = tmp_path / "etc" / "hardware-id" + stored.parent.mkdir() + injected.parent.mkdir() + monkeypatch.setattr(hardware_id, "STORED_PATH", stored) + monkeypatch.setattr(hardware_id, "INJECTED_PATH", injected) + return {"stored": stored, "injected": injected} + + +def test_parse_record() -> None: + record = hardware_id.parse_record(RECORD, "device") + assert record is not None + assert record.value == ID + assert [e.kind for e in record.elements] == ["serial", "mac", "cpu"] + + +@pytest.mark.parametrize( + "text", + ["", "hw2:abc\n", ID.upper() + "\n", f"{ID}\nwifi {'1' * 64}\n", f"{ID}\nmac 12\n"], +) +def test_malformed_records_are_ignored(text: str) -> None: + assert hardware_id.parse_record(text, "device") is None + + +def test_stored_record_is_advertised(paths: dict[str, pathlib.Path]) -> None: + paths["stored"].write_text(RECORD) + record = hardware_id.current() + assert record is not None and record.value == ID and record.source == "device" + + +def test_injected_record_wins(paths: dict[str, pathlib.Path]) -> None: + paths["stored"].write_text(RECORD) + paths["injected"].write_text(RECORD.replace(ID, OTHER_ID)) + record = hardware_id.current() + assert record is not None and record.value == OTHER_ID and record.source == "orchestrator" + + +def test_no_record_means_none(paths: dict[str, pathlib.Path]) -> None: + assert hardware_id.current() is None + + +def test_endpoint_requires_login(client, paths) -> None: + assert client.get("/api/hardware-id").status_code == 401 + + +def test_endpoint_returns_the_record(client, admin_token, paths) -> None: + paths["stored"].write_text(RECORD) + body = client.get("/api/hardware-id", headers=auth(admin_token)).get_json() + assert body == { + "hardwareId": ID, + "source": "device", + "elements": [ + {"kind": "serial", "hash": "1" * 64}, + {"kind": "mac", "hash": "2" * 64}, + {"kind": "cpu", "hash": "3" * 64}, + ], + } + + +def test_endpoint_open_to_non_admin_users(client, admin_token, paths) -> None: + paths["stored"].write_text(RECORD) + assert ( + create_user(client, "operator", "operator-pass", admin_token, role="user").status_code + == 201 + ) + token = token_for(client, "operator", "operator-pass") + assert client.get("/api/hardware-id", headers=auth(token)).get_json()["hardwareId"] == ID + + +def test_endpoint_without_an_id(client, admin_token, paths) -> None: + body = client.get("/api/hardware-id", headers=auth(admin_token)).get_json() + assert body == {"hardwareId": None, "source": None, "elements": []} + + +class _Recorder: + def __init__(self) -> None: + self.calls: list[list[str]] = [] + + def __call__(self, args: list[str], **_kwargs: object) -> subprocess.CompletedProcess: + self.calls.append(args) + pathlib.Path(args[-1]).write_text(RECORD) + return subprocess.CompletedProcess(args, 0, "", "") + + +@pytest.fixture() +def recorder(monkeypatch: pytest.MonkeyPatch) -> _Recorder: + rec = _Recorder() + monkeypatch.setattr(hardware_id.subprocess, "run", rec) + monkeypatch.setattr(hardware_id.platform, "system", lambda: "Linux") + return rec + + +def _host(monkeypatch: pytest.MonkeyPatch, container: bool, bootloader: bool) -> None: + monkeypatch.setattr(hardware_id, "is_running_in_container", lambda: container) + if bootloader: + monkeypatch.setenv("OPENPLC_PERSISTENT_DATA_DIR", "/var/lib/openplc-runtime") + else: + monkeypatch.delenv("OPENPLC_PERSISTENT_DATA_DIR", raising=False) + + +def test_native_computes_when_missing(paths, recorder, monkeypatch) -> None: + _host(monkeypatch, container=False, bootloader=False) + hardware_id.ensure_stored() + assert recorder.calls == [ + ["bash", str(hardware_id.SCRIPT_PATH), "--store", str(paths["stored"])] + ] + assert hardware_id.current().value == ID + + +def test_bootloader_container_computes_when_missing(paths, recorder, monkeypatch) -> None: + _host(monkeypatch, container=True, bootloader=True) + hardware_id.ensure_stored() + assert len(recorder.calls) == 1 + + +def test_vplc_never_computes(paths, recorder, monkeypatch) -> None: + _host(monkeypatch, container=True, bootloader=False) + hardware_id.ensure_stored() + assert recorder.calls == [] + + +def test_stored_record_is_never_recomputed(paths, recorder, monkeypatch) -> None: + _host(monkeypatch, container=False, bootloader=False) + paths["stored"].write_text(RECORD) + hardware_id.ensure_stored() + assert recorder.calls == [] + + +def test_injected_record_skips_computation(paths, recorder, monkeypatch) -> None: + _host(monkeypatch, container=True, bootloader=True) + paths["injected"].write_text(RECORD) + hardware_id.ensure_stored() + assert recorder.calls == [] + + +def test_non_linux_never_computes(paths, recorder, monkeypatch) -> None: + _host(monkeypatch, container=False, bootloader=False) + monkeypatch.setattr(hardware_id.platform, "system", lambda: "Windows") + hardware_id.ensure_stored() + assert recorder.calls == [] diff --git a/webserver/app.py b/webserver/app.py index 6a662583..433b118b 100644 --- a/webserver/app.py +++ b/webserver/app.py @@ -27,7 +27,7 @@ import flask import flask_login -from webserver import project_snapshot +from webserver import hardware_id, project_snapshot from webserver.credentials import CertGen from webserver.debug_websocket import init_debug_websocket from webserver.discovery.discovery_routes import discovery_bp @@ -570,6 +570,8 @@ def run_https(): socketio = init_debug_websocket(app_restapi, runtime_manager.runtime_socket) + hardware_id.ensure_stored() + with app_restapi.app_context(): try: db.create_all() diff --git a/webserver/hardware_id.py b/webserver/hardware_id.py new file mode 100644 index 00000000..9240a28f --- /dev/null +++ b/webserver/hardware_id.py @@ -0,0 +1,116 @@ +# SPDX-License-Identifier: GPL-3.0-only +# Copyright (c) 2026 Autonomy® + +"""The hardware ID v2 this runtime advertises (spec: docs/HARDWARE_ID.md). + +Two sources, in order: + + * ``INJECTED_PATH`` -- a read-only record that orchestrator-agent places in a + vPLC container before its first boot. A vPLC cannot see the host hardware, + so this is the only ID it may advertise. + * ``STORED_PATH`` -- the record the installers wrote to the persistent data + directory (native install and bootloader container). + +Installers compute the record once. ``ensure_stored()`` covers devices installed +before the ID existed: on start it computes the record only when none is stored, +and only where the host hardware is visible (native, or the bootloader's +privileged container). It never runs inside a vPLC, where a partial view of the +hardware would produce an ID that differs from the host's. +""" + +import os +import platform +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path + +from webserver.config import PERSISTENT_DATA_DIR, is_running_in_container +from webserver.logger import get_logger + +logger, _ = get_logger("logger", use_buffer=True) + +SCRIPT_PATH = Path(__file__).resolve().parent.parent / "scripts" / "hardware-id.sh" +STORED_PATH = PERSISTENT_DATA_DIR / "hardware-id" +INJECTED_PATH = Path("/etc/openplc-runtime/hardware-id") +COMPUTE_TIMEOUT_S = 30 + +_ID_RE = re.compile(r"^hw2:[0-9a-f]{64}$") +_ELEMENT_RE = re.compile(r"^(serial|uuid|board|mac|cpu) ([0-9a-f]{64})$") + + +@dataclass(frozen=True) +class Element: + kind: str + hash: str + + +@dataclass(frozen=True) +class HardwareId: + value: str + elements: tuple[Element, ...] + source: str + + +def parse_record(text: str, source: str) -> HardwareId | None: + """Parse a record: the ID line, then one `` `` line per element.""" + lines = [line.strip() for line in text.splitlines() if line.strip()] + if not lines or not _ID_RE.match(lines[0]): + return None + elements = [] + for line in lines[1:]: + match = _ELEMENT_RE.match(line) + if not match: + return None + elements.append(Element(kind=match.group(1), hash=match.group(2))) + return HardwareId(value=lines[0], elements=tuple(elements), source=source) + + +def _read(path: Path, source: str) -> HardwareId | None: + try: + text = path.read_text(encoding="utf-8") + except FileNotFoundError: + return None + except OSError as exc: + logger.warning("Cannot read hardware ID record %s: %s", path, exc) + return None + record = parse_record(text, source) + if record is None: + logger.warning("Ignoring malformed hardware ID record %s", path) + return record + + +def current() -> HardwareId | None: + """The ID to advertise: the orchestrator's record in a vPLC, else the stored one.""" + return _read(INJECTED_PATH, "orchestrator") or _read(STORED_PATH, "device") + + +def _sees_host_hardware() -> bool: + if platform.system() != "Linux": + return False + if not is_running_in_container(): + return True + # The bootloader always sets this on the runtime container; vPLCs never do. + return bool(os.environ.get("OPENPLC_PERSISTENT_DATA_DIR")) + + +def ensure_stored() -> None: + """Compute and store the record once, when this runtime sees the host and has none.""" + if INJECTED_PATH.exists() or STORED_PATH.exists() or not _sees_host_hardware(): + return + try: + result = subprocess.run( + ["bash", str(SCRIPT_PATH), "--store", str(STORED_PATH)], + capture_output=True, + text=True, + timeout=COMPUTE_TIMEOUT_S, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + logger.warning("Hardware ID computation failed: %s", exc) + return + if result.returncode == 0: + record = _read(STORED_PATH, "device") + logger.info("Hardware ID stored: %s", record.value if record else "unreadable") + else: + logger.info("No hardware ID can be derived on this device (exit %d)", result.returncode) diff --git a/webserver/restapi.py b/webserver/restapi.py index 5b1a0e86..902c6886 100644 --- a/webserver/restapi.py +++ b/webserver/restapi.py @@ -23,7 +23,7 @@ from werkzeug.security import check_password_hash, generate_password_hash import webserver.config -from webserver import project_snapshot +from webserver import hardware_id, project_snapshot from webserver.logger import get_logger from webserver.version import MIN_EDITOR_VERSION, RUNTIME_VERSION @@ -794,6 +794,58 @@ def stream(): return current_app.response_class(stream(), mimetype="application/json") +@restapi_bp.route("/hardware-id", methods=["GET"]) +@jwt_required() +def get_hardware_id(): + """Return the hardware ID v2 this runtime is bound to, with its element hashes. + + Any authenticated user. ``hardwareId`` is null when no ID can be derived on + this device. ``source`` is ``device`` for an ID computed on this hardware and + ``orchestrator`` for the host ID orchestrator-agent placed in a vPLC. + --- + tags: + - Runtime + security: + - BearerAuth: [] + responses: + 200: + description: Hardware ID retrieved + schema: + type: object + properties: + hardwareId: + type: string + description: hw2 followed by 64 lowercase hex characters, or null + source: + type: string + description: device or orchestrator, or null + elements: + type: array + items: + type: object + properties: + kind: + type: string + hash: + type: string + 401: + description: Authentication required + """ + record = hardware_id.current() + if record is None: + return jsonify({"hardwareId": None, "source": None, "elements": []}), 200 + return ( + jsonify( + { + "hardwareId": record.value, + "source": record.source, + "elements": [{"kind": e.kind, "hash": e.hash} for e in record.elements], + } + ), + 200, + ) + + @restapi_bp.route("/login", methods=["POST"]) def login(): """