admin/provisioning-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: persist hardware snapshots during activation

Browse Source
This commit is contained in:
Thomas Stallinger 2026-07-13 09:00:02 +02:00
parent ba2dccff57
commit 9c773e7b9b
1 changed files with 124 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
app.py
View File

@ -23,6 +23,7 @@ class ActivationRequest(BaseModel):
hostname: str | None = None
machine_id: str | None = None
client_version: str | None = None
hardware: dict
def load_config() -> dict:
@ -138,6 +139,122 @@ def load_or_create_device(
return device, created
def store_hardware_snapshot(
device_id: str,
hardware: dict,
) -> str:
snapshot_id = uuid4()
identity = hardware.get("identity", {})
system = hardware.get("system", {})
mainboard = hardware.get("mainboard", {})
firmware = hardware.get("firmware", {})
security = hardware.get("security", {})
cpu = system.get("cpu", {})
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO hardware_snapshots (
id,
device_id,
architecture,
manufacturer,
product_name,
product_version,
system_uuid,
system_serial,
board_vendor,
board_name,
board_serial,
bios_vendor,
bios_version,
boot_mode,
secure_boot,
tpm_version,
cpu_model,
cpu_logical_count,
memory_bytes
)
VALUES (
%s, %s, %s, %s, %s,
%s, %s, %s, %s, %s,
%s, %s, %s, %s, %s,
%s, %s, %s, %s
)
""",
(
snapshot_id,
device_id,
system["architecture"],
system.get("manufacturer"),
system.get("product_name"),
system.get("product_version"),
identity.get("system_uuid"),
identity.get("system_serial"),
mainboard.get("vendor"),
mainboard.get("name"),
identity.get("board_serial"),
firmware.get("bios_vendor"),
firmware.get("bios_version"),
firmware.get("boot_mode"),
firmware.get("secure_boot"),
security.get("tpm_version"),
cpu.get("model"),
cpu.get("logical_count"),
system.get("memory_bytes"),
),
)
for interface in hardware.get("network_interfaces", []):
cur.execute(
"""
INSERT INTO network_interfaces (
id,
hardware_snapshot_id,
name,
type,
mac_address
)
VALUES (%s, %s, %s, %s, %s)
""",
(
uuid4(),
snapshot_id,
interface["name"],
interface["type"],
interface["mac"],
),
)
for storage_device in hardware.get("storage_devices", []):
cur.execute(
"""
INSERT INTO storage_devices (
id,
hardware_snapshot_id,
name,
model,
serial,
size_bytes,
transport
)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""",
(
uuid4(),
snapshot_id,
storage_device["name"],
storage_device.get("model"),
storage_device.get("serial"),
storage_device["size_bytes"],
storage_device.get("transport"),
),
)
return str(snapshot_id)
@app.get("/health")
def health():
return {
@ -156,6 +273,8 @@ def activate(payload: ActivationRequest, request: Request):
device = None
device_created = False
hardware_snapshot_id = None
if success:
organization_id = activation_entry["customer"]["organization_id"]
@ -164,6 +283,10 @@ def activate(payload: ActivationRequest, request: Request):
payload.device_fingerprint,
payload.hostname,
)
hardware_snapshot_id = store_hardware_snapshot(
device["id"],
payload.hardware,
)
write_log({
"timestamp": datetime.now(timezone.utc).isoformat(),
@ -198,6 +321,7 @@ def activate(payload: ActivationRequest, request: Request):
"fingerprint": device["device_fingerprint"],
"hostname": device["hostname"],
"created": device_created,
"hardware_snapshot_id": hardware_snapshot_id,
},
"customer": activation_entry["customer"],