845 lines
36 KiB
Python
Executable File
845 lines
36 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Hawkbit DDI Client — lightweight pure-Python OTA update agent.
|
|
=================================================================
|
|
No dependencies beyond Python 3 standard library.
|
|
Runs on Android (adb shell) and Linux hosts.
|
|
|
|
DDI API version: v1
|
|
Ref: https://github.com/eclipse-hawkbit/hawkbit
|
|
|
|
Capabilities:
|
|
• Poll Hawkbit for deployment actions
|
|
• Resumable (Range-request) artifact download
|
|
• SHA256 file integrity verification
|
|
• Status progression: DOWNLOAD → DOWNLOADED → PROCEEDING → FINISHED / FAILURE
|
|
• Auto-detects device serial number as controllerId (Android ro.serialno, DMI, machine-id)
|
|
• Reports device attributes (osType, platform, hwRevision, swVersion, kernel…) to Hawkbit
|
|
• Authentication: GatewayToken or TargetToken (also from file: --gateway-token @/path/to/key)
|
|
|
|
Quick start:
|
|
# Gateway token
|
|
ddi-client.py --base-url https://hawkbit:8443 --tenant DEFAULT --gateway-token mykey
|
|
|
|
# Target token (per-device)
|
|
ddi-client.py --base-url http://10.0.0.1:8080 --tenant DEFAULT --target-token abc123
|
|
|
|
# One-shot poll
|
|
ddi-client.py --base-url http://localhost:8080 --tenant DEFAULT --gateway-token k --once
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import ssl
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
# ── logging ─────────────────────────────────────────────────────────────────
|
|
logger = logging.getLogger("hawkbit-ddi")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# HawkbitDDIClient
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
class HawkbitDDIClient:
|
|
"""Lightweight DDI (Direct Device Integration) client for Eclipse Hawkbit."""
|
|
|
|
# DDI execution statuses (mapped to Hawkbit action states)
|
|
EXEC_DOWNLOAD = "download"
|
|
EXEC_DOWNLOADED = "downloaded"
|
|
EXEC_PROCEEDING = "proceeding"
|
|
EXEC_CLOSED = "closed"
|
|
EXEC_CANCELED = "canceled"
|
|
EXEC_REJECTED = "rejected"
|
|
EXEC_SCHEDULED = "scheduled"
|
|
EXEC_RESUMED = "resumed"
|
|
|
|
RESULT_SUCCESS = "success"
|
|
RESULT_FAILURE = "failure"
|
|
RESULT_NONE = "none"
|
|
|
|
# ── constructor ──────────────────────────────────────────────────────
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
tenant: str,
|
|
controller_id: Optional[str] = None,
|
|
auth_header: Optional[str] = None,
|
|
download_dir: str = "/tmp/hawkbit",
|
|
verify_sha256: bool = True,
|
|
polling_interval: int = 60,
|
|
ssl_verify: bool = True,
|
|
resume_download: bool = True,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.tenant = tenant
|
|
self.controller_id = controller_id or self._detect_device_sn()
|
|
self.auth_header = auth_header
|
|
self.download_dir = download_dir
|
|
self.verify_sha256 = verify_sha256
|
|
self.polling_interval = polling_interval
|
|
self.ssl_verify = ssl_verify
|
|
self.resume_download = resume_download
|
|
self._attrs_reported = False
|
|
|
|
os.makedirs(self.download_dir, exist_ok=True)
|
|
|
|
# ── device serial detection ───────────────────────────────────────────
|
|
@staticmethod
|
|
def _detect_device_sn() -> str:
|
|
"""Best-effort device serial number detection.
|
|
|
|
Priority: Android ro.serialno → DMI product_serial → /etc/machine-id → hostname
|
|
"""
|
|
# 1. Android property
|
|
try:
|
|
out = subprocess.run(
|
|
["getprop", "ro.serialno"],
|
|
capture_output=True, text=True, timeout=2,
|
|
)
|
|
sn = out.stdout.strip()
|
|
if sn:
|
|
logger.info("Detected Android serial: %s", sn)
|
|
return sn
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
pass
|
|
|
|
# 2. Linux DMI product serial
|
|
for p in ("/sys/class/dmi/id/product_serial",
|
|
"/sys/devices/virtual/dmi/id/product_serial"):
|
|
try:
|
|
with open(p) as fh:
|
|
sn = fh.read().strip()
|
|
if sn and sn not in ("0", "Not Specified", "None", ""):
|
|
logger.info("Detected DMI serial: %s", sn)
|
|
return sn
|
|
except (FileNotFoundError, PermissionError):
|
|
pass
|
|
|
|
# 3. /etc/machine-id (systemd)
|
|
try:
|
|
with open("/etc/machine-id") as fh:
|
|
mid = fh.read().strip()
|
|
if mid:
|
|
logger.info("Using machine-id: %s", mid)
|
|
return mid
|
|
except (FileNotFoundError, PermissionError):
|
|
pass
|
|
|
|
# 4. Fallback — hostname
|
|
import socket
|
|
hostname = socket.gethostname()
|
|
logger.warning("No serial found; falling back to hostname: %s", hostname)
|
|
return hostname
|
|
|
|
# ── device attributes detection ─────────────────────────────────────
|
|
@staticmethod
|
|
def _detect_device_attrs() -> Dict[str, str]:
|
|
"""Collect device metadata for Hawkbit target attributes.
|
|
|
|
These attributes are used by Hawkbit target filters to auto-assign updates.
|
|
Returns a dict of key-value pairs.
|
|
"""
|
|
attrs: Dict[str, str] = {}
|
|
|
|
# ── OS type ─────────────────────────────────────────────────
|
|
attrs["hostname"] = HawkbitDDIClient._safe_hostname()
|
|
|
|
if os.path.exists("/system/build.prop"):
|
|
attrs["osType"] = "android"
|
|
else:
|
|
attrs["osType"] = "linux"
|
|
|
|
# ── platform / architecture ─────────────────────────────────
|
|
try:
|
|
import platform
|
|
attrs["platform"] = platform.machine()
|
|
except Exception:
|
|
pass
|
|
|
|
# ── kernel version ──────────────────────────────────────────
|
|
try:
|
|
attrs["kernel"] = os.uname().release
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Android: detailed attributes ────────────────────────────
|
|
if attrs.get("osType") == "android":
|
|
for prop, attr_key in (
|
|
("ro.build.version.release", "androidVersion"),
|
|
("ro.product.model", "productModel"),
|
|
("ro.product.manufacturer", "manufacturer"),
|
|
("ro.build.version.incremental", "buildNumber"),
|
|
("ro.build.fingerprint", "buildFingerprint"),
|
|
):
|
|
try:
|
|
out = subprocess.run(
|
|
["getprop", prop], capture_output=True, text=True, timeout=2,
|
|
)
|
|
val = out.stdout.strip()
|
|
if val:
|
|
attrs[attr_key] = val
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
pass
|
|
|
|
# ── Linux: DMI hardware info ────────────────────────────────
|
|
if attrs.get("osType") == "linux":
|
|
for path, attr_key in (
|
|
("/sys/class/dmi/id/product_name", "productName"),
|
|
("/sys/class/dmi/id/product_version", "productVersion"),
|
|
("/sys/class/dmi/id/product_sku", "productSku"),
|
|
("/sys/class/dmi/id/sys_vendor", "manufacturer"),
|
|
):
|
|
try:
|
|
with open(path) as fh:
|
|
val = fh.read().strip()
|
|
if val and val not in ("0", "Not Specified", "None", "", "System Product Name"):
|
|
attrs[attr_key] = val
|
|
except (FileNotFoundError, PermissionError):
|
|
pass
|
|
|
|
# hwRevision: combine product name + version
|
|
if "productName" in attrs:
|
|
rev = attrs["productName"]
|
|
if "productVersion" in attrs:
|
|
rev += " " + attrs["productVersion"]
|
|
attrs["hwRevision"] = rev
|
|
|
|
# ── current software version (if installed) ─────────────────
|
|
version_file = "/etc/hawkbit/current_version"
|
|
try:
|
|
with open(version_file) as fh:
|
|
v = fh.read().strip()
|
|
if v:
|
|
attrs["swVersion"] = v
|
|
except (FileNotFoundError, PermissionError):
|
|
pass
|
|
|
|
# ── merge user-supplied attrs from file ─────────────────────
|
|
attrs_file = "/etc/hawkbit/device_attrs.json"
|
|
try:
|
|
with open(attrs_file) as fh:
|
|
extra = json.load(fh)
|
|
if isinstance(extra, dict):
|
|
attrs.update({k: str(v) for k, v in extra.items()})
|
|
except (FileNotFoundError, PermissionError, json.JSONDecodeError):
|
|
pass
|
|
|
|
return attrs
|
|
|
|
@staticmethod
|
|
def _safe_hostname() -> str:
|
|
try:
|
|
import socket
|
|
return socket.gethostname()
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
# ── report device attributes to Hawkbit ──────────────────────────────
|
|
def report_attributes(self) -> bool:
|
|
"""PUT /{tenant}/controller/v1/{controllerId}/configData
|
|
|
|
Reports device metadata so Hawkbit can use target filters to assign updates.
|
|
Uses ``mode: merge`` — only updates the keys sent, preserves existing attributes.
|
|
"""
|
|
attrs = self._detect_device_attrs()
|
|
url = self._ddi_url(f"{self.controller_id}/configData")
|
|
payload = {"mode": "merge", "data": attrs}
|
|
logger.info("Reporting %d device attributes → %s", len(attrs), url)
|
|
logger.debug("Attributes: %s", json.dumps(attrs))
|
|
resp = self._http(url, method="PUT", data=payload)
|
|
return resp is not None
|
|
def _ddi_url(self, path: str) -> str:
|
|
"""Build a DDI v1 URL: <base>/<tenant>/controller/v1/<path>"""
|
|
return f"{self.base_url}/{self.tenant}/controller/v1/{path}"
|
|
|
|
# ── low-level HTTP ────────────────────────────────────────────────────
|
|
def _http(
|
|
self,
|
|
url: str,
|
|
method: str = "GET",
|
|
data: Any = None,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
stream: bool = False,
|
|
):
|
|
"""Make an HTTP request. Returns decoded JSON, or raw file handle when
|
|
``stream=True``. Returns ``None`` on HTTP errors."""
|
|
req_headers: Dict[str, str] = {"Accept": "application/hal+json, application/json"}
|
|
if self.auth_header:
|
|
req_headers["Authorization"] = self.auth_header
|
|
if data is not None:
|
|
req_headers["Content-Type"] = "application/json"
|
|
if headers:
|
|
req_headers.update(headers)
|
|
|
|
body = None if data is None else json.dumps(data).encode("utf-8")
|
|
|
|
ctx = None
|
|
if not self.ssl_verify:
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
req = urllib.request.Request(url, data=body, headers=req_headers, method=method)
|
|
|
|
try:
|
|
resp = urllib.request.urlopen(req, context=ctx)
|
|
except urllib.error.HTTPError as exc:
|
|
err_body = exc.read().decode("utf-8", errors="replace")[:500]
|
|
logger.error("HTTP %d %s %s → %s", exc.code, method, url, err_body)
|
|
return None
|
|
|
|
if stream:
|
|
return resp
|
|
|
|
raw = resp.read()
|
|
if not raw:
|
|
return {}
|
|
return json.loads(raw.decode("utf-8"))
|
|
|
|
# ── helper: extract best download link from artifact HAL links ─────────
|
|
@staticmethod
|
|
def _pick_download_url(artifact: dict) -> Optional[str]:
|
|
"""Pick the preferred download URL from the artifact's _links.
|
|
|
|
Priority: download (HTTPS) → download-http (HTTP) → construct from filename
|
|
"""
|
|
links = artifact.get("_links", {})
|
|
if not isinstance(links, dict):
|
|
return None
|
|
|
|
# HAL links are keyed by rel; values can be a single object or a list.
|
|
for rel in ("download", "download-http"):
|
|
entry = links.get(rel)
|
|
if isinstance(entry, list):
|
|
for item in entry:
|
|
if isinstance(item, dict) and "href" in item:
|
|
return item["href"]
|
|
elif isinstance(entry, dict) and "href" in entry:
|
|
return entry["href"]
|
|
return None
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Public API
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
# ── poll ──────────────────────────────────────────────────────────────
|
|
def poll(self) -> Optional[dict]:
|
|
"""GET /{tenant}/controller/v1/{controllerId}
|
|
|
|
Returns parsed controller base or None on failure.
|
|
"""
|
|
url = self._ddi_url(self.controller_id)
|
|
logger.debug("Polling GET %s", url)
|
|
body = self._http(url)
|
|
if body is None:
|
|
return None
|
|
return self._parse_poll(body)
|
|
|
|
def _parse_poll(self, body: dict) -> dict:
|
|
result: Dict[str, Any] = {
|
|
"config": body.get("config", {}),
|
|
"links": {},
|
|
}
|
|
|
|
links = body.get("_links", {})
|
|
if isinstance(links, dict):
|
|
for rel in ("deploymentBase", "cancelAction",
|
|
"confirmationBase", "installedBase", "configData"):
|
|
entry = links.get(rel)
|
|
href = None
|
|
if isinstance(entry, dict):
|
|
href = entry.get("href")
|
|
elif isinstance(entry, list) and entry:
|
|
href = entry[0].get("href") if isinstance(entry[0], dict) else None
|
|
if href:
|
|
result["links"][rel] = href
|
|
|
|
# Adopt server-suggested polling interval
|
|
sleep_str = body.get("config", {}).get("polling", {}).get("sleep", "")
|
|
if sleep_str:
|
|
try:
|
|
h, m, s = (int(x) for x in sleep_str.split(":"))
|
|
seconds = h * 3600 + m * 60 + s
|
|
if seconds > 0:
|
|
self.polling_interval = seconds
|
|
except ValueError:
|
|
pass
|
|
|
|
return result
|
|
|
|
# ── get deployment base ───────────────────────────────────────────────
|
|
def get_deployment(self, deployment_url: str) -> Optional[dict]:
|
|
"""GET /{tenant}/controller/v1/{controllerId}/deploymentBase/{actionId}
|
|
|
|
Returns structured deployment info or None.
|
|
"""
|
|
logger.debug("Fetching deployment: %s", deployment_url)
|
|
body = self._http(deployment_url)
|
|
if body is None:
|
|
return None
|
|
return self._parse_deployment(body)
|
|
|
|
def _parse_deployment(self, body: dict) -> dict:
|
|
info = body.get("deployment", {})
|
|
|
|
chunks = []
|
|
for ch in info.get("chunks", []):
|
|
artifacts = []
|
|
for art in ch.get("artifacts", []):
|
|
artifacts.append({
|
|
"filename": art["filename"],
|
|
"size": art.get("size", 0),
|
|
"hashes": art.get("hashes", {}),
|
|
"_links": art.get("_links", {}),
|
|
})
|
|
chunks.append({
|
|
"part": ch.get("part", ""),
|
|
"version": ch.get("version", ""),
|
|
"name": ch.get("name", ""),
|
|
"artifacts": artifacts,
|
|
})
|
|
|
|
return {
|
|
"action_id": body.get("id"),
|
|
"chunks": chunks,
|
|
"handling": {
|
|
"download": info.get("download", "attempt"),
|
|
"update": info.get("update", "attempt"),
|
|
"maintenanceWindow": info.get("maintenanceWindow"),
|
|
},
|
|
}
|
|
|
|
# ── download artifact ─────────────────────────────────────────────────
|
|
def download_artifact(self, artifact: dict, action_id: str = "") -> Optional[str]:
|
|
"""Download one artifact with Range-request resume and SHA256 verification.
|
|
|
|
Returns the local file path on success, or None on failure.
|
|
"""
|
|
filename = artifact["filename"]
|
|
expected_sha256 = artifact.get("hashes", {}).get("sha256")
|
|
expected_size = artifact.get("size", 0)
|
|
|
|
# Resolve the download URL
|
|
download_url = self._pick_download_url(artifact)
|
|
if not download_url:
|
|
# Fallback — construct from DDI path
|
|
# Try to extract smId from any artifact HAL link
|
|
logger.warning("No download link in artifact — using fallback URL")
|
|
import re
|
|
sm_id = "0"
|
|
links = artifact.get("_links", {})
|
|
for entry in links.values():
|
|
href = None
|
|
if isinstance(entry, dict) and "href" in entry:
|
|
href = entry["href"]
|
|
elif isinstance(entry, list):
|
|
for item in entry:
|
|
if isinstance(item, dict) and "href" in item:
|
|
href = item["href"]
|
|
break
|
|
if href:
|
|
m = re.search(r"/softwaremodules/(\d+)/", href)
|
|
if m:
|
|
sm_id = m.group(1)
|
|
break
|
|
download_url = self._ddi_url(
|
|
f"{self.controller_id}/softwaremodules/{sm_id}/artifacts/{filename}"
|
|
)
|
|
|
|
local_path = os.path.join(self.download_dir, filename)
|
|
tmp_path = local_path + ".tmp"
|
|
|
|
# Already downloaded and verified?
|
|
if os.path.exists(local_path):
|
|
if self._sha256_check(local_path, expected_sha256):
|
|
logger.info("Already cached & verified: %s", local_path)
|
|
return local_path
|
|
logger.warning("Cached file fails SHA256 — re-downloading")
|
|
|
|
# ── resume ────────────────────────────────────────────────────
|
|
downloaded = 0
|
|
if self.resume_download and os.path.exists(tmp_path):
|
|
downloaded = os.path.getsize(tmp_path)
|
|
logger.info("Resuming from byte %d", downloaded)
|
|
|
|
req_headers: Dict[str, str] = {}
|
|
if downloaded > 0:
|
|
req_headers["Range"] = f"bytes={downloaded}-"
|
|
|
|
logger.info("Downloading %s (%s bytes) → %s",
|
|
filename, expected_size or "?", tmp_path)
|
|
|
|
resp = self._http(download_url, headers=req_headers, stream=True)
|
|
if resp is None:
|
|
logger.error("Download request failed")
|
|
return None
|
|
|
|
# Determine total size (from Content-Range header)
|
|
total = expected_size
|
|
cr = resp.headers.get("Content-Range", "")
|
|
if cr:
|
|
try:
|
|
total = int(cr.rsplit("/", 1)[-1])
|
|
except ValueError:
|
|
pass
|
|
|
|
mode = "ab" if downloaded > 0 else "wb"
|
|
try:
|
|
with open(tmp_path, mode) as fh:
|
|
while True:
|
|
chunk = resp.read(64 * 1024)
|
|
if not chunk:
|
|
break
|
|
fh.write(chunk)
|
|
downloaded += len(chunk)
|
|
if total:
|
|
pct = downloaded / total * 100
|
|
logger.debug(" %6.1f %% (%d / %d)", pct, downloaded, total)
|
|
except Exception as exc:
|
|
logger.error("Download interrupted: %s (tmp kept for resume)", exc)
|
|
return None
|
|
|
|
logger.info("Download finished — %d bytes", downloaded)
|
|
|
|
# SHA256
|
|
if self.verify_sha256 and expected_sha256:
|
|
if not self._sha256_check(tmp_path, expected_sha256):
|
|
logger.error("SHA256 MISMATCH for %s", filename)
|
|
return None
|
|
|
|
# Atomic rename
|
|
if os.path.exists(local_path):
|
|
os.remove(local_path)
|
|
os.rename(tmp_path, local_path)
|
|
return local_path
|
|
|
|
@staticmethod
|
|
def _sha256_check(filepath: str, expected: Optional[str]) -> bool:
|
|
if not expected:
|
|
return True # nothing to compare against
|
|
h = hashlib.sha256()
|
|
try:
|
|
with open(filepath, "rb") as fh:
|
|
while True:
|
|
chunk = fh.read(64 * 1024)
|
|
if not chunk:
|
|
break
|
|
h.update(chunk)
|
|
except OSError:
|
|
return False
|
|
actual = h.hexdigest()
|
|
ok = actual.lower() == expected.lower()
|
|
if ok:
|
|
logger.info("SHA256 verified %s", actual)
|
|
else:
|
|
logger.error("SHA256 mismatch!\n expected: %s\n actual: %s", expected, actual)
|
|
return ok
|
|
|
|
# ── send feedback ─────────────────────────────────────────────────────
|
|
def send_feedback(
|
|
self,
|
|
deployment_url: str,
|
|
execution: str,
|
|
result: str = RESULT_NONE,
|
|
progress: Optional[Tuple[int, int]] = None,
|
|
details: Optional[List[str]] = None,
|
|
code: Optional[int] = None,
|
|
) -> Optional[dict]:
|
|
"""POST …/feedback — report execution status to the server.
|
|
|
|
Args:
|
|
deployment_url: full URL to the deployment base / cancel action / confirmation base.
|
|
execution: one of the EXEC_* constants.
|
|
result: ``success``, ``failure``, or ``none``.
|
|
progress: ``(cnt, of)`` e.g. ``(2, 5)``.
|
|
details: human-readable messages (shows in Hawkbit UI).
|
|
code: optional numeric status code.
|
|
"""
|
|
# strip query params — they belong to GET, not POST /feedback
|
|
base = deployment_url.split("?")[0]
|
|
url = base + "/feedback"
|
|
|
|
result_obj: Dict[str, Any] = {"finished": result}
|
|
if progress:
|
|
result_obj["progress"] = {"cnt": progress[0], "of": progress[1]}
|
|
|
|
status: Dict[str, Any] = {
|
|
"execution": execution,
|
|
"result": result_obj,
|
|
"details": details or [],
|
|
}
|
|
if code is not None:
|
|
status["code"] = code
|
|
|
|
payload = {
|
|
"timestamp": int(time.time() * 1000),
|
|
"status": status,
|
|
}
|
|
|
|
logger.info("Feedback %-12s / %-7s → %s", execution, result, url)
|
|
return self._http(url, method="POST", data=payload)
|
|
|
|
# ── process deployment ────────────────────────────────────────────────
|
|
def process_deployment(self, deployment_url: str) -> bool:
|
|
"""Run the full deployment lifecycle: fetch metadata → download → verify → report."""
|
|
|
|
dep = self.get_deployment(deployment_url)
|
|
if not dep:
|
|
logger.error("Cannot fetch deployment details")
|
|
self.send_feedback(deployment_url, self.EXEC_CLOSED, self.RESULT_FAILURE,
|
|
details=["Failed to fetch deployment base"])
|
|
return False
|
|
|
|
action_id = dep["action_id"]
|
|
handling = dep["handling"]
|
|
logger.info("══ Action %s ══ download=%s update=%s",
|
|
action_id, handling["download"], handling["update"])
|
|
|
|
# 1. DOWNLOAD phase — report start
|
|
self.send_feedback(deployment_url, self.EXEC_DOWNLOAD, self.RESULT_NONE,
|
|
details=["Starting download"])
|
|
|
|
# 2. Fetch every artifact
|
|
downloaded: List[str] = []
|
|
for chunk in dep["chunks"]:
|
|
logger.info("Chunk: %s v%s (%d artifacts)",
|
|
chunk["name"], chunk["version"], len(chunk["artifacts"]))
|
|
for idx, art in enumerate(chunk["artifacts"], 1):
|
|
# Per-artifact progress
|
|
self.send_feedback(
|
|
deployment_url, self.EXEC_PROCEEDING, self.RESULT_NONE,
|
|
progress=(idx, len(chunk["artifacts"])),
|
|
details=[f"Downloading {art['filename']}"],
|
|
)
|
|
local = self.download_artifact(art, action_id)
|
|
if local is None:
|
|
self.send_feedback(
|
|
deployment_url, self.EXEC_CLOSED, self.RESULT_FAILURE,
|
|
details=[f"Download failed for {art['filename']}"],
|
|
)
|
|
return False
|
|
downloaded.append(local)
|
|
|
|
# 3. DOWNLOADED — all artifacts on disk, hashes ok
|
|
self.send_feedback(deployment_url, self.EXEC_DOWNLOADED, self.RESULT_NONE,
|
|
details=[f"Downloaded {len(downloaded)} artifact(s)"])
|
|
|
|
# 4. PROCEEDING — install
|
|
self.send_feedback(deployment_url, self.EXEC_PROCEEDING, self.RESULT_NONE,
|
|
details=["Installing update …"])
|
|
|
|
success = self._install(downloaded, handling)
|
|
|
|
# 5. CLOSED — success or failure
|
|
if success:
|
|
self.send_feedback(deployment_url, self.EXEC_CLOSED, self.RESULT_SUCCESS,
|
|
details=["Installation complete"], code=200)
|
|
else:
|
|
self.send_feedback(deployment_url, self.EXEC_CLOSED, self.RESULT_FAILURE,
|
|
details=["Installation failed"], code=500)
|
|
return success
|
|
|
|
# ── install hook — override this for your device ──────────────────────
|
|
def _install(self, files: List[str], handling: dict) -> bool:
|
|
"""Apply the downloaded update. **Override in subclass or monkey-patch.**"""
|
|
logger.info("Install hook — files: %s", files)
|
|
# Default: look for update.zip and report success.
|
|
for fp in files:
|
|
if os.path.basename(fp).lower().endswith(".zip"):
|
|
# Example: subprocess.run(["unzip", "-o", fp, "-d", "/data/ota"], check=True)
|
|
logger.info("Would install: %s", fp)
|
|
return True
|
|
return bool(files)
|
|
|
|
# ── cancel action ─────────────────────────────────────────────────────
|
|
def process_cancel(self, cancel_url: str) -> None:
|
|
"""Acknowledge a cancel action."""
|
|
self.send_feedback(cancel_url + "/feedback", self.EXEC_CANCELED,
|
|
self.RESULT_NONE, details=["Cancel acknowledged"])
|
|
|
|
# ── confirmation (auto-confirm for headless devices) ──────────────────
|
|
def process_confirmation(self, confirmation_url: str) -> None:
|
|
"""Auto-confirm a WAITING_FOR_CONFIRMATION action."""
|
|
logger.info("Auto-confirming: %s", confirmation_url)
|
|
body = {
|
|
"confirmation": "confirmed",
|
|
"code": 200,
|
|
"details": ["Auto-confirmed by DDI client"],
|
|
}
|
|
self._http(confirmation_url + "/feedback", method="POST", data=body)
|
|
|
|
# ── main loop ─────────────────────────────────────────────────────────
|
|
def run(self, *, daemon: bool = True, max_cycles: Optional[int] = None) -> bool:
|
|
"""Run the polling loop.
|
|
|
|
Args:
|
|
daemon: If False, poll once and return.
|
|
max_cycles: Maximum number of poll cycles (None = forever).
|
|
"""
|
|
logger.info("╔══════════════════════════════════════════════════════════╗")
|
|
logger.info("║ Hawkbit DDI Client v1.0 ║")
|
|
logger.info("╠══════════════════════════════════════════════════════════╣")
|
|
logger.info("║ Server: %s", self.base_url)
|
|
logger.info("║ Tenant: %s", self.tenant)
|
|
logger.info("║ Device: %s", self.controller_id)
|
|
logger.info("║ Auth: %s",
|
|
self.auth_header.split(" ", 1)[0] if self.auth_header else "none")
|
|
logger.info("╚══════════════════════════════════════════════════════════╝")
|
|
|
|
cycle = 0
|
|
while max_cycles is None or cycle < max_cycles:
|
|
cycle += 1
|
|
logger.info("── Poll cycle %d ──", cycle)
|
|
|
|
poll_result = self.poll()
|
|
if poll_result is None:
|
|
logger.warning("Poll failed; retrying in %ds", min(self.polling_interval, 30))
|
|
if not daemon:
|
|
return False
|
|
time.sleep(min(self.polling_interval, 30))
|
|
continue
|
|
|
|
# ── report device attributes after first successful poll ──
|
|
# Target is auto-created by GET /controller/v1/{SN} (findOrRegister…);
|
|
# configData requires an existing target → poll first, then report.
|
|
if not self._attrs_reported:
|
|
self.report_attributes()
|
|
self._attrs_reported = True
|
|
|
|
links = poll_result.get("links", {})
|
|
|
|
if "deploymentBase" in links:
|
|
self.process_deployment(links["deploymentBase"])
|
|
|
|
elif "cancelAction" in links:
|
|
self.process_cancel(links["cancelAction"])
|
|
|
|
elif "confirmationBase" in links:
|
|
self.process_confirmation(links["confirmationBase"])
|
|
|
|
if "deploymentBase" not in links and "cancelAction" not in links and "confirmationBase" not in links:
|
|
logger.debug("No pending action.")
|
|
|
|
if not daemon:
|
|
return True
|
|
|
|
logger.info("Sleeping %ds …", self.polling_interval)
|
|
time.sleep(self.polling_interval)
|
|
|
|
return True
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# CLI entry point
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
def main(argv: Optional[List[str]] = None) -> None:
|
|
ap = argparse.ArgumentParser(
|
|
description="Hawkbit DDI Client — lightweight OTA agent (pure Python)",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Gateway token, poll every 30s
|
|
%(prog)s -u https://hawkbit.example.com -t DEFAULT --gateway-token s3cret -i 30
|
|
|
|
# Target token, one-shot
|
|
%(prog)s -u http://10.0.0.1:8080 -t DEFAULT --target-token tok --once
|
|
|
|
# Custom controller-id, insecure SSL, custom download dir
|
|
%(prog)s -u https://hawkbit:8443 -t prod --gateway-token k \\
|
|
--controller-id edge-042 -d /data/ota --no-ssl-verify -v
|
|
""",
|
|
)
|
|
|
|
# ── required ──────────────────────────────────────────────────────
|
|
ap.add_argument("-u", "--base-url", required=True,
|
|
help="Hawkbit base URL, e.g. https://hawkbit.example.com")
|
|
ap.add_argument("-t", "--tenant", required=True,
|
|
help="Tenant name, e.g. DEFAULT")
|
|
|
|
# ── authentication (mutually-exclusive group would be nice but accepts either) ─
|
|
ap.add_argument("--gateway-token", metavar="TOKEN",
|
|
help="Gateway security token → Authorization: GatewayToken TOKEN")
|
|
ap.add_argument("--target-token", metavar="TOKEN",
|
|
help="Target security token → Authorization: TargetToken TOKEN")
|
|
|
|
# ── optional ──────────────────────────────────────────────────────
|
|
ap.add_argument("--controller-id",
|
|
help="Controller ID (default: auto-detect from device)")
|
|
ap.add_argument("-d", "--download-dir", default="/tmp/hawkbit",
|
|
help="Directory for downloaded artifacts [default: /tmp/hawkbit]")
|
|
ap.add_argument("-i", "--polling-interval", type=int, default=60,
|
|
help="Default polling interval in seconds (overridden by server) [default: 60]")
|
|
ap.add_argument("--no-verify", action="store_true",
|
|
help="Skip SHA256 verification")
|
|
ap.add_argument("--no-resume", action="store_true",
|
|
help="Disable resumable download")
|
|
ap.add_argument("--no-ssl-verify", action="store_true",
|
|
help="Disable TLS certificate verification (INSECURE)")
|
|
ap.add_argument("--once", action="store_true",
|
|
help="Poll exactly once then exit")
|
|
ap.add_argument("-v", "--verbose", action="store_true",
|
|
help="Debug-level logging")
|
|
|
|
args = ap.parse_args(argv)
|
|
|
|
# ── logging ───────────────────────────────────────────────────────
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(asctime)s [%(levelname)-5s] %(name)s: %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
)
|
|
|
|
# ── auth header ───────────────────────────────────────────────────
|
|
auth = None
|
|
if args.gateway_token:
|
|
token = args.gateway_token
|
|
if token.startswith("@"):
|
|
with open(token[1:]) as f:
|
|
token = f.read().strip()
|
|
logger.info("Read gateway token from file: %s", args.gateway_token)
|
|
auth = f"GatewayToken {token}"
|
|
elif args.target_token:
|
|
token = args.target_token
|
|
if token.startswith("@"):
|
|
with open(token[1:]) as f:
|
|
token = f.read().strip()
|
|
logger.info("Read target token from file: %s", args.target_token)
|
|
auth = f"TargetToken {token}"
|
|
|
|
if not auth:
|
|
logger.warning("No authentication token provided — server may reject requests")
|
|
|
|
# ── run ───────────────────────────────────────────────────────────
|
|
client = HawkbitDDIClient(
|
|
base_url=args.base_url,
|
|
tenant=args.tenant,
|
|
controller_id=args.controller_id,
|
|
auth_header=auth,
|
|
download_dir=args.download_dir,
|
|
verify_sha256=not args.no_verify,
|
|
polling_interval=args.polling_interval,
|
|
ssl_verify=not args.no_ssl_verify,
|
|
resume_download=not args.no_resume,
|
|
)
|
|
|
|
try:
|
|
ok = client.run(daemon=not args.once)
|
|
sys.exit(0 if ok else 1)
|
|
except KeyboardInterrupt:
|
|
logger.info("Stopped (Ctrl+C)")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|