375 lines
10 KiB
Python
375 lines
10 KiB
Python
from pathlib import Path
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import secrets
|
|
import time
|
|
|
|
import httpx
|
|
|
|
|
|
VERSION = "1.0.0"
|
|
|
|
SHARE = Path("/share/homecore")
|
|
CONFIG = Path("/config")
|
|
SECRET_LOCAL = CONFIG / "updater-secret"
|
|
SECRET_SHARED = SHARE / "updater-secret"
|
|
REQUEST_FILE = SHARE / "update-request.json"
|
|
RESULT_FILE = SHARE / "update-result.json"
|
|
STATE_FILE = CONFIG / "updater-state.json"
|
|
|
|
SUPERVISOR = "http://supervisor"
|
|
|
|
SHARE.mkdir(parents=True, exist_ok=True)
|
|
CONFIG.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def atomic_write(path: Path, data: bytes):
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_bytes(data)
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def canonical(data):
|
|
return json.dumps(
|
|
data,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode()
|
|
|
|
|
|
def ensure_secret():
|
|
if SECRET_LOCAL.exists():
|
|
secret = SECRET_LOCAL.read_text().strip()
|
|
else:
|
|
secret = secrets.token_hex(32)
|
|
SECRET_LOCAL.write_text(secret + "\n")
|
|
os.chmod(SECRET_LOCAL, 0o600)
|
|
|
|
atomic_write(SECRET_SHARED, (secret + "\n").encode())
|
|
try:
|
|
os.chmod(SECRET_SHARED, 0o600)
|
|
except Exception:
|
|
pass
|
|
|
|
return secret.encode()
|
|
|
|
|
|
SECRET = ensure_secret()
|
|
|
|
|
|
def load_json(path):
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def save_json(path, data):
|
|
atomic_write(
|
|
path,
|
|
(json.dumps(data, indent=2, sort_keys=True) + "\n").encode(),
|
|
)
|
|
|
|
|
|
def supervisor_headers():
|
|
token = os.getenv("SUPERVISOR_TOKEN")
|
|
if not token:
|
|
raise RuntimeError("SUPERVISOR_TOKEN is not available")
|
|
return {
|
|
"Authorization": "Bearer " + token,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def supervisor_request(method, path, payload=None, timeout=120):
|
|
r = httpx.request(
|
|
method,
|
|
SUPERVISOR + path,
|
|
headers=supervisor_headers(),
|
|
json=payload,
|
|
timeout=timeout,
|
|
)
|
|
r.raise_for_status()
|
|
|
|
try:
|
|
body = r.json()
|
|
except Exception:
|
|
return {}
|
|
|
|
if isinstance(body, dict):
|
|
if body.get("result") == "error":
|
|
raise RuntimeError(body.get("message") or "Supervisor error")
|
|
if "data" in body:
|
|
return body["data"]
|
|
|
|
return body
|
|
|
|
|
|
def find_homecore_controller():
|
|
data = supervisor_request("GET", "/store/addons", timeout=30)
|
|
apps = data.get("addons", []) if isinstance(data, dict) else (data or [])
|
|
|
|
candidates = []
|
|
for app in apps:
|
|
slug = str(app.get("slug", ""))
|
|
name = str(app.get("name", ""))
|
|
if (
|
|
name == "HomeCore Controller"
|
|
or slug == "homecore_controller"
|
|
or slug.endswith("_homecore_controller")
|
|
):
|
|
candidates.append(app)
|
|
|
|
if not candidates:
|
|
raise RuntimeError(
|
|
"HomeCore Controller was not found in the Home Assistant App store"
|
|
)
|
|
|
|
installed = [x for x in candidates if x.get("installed")]
|
|
return (installed or candidates)[0]
|
|
|
|
|
|
def sign_result(body):
|
|
sig = hmac.new(SECRET, canonical(body), hashlib.sha256).hexdigest()
|
|
return {**body, "signature": sig}
|
|
|
|
|
|
def write_result(status, request, detail=""):
|
|
body = {
|
|
"schema": 1,
|
|
"request_id": str(request.get("request_id", "")),
|
|
"target_version": str(request.get("target_version", "")),
|
|
"status": status,
|
|
"detail": detail,
|
|
"updated_at": int(time.time()),
|
|
"updater_version": VERSION,
|
|
}
|
|
save_json(RESULT_FILE, sign_result(body))
|
|
print(
|
|
"HOMECORE UPDATER:"
|
|
f" {status}"
|
|
f" request={body['request_id'] or '-'}"
|
|
f" target={body['target_version'] or '-'}"
|
|
f" {detail}",
|
|
flush=True,
|
|
)
|
|
|
|
|
|
def verify_request(request):
|
|
if not isinstance(request, dict):
|
|
raise RuntimeError("Update request is not valid JSON")
|
|
|
|
signature = str(request.get("signature", "")).strip()
|
|
if not signature:
|
|
raise RuntimeError("Update request signature is missing")
|
|
|
|
body = {k: v for k, v in request.items() if k != "signature"}
|
|
expected = hmac.new(
|
|
SECRET,
|
|
canonical(body),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
if not hmac.compare_digest(signature, expected):
|
|
raise RuntimeError("Update request signature is invalid")
|
|
|
|
if int(body.get("schema", 0)) != 1:
|
|
raise RuntimeError("Unsupported update request schema")
|
|
|
|
request_id = str(body.get("request_id", "")).strip()
|
|
target = str(body.get("target_version", "")).strip()
|
|
controller_id = str(body.get("controller_id", "")).strip()
|
|
|
|
if not request_id or not target or not controller_id:
|
|
raise RuntimeError("Update request is incomplete")
|
|
|
|
age = int(time.time()) - int(body.get("issued_at", 0) or 0)
|
|
if age < -120 or age > 900:
|
|
raise RuntimeError("Update request has expired")
|
|
|
|
return body
|
|
|
|
|
|
def installed_version(app):
|
|
value = app.get("version")
|
|
return str(value or "")
|
|
|
|
|
|
def repository_version(app):
|
|
return str(
|
|
app.get("version_latest")
|
|
or app.get("version_available")
|
|
or app.get("version")
|
|
or ""
|
|
)
|
|
|
|
|
|
def perform_update(request):
|
|
request_id = request["request_id"]
|
|
target = request["target_version"]
|
|
|
|
write_result("preparing", request, "Manual update accepted by HomeCore Updater")
|
|
|
|
supervisor_request("POST", "/store/reload", timeout=60)
|
|
time.sleep(3)
|
|
|
|
app = find_homecore_controller()
|
|
slug = str(app.get("slug", "")).strip()
|
|
installed = installed_version(app)
|
|
latest = repository_version(app)
|
|
|
|
print(
|
|
"HOMECORE UPDATER: repository check"
|
|
f" installed={installed or '-'}"
|
|
f" latest={latest or '-'}"
|
|
f" target={target}",
|
|
flush=True,
|
|
)
|
|
|
|
if not slug:
|
|
raise RuntimeError("HomeCore Controller slug could not be determined")
|
|
|
|
if latest != target:
|
|
raise RuntimeError(
|
|
"Home Assistant repository currently offers "
|
|
+ (latest or "no version")
|
|
+ "; target is "
|
|
+ target
|
|
)
|
|
|
|
if installed == target:
|
|
write_result(
|
|
"completed",
|
|
request,
|
|
"HomeCore Controller already has the requested version",
|
|
)
|
|
return
|
|
|
|
supervisor_request("GET", f"/store/addons/{slug}/availability", timeout=60)
|
|
|
|
write_result(
|
|
"installing",
|
|
request,
|
|
"Backup requested; Supervisor is updating HomeCore Controller",
|
|
)
|
|
|
|
# This app is deliberately separate from HomeCore Controller, so this is
|
|
# not a self-update. Supervisor is allowed to update the other app.
|
|
supervisor_request(
|
|
"POST",
|
|
f"/store/addons/{slug}/update",
|
|
{
|
|
"backup": True,
|
|
"background": True,
|
|
},
|
|
timeout=120,
|
|
)
|
|
|
|
deadline = time.time() + 600
|
|
last_seen = ""
|
|
|
|
while time.time() < deadline:
|
|
time.sleep(5)
|
|
|
|
supervisor_request("POST", "/store/reload", timeout=60)
|
|
app = find_homecore_controller()
|
|
current = installed_version(app)
|
|
|
|
if current != last_seen:
|
|
print(
|
|
"HOMECORE UPDATER: waiting for controller"
|
|
f" current={current or '-'} target={target}",
|
|
flush=True,
|
|
)
|
|
last_seen = current
|
|
|
|
if current == target:
|
|
write_result(
|
|
"completed",
|
|
request,
|
|
"HomeCore Controller update completed successfully",
|
|
)
|
|
return
|
|
|
|
raise RuntimeError(
|
|
"Timed out waiting for HomeCore Controller to reach " + target
|
|
)
|
|
|
|
|
|
def main():
|
|
print(
|
|
"HOMECORE UPDATER:"
|
|
f" version={VERSION}"
|
|
" supervisor_token="
|
|
+ ("available" if os.getenv("SUPERVISOR_TOKEN") else "missing"),
|
|
flush=True,
|
|
)
|
|
print(
|
|
"HOMECORE UPDATER: waiting for signed requests in "
|
|
+ str(REQUEST_FILE),
|
|
flush=True,
|
|
)
|
|
|
|
last_seen_request = ""
|
|
|
|
while True:
|
|
try:
|
|
if REQUEST_FILE.exists():
|
|
raw = load_json(REQUEST_FILE)
|
|
req_id = str(raw.get("request_id", ""))
|
|
|
|
if req_id and req_id != last_seen_request:
|
|
request = verify_request(raw)
|
|
state = load_json(STATE_FILE)
|
|
|
|
if str(state.get("request_id", "")) == req_id:
|
|
# Never execute an identical request twice.
|
|
last_seen_request = req_id
|
|
else:
|
|
save_json(
|
|
STATE_FILE,
|
|
{
|
|
"request_id": req_id,
|
|
"target_version": request["target_version"],
|
|
"status": "running",
|
|
"started_at": int(time.time()),
|
|
},
|
|
)
|
|
|
|
try:
|
|
perform_update(request)
|
|
save_json(
|
|
STATE_FILE,
|
|
{
|
|
"request_id": req_id,
|
|
"target_version": request["target_version"],
|
|
"status": "completed",
|
|
"completed_at": int(time.time()),
|
|
},
|
|
)
|
|
except Exception as exc:
|
|
write_result("failed", request, str(exc))
|
|
save_json(
|
|
STATE_FILE,
|
|
{
|
|
"request_id": req_id,
|
|
"target_version": request["target_version"],
|
|
"status": "failed",
|
|
"detail": str(exc),
|
|
"failed_at": int(time.time()),
|
|
},
|
|
)
|
|
|
|
last_seen_request = req_id
|
|
|
|
except Exception as exc:
|
|
print("HOMECORE UPDATER: watcher error:", exc, flush=True)
|
|
|
|
time.sleep(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|