homecore-apps/homecore_controller/app/main.py

from pathlib import Path
import os
import json
import base64
import hashlib
import time
import threading
import secrets
import html

import httpx
from fastapi import FastAPI, Form, HTTPException
from fastapi.responses import HTMLResponse
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
import uvicorn

VERSION = "0.1.1"

DATA = Path("/config")
DATA.mkdir(parents=True, exist_ok=True)

STATE = DATA / "state.json"
KEY = DATA / "controller_ed25519.pem"
OPTIONS = Path("/data/options.json")

PENDING = {}

def options():
    try:
        return json.loads(OPTIONS.read_text())
    except Exception:
        return {"cloud_api": "https://api.homecore.home-core.co.uk"}

def ensure_key():
    if KEY.exists():
        return serialization.load_pem_private_key(KEY.read_bytes(), password=None)

    k = Ed25519PrivateKey.generate()
    KEY.write_bytes(
        k.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        )
    )
    return k

KEYOBJ = ensure_key()

def pub_pem():
    return KEYOBJ.public_key().public_bytes(
        serialization.Encoding.PEM,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    ).decode()

def fingerprint():
    raw = KEYOBJ.public_key().public_bytes(
        serialization.Encoding.Raw,
        serialization.PublicFormat.Raw,
    )
    return hashlib.sha256(raw).hexdigest()

def load_state():
    try:
        return json.loads(STATE.read_text())
    except Exception:
        return {}

def save_state(s):
    STATE.write_text(json.dumps(s, indent=2))

def supervisor_get(path):
    token = os.getenv("SUPERVISOR_TOKEN")
    if not token:
        return None

    try:
        r = httpx.get(
            "http://supervisor" + path,
            headers={"Authorization": "Bearer " + token},
            timeout=5,
        )
        r.raise_for_status()
        return r.json()
    except Exception:
        return None

def ha_version():
    info = supervisor_get("/core/info") or {}
    if isinstance(info, dict):
        return ((info.get("data") or {}).get("version") or "")
    return ""

def cloud_detail(response):
    try:
        data = response.json()
        if isinstance(data, dict):
            return data.get("detail") or str(data)
    except Exception:
        pass
    return f"HomeCore Cloud returned HTTP {response.status_code}"

def controller_session():
    s = load_state()
    if not s.get("controller_id"):
        return None

    api = options()["cloud_api"].rstrip("/")

    with httpx.Client(timeout=15) as c:
        ch = c.get(f"{api}/api/v1/controllers/{s['controller_id']}/challenge")
        ch.raise_for_status()

        challenge = ch.json()["challenge"]

        sig = base64.b64encode(
            KEYOBJ.sign(challenge.encode())
        ).decode()

        r = c.post(
            f"{api}/api/v1/controllers/{s['controller_id']}/session",
            json={
                "challenge": challenge,
                "signature_b64": sig,
            },
        )
        r.raise_for_status()

        return r.json()["controller_token"]

def heartbeat_loop():
    while True:
        try:
            s = load_state()

            if s.get("controller_id"):
                token = controller_session()
                api = options()["cloud_api"].rstrip("/")

                httpx.post(
                    f"{api}/api/v1/controllers/{s['controller_id']}/heartbeat",
                    headers={"Authorization": "Bearer " + token},
                    json={
                        "software_version": VERSION,
                        "ha_version": ha_version(),
                    },
                    timeout=10,
                )

        except Exception:
            pass

        time.sleep(60)

app = FastAPI(
    title="HomeCore Controller",
    version=VERSION,
)

STYLE = '''
<style>
body{
    font-family:Inter,Arial,sans-serif;
    background:#0d1520;
    color:#edf4ff;
    margin:0
}
.wrap{
    max-width:720px;
    margin:60px auto;
    padding:28px
}
.card{
    background:#162231;
    border:1px solid #26384b;
    border-radius:18px;
    padding:28px;
    box-shadow:0 18px 50px #0005
}
h1{
    font-size:36px;
    margin:0 0 4px
}
.core{
    color:#78a9ff
}
p{
    color:#aebed0;
    line-height:1.5
}
label{
    display:block;
    font-size:13px;
    color:#aebed0;
    margin-bottom:6px
}
input,select{
    width:100%;
    box-sizing:border-box;
    padding:13px;
    margin:0 0 16px;
    border-radius:10px;
    border:1px solid #36506a;
    background:#0e1926;
    color:white
}
button{
    background:#377dff;
    color:white;
    border:0;
    border-radius:10px;
    padding:13px 18px;
    font-weight:700;
    cursor:pointer
}
.ok{
    color:#55d68b
}
.error{
    color:#ff7b7b
}
.small{
    color:#8195aa;
    font-size:13px
}
</style>

<script>
function homecoreAction(form, endpoint) {
    const parts = window.location.pathname.split('/').filter(Boolean);

    let base = '';

    if (parts.length && parts[0].includes('homecore_controller')) {
        base = '/' + parts[0];
    }

    form.action = base + '/' + endpoint;
    return true;
}
</script>
'''

def page(body):
    return HTMLResponse(
        STYLE +
        '<div class="wrap"><div class="card">' +
        body +
        '</div></div>'
    )

def error_page(message):
    return page(
        '<h1>Home<span class="core">Core</span></h1>'
        '<h2 class="error">Unable to continue</h2>'
        f'<p>{html.escape(message)}</p>'
        '<p class="small">No controller binding has been changed.</p>'
    )

@app.get("/", response_class=HTMLResponse)
def setup_page():

    s = load_state()

    if s.get("controller_id"):
        return page(
            '<h1>Home<span class="core">Core</span></h1>'
            '<h2 class="ok">Controller connected</h2>'
            f'<p>Controller: {html.escape(s["controller_id"])}</p>'
            f'<p>Property: {html.escape(s.get("property_id",""))}</p>'
            f'<p>HomeCore Controller version: {VERSION}</p>'
            '<p>This Home Assistant Green is permanently bound to this HomeCore property.</p>'
        )

    return page(
        '<h1>Home<span class="core">Core</span></h1>'
        '<h2>Set up this controller</h2>'
        '<p>Sign in with the customer\'s HomeCore Owner account. '
        'You will then choose one of the homes already created in HomeCore.</p>'

        '<form method="post" '
        'onsubmit="return homecoreAction(this,\\'select-property\\')">'

        '<label>Owner email</label>'
        '<input name="email" type="email" autocomplete="email" required>'

        '<label>Password</label>'
        '<input name="password" type="password" autocomplete="current-password" required>'

        '<button>Continue</button>'
        '</form>'

        '<p class="small">'
        'Homes must be created in the HomeCore customer portal before a controller can be connected.'
        '</p>'
    )

@app.post("/select-property", response_class=HTMLResponse)
def select_property(
    email: str = Form(...),
    password: str = Form(...),
):

    api = options()["cloud_api"].rstrip("/")

    try:
        with httpx.Client(timeout=20) as c:

            r = c.post(
                api + "/api/v1/auth/login",
                json={
                    "email": email,
                    "password": password,
                },
            )

            if r.status_code >= 400:
                return error_page("Incorrect HomeCore email address or password.")

            token = r.json()["access_token"]

            headers = {
                "Authorization": "Bearer " + token
            }

            r = c.get(
                api + "/api/v1/properties",
                headers=headers,
            )

            if r.status_code >= 400:
                return error_page(cloud_detail(r))

            owned = [
                p for p in r.json()
                if p.get("owner") is True
            ]

    except Exception as e:
        return error_page(
            "The controller could not contact HomeCore Cloud. " + str(e)
        )

    if not owned:
        return error_page(
            "This Owner account does not have a HomeCore home yet. "
            "Create the home at homecore.home-core.co.uk, then return here."
        )

    nonce = secrets.token_urlsafe(32)

    PENDING[nonce] = {
        "token": token,
        "expires": time.time() + 300,
        "property_ids": [p["id"] for p in owned],
    }

    choices = "".join(
        f'<option value="{html.escape(p["id"])}">'
        f'{html.escape(p["name"])}</option>'
        for p in owned
    )

    return page(
        '<h1>Home<span class="core">Core</span></h1>'
        '<h2>Select the customer home</h2>'
        '<p>Only homes owned by this HomeCore account can be connected.</p>'

        '<form method="post" '
        'onsubmit="return homecoreAction(this,\\'bind\\')">'

        f'<input type="hidden" name="nonce" value="{html.escape(nonce)}">'

        '<label>Home</label>'
        f'<select name="property_id" required>{choices}</select>'

        '<button>Connect HomeCore</button>'
        '</form>'

        '<p class="small">'
        'After connection, this controller cannot be claimed by another HomeCore account.'
        '</p>'
    )

@app.post("/bind", response_class=HTMLResponse)
def bind(
    nonce: str = Form(...),
    property_id: str = Form(...),
):

    if load_state().get("controller_id"):
        raise HTTPException(
            409,
            "Controller already bound",
        )

    pending = PENDING.get(nonce)

    if not pending:
        return error_page(
            "The setup session has expired. Sign in again."
        )

    if pending["expires"] < time.time():
        PENDING.pop(nonce, None)
        return error_page(
            "The setup session has expired. Sign in again."
        )

    if property_id not in pending["property_ids"]:
        return error_page(
            "This HomeCore account does not own the selected property."
        )

    token = pending["token"]

    api = options()["cloud_api"].rstrip("/")

    headers = {
        "Authorization": "Bearer " + token
    }

    payload = {
        "property_id": property_id,
        "public_key": pub_pem(),
        "fingerprint": fingerprint(),
        "software_version": VERSION,
        "ha_version": ha_version(),
    }

    try:
        with httpx.Client(timeout=20) as c:

            r = c.post(
                api + "/api/v1/controllers/bind",
                headers=headers,
                json=payload,
            )

            if r.status_code >= 400:
                return error_page(cloud_detail(r))

            out = r.json()

    except Exception as e:
        return error_page(
            "The controller could not contact HomeCore Cloud. " + str(e)
        )

    PENDING.pop(nonce, None)

    save_state(out)

    return page(
        '<h1>Home<span class="core">Core</span></h1>'
        '<h2 class="ok">Controller connected</h2>'
        '<p>This Home Assistant Green is now permanently bound to the selected HomeCore home.</p>'
        '<p>The customer can now sign into HomeCore using their normal HomeCore account.</p>'
        f'<p class="small">HomeCore Controller {VERSION}</p>'
    )

@app.get("/health")
def health():
    return {
        "status": "ok",
        "bound": bool(load_state().get("controller_id")),
        "fingerprint": fingerprint(),
        "version": VERSION,
    }

threading.Thread(
    target=heartbeat_loop,
    daemon=True,
).start()

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8099,
    )