83 lines
6.1 KiB
Python
83 lines
6.1 KiB
Python
from pathlib import Path
|
|
import os,json,base64,hashlib,time,threading
|
|
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
|
|
|
|
DATA=Path('/config');DATA.mkdir(parents=True,exist_ok=True)
|
|
STATE=DATA/'state.json';KEY=DATA/'controller_ed25519.pem';OPTIONS=Path('/data/options.json')
|
|
|
|
def options():
|
|
try:return json.loads(OPTIONS.read_text())
|
|
except Exception:return {'cloud_api':'https://api.homecore.example.com'}
|
|
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():return hashlib.sha256(KEYOBJ.public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.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 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('/')
|
|
info=supervisor_get('/core/info') or {}; ha_ver=((info.get('data') or {}).get('version') or '') if isinstance(info,dict) else ''
|
|
httpx.post(f"{api}/api/v1/controllers/{s['controller_id']}/heartbeat",headers={'Authorization':'Bearer '+token},json={'software_version':'0.1.0','ha_version':ha_ver},timeout=10)
|
|
except Exception:pass
|
|
time.sleep(60)
|
|
|
|
app=FastAPI(title='HomeCore Controller',version='0.1.0')
|
|
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{font-size:13px;color:#aebed0}input{width:100%;box-sizing:border-box;padding:13px;margin:6px 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}</style>'''
|
|
|
|
@app.get('/',response_class=HTMLResponse)
|
|
def setup_page():
|
|
s=load_state()
|
|
if s.get('controller_id'):
|
|
return STYLE+f'<div class="wrap"><div class="card"><h1>Home<span class="core">Core</span></h1><h2 class="ok">Controller connected</h2><p>Controller: {s["controller_id"]}</p><p>Property: {s.get("property_id")}</p><p>This Green is permanently bound to the HomeCore property.</p></div></div>'
|
|
return STYLE+'''<div class="wrap"><div class="card"><h1>Home<span class="core">Core</span></h1><h2>Set up this controller</h2><p>Sign in with the customer's HomeCore Owner account. HomeCore will create or select the property and permanently bind this Home Assistant Green.</p><form method="post" action="./bind"><label>Owner email</label><input name="email" type="email" required><label>Password</label><input name="password" type="password" required><label>Property name</label><input name="property_name" required><button>Connect HomeCore</button></form></div></div>'''
|
|
|
|
@app.post('/bind',response_class=HTMLResponse)
|
|
def bind(email:str=Form(...),password:str=Form(...),property_name:str=Form(...)):
|
|
if load_state().get('controller_id'):raise HTTPException(409,'Controller already bound')
|
|
api=options()['cloud_api'].rstrip('/')
|
|
with httpx.Client(timeout=20) as c:
|
|
r=c.post(api+'/api/v1/auth/login',json={'email':email,'password':password});r.raise_for_status();token=r.json()['access_token'];headers={'Authorization':'Bearer '+token}
|
|
r=c.get(api+'/api/v1/properties',headers=headers);r.raise_for_status();found=next((p for p in r.json() if p['name'].lower()==property_name.lower() and p.get('owner')),None)
|
|
if not found:
|
|
r=c.post(api+'/api/v1/properties',headers=headers,json={'name':property_name,'timezone':'Europe/London'});r.raise_for_status();found=r.json()
|
|
info=supervisor_get('/core/info') or {};ha_ver=((info.get('data') or {}).get('version') or '') if isinstance(info,dict) else ''
|
|
payload={'property_id':found['id'],'public_key':pub_pem(),'fingerprint':fingerprint(),'software_version':'0.1.0','ha_version':ha_ver}
|
|
r=c.post(api+'/api/v1/controllers/bind',headers=headers,json=payload);r.raise_for_status();out=r.json()
|
|
save_state(out)
|
|
return HTMLResponse(STYLE+'<div class="wrap"><div class="card"><h1>Home<span class="core">Core</span></h1><h2 class="ok">Connected</h2><p>This controller is now bound to the customer property. Sign into the HomeCore app with the same Owner account.</p></div></div>')
|
|
|
|
@app.get('/health')
|
|
def health():return {'status':'ok','bound':bool(load_state().get('controller_id')),'fingerprint':fingerprint()}
|
|
|
|
threading.Thread(target=heartbeat_loop,daemon=True).start()
|
|
if __name__=='__main__':uvicorn.run(app,host='0.0.0.0',port=8099)
|