veenm/ha-dashboard-studio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dashboard Studio 2.0.0: Admin, Builder and Dashboard screens, roles, Haven

Browse Source 
- Home Assistant administrators are admins: Dashboard, Builder and Admin screens
- Everyone else only sees the dashboards an admin chose, always in kiosk mode;
  the server enforces it (saving, people, devices, history are admin-only, and
  the websocket only allows reading states and controlling devices)
- Admin screen: dashboards, people and what each sees (with preview), devices,
  messages to the screens, the Haven app, activity, about
- Haven: new Android app (haven/) that only shows dashboards; replaces the
  Dashboard Studio Android app and signs in through Home Assistant's page
- Screens pick up dashboard and access changes within a minute
- Version history records who made each change; panel visible to all users

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Mel van Veen
2026-10-26 23:20:55 +01:00
parent 297fb2a7a0
commit 3d958d0411
37 changed files with 1096 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dashboard_studio/server/main.py
+228 -29
View File
@@ -119,7 +119,7 @@ def load_history(dash_id):
return []
def record_history(old, new):
def record_history(old, new, by=""):
"""Add every dashboard that changed in this save to its history."""
before = {d.get("id"): d for d in old.get("dashboards", []) if isinstance(d, dict)}
now = time.time()
@@ -134,10 +134,10 @@ def record_history(old, new):
except OSError:
saved_at = now - HISTORY_MERGE_S - 1
entries.append({"at": saved_at, "data": before[dash_id]})
if entries and now - entries[-1]["at"] < HISTORY_MERGE_S:
entries[-1] = {"at": now, "data": d}
if entries and now - entries[-1]["at"] < HISTORY_MERGE_S and entries[-1].get("by", "") == by:
entries[-1] = {"at": now, "data": d, "by": by}
else:
entries.append({"at": now, "data": d})
entries.append({"at": now, "data": d, "by": by})
HISTORY_DIR.mkdir(parents=True, exist_ok=True)
tmp = history_file(dash_id).with_suffix(".tmp")
tmp.write_text(json.dumps(entries[-HISTORY_MAX:]))
@@ -155,37 +155,152 @@ async def index(request):
USERS_FILE = DATA_DIR / "users.json"
DEVICES_FILE = DATA_DIR / "devices.json"
def who(request):
"""The Home Assistant user, from the headers the Supervisor's ingress proxy adds."""
if MOCK:
return {"id": os.environ.get("MOCK_USER_ID", "mock-user-alex"), "name": "alex", "display": "Alex"}
if MOCK: # locally, a cookie picks one of the mock users, to try the app as someone else
uid = request.cookies.get("mock_user") or os.environ.get("MOCK_USER_ID", "mock-user-alex")
u = MockHA.USERS.get(uid) or MockHA.USERS["mock-user-alex"]
return {"id": uid if uid in MockHA.USERS else "mock-user-alex", "name": u["username"], "display": u["name"]}
h = request.headers
uid = h.get("X-Remote-User-Id")
return {"id": uid, "name": h.get("X-Remote-User-Name", ""), "display": h.get("X-Remote-User-Display-Name", "")} if uid else None
def load_users():
async def ha_call(app, msg):
"""One request to Home Assistant's websocket API with the Supervisor token."""
async with app["http"].ws_connect(HA_WS_URL, max_msg_size=0) as ws:
await ws.receive_json() # auth_required
await ws.send_json({"type": "auth", "access_token": SUPERVISOR_TOKEN})
if (await ws.receive_json()).get("type") != "auth_ok":
raise RuntimeError("Home Assistant rejected the Supervisor token")
await ws.send_json({"id": 1, **msg})
while True:
r = await ws.receive_json()
if r.get("id") == 1:
break
if not r.get("success"):
raise RuntimeError((r.get("error") or {}).get("message", "request failed"))
return r["result"]
USERS_TTL = 60
async def ha_users(app):
"""Home Assistant's people (not its system users), with whether each is an administrator. Cached for a minute."""
cached = app.get("users_cache")
if cached and time.time() - cached[0] < USERS_TTL:
return cached[1]
rows = MockHA.USERS_LIST() if MOCK else await ha_call(app, {"type": "config/auth/list"})
users = {u["id"]: {"id": u["id"], "name": u.get("name") or u.get("username") or "", "username": u.get("username") or "",
"admin": bool(u.get("is_owner")) or "system-admin" in (u.get("group_ids") or []),
"active": u.get("is_active", True)}
for u in rows if not u.get("system_generated")}
app["users_cache"] = (time.time(), users)
return users
async def role_of(request):
""""admin" for Home Assistant administrators, "basic" for everyone else."""
if "role" in request:
return request["role"]
me = who(request)
role = "basic"
if me:
try:
role = "admin" if (await ha_users(request.app)).get(me["id"], {}).get("admin") else "basic"
except Exception: # noqa: BLE001 - never let a failed lookup make someone an admin
LOG.exception("Could not look up Home Assistant's users")
request["role"] = role
return role
def admin_only(handler):
async def wrapped(request):
if await role_of(request) != "admin":
return web.json_response({"error": "Only Home Assistant administrators can do this"}, status=403)
return await handler(request)
return wrapped
def load_json(path, default):
try:
return json.loads(USERS_FILE.read_text())
return json.loads(path.read_text())
except (OSError, json.JSONDecodeError):
return {}
return default
def save_json(path, data):
DATA_DIR.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(data, indent=2))
tmp.replace(path)
def load_users():
return load_json(USERS_FILE, {})
async def whoami(request):
me = who(request)
if me: # remember who has used the app, so dashboards can be assigned to them
if me: # remember who has used the app and when
users = load_users()
if users.get(me["id"], {}).get("display") != me["display"] or me["id"] not in users:
users[me["id"]] = {"name": me["name"], "display": me["display"]}
DATA_DIR.mkdir(parents=True, exist_ok=True)
USERS_FILE.write_text(json.dumps(users, indent=2))
return web.json_response(me or {}, headers={"Cache-Control": "no-store"})
prev = users.get(me["id"], {})
if prev.get("display") != me["display"] or time.time() - prev.get("seen", 0) > 300:
users[me["id"]] = {"name": me["name"], "display": me["display"], "seen": time.time()}
save_json(USERS_FILE, users)
return web.json_response({**(me or {}), "admin": await role_of(request) == "admin"}, headers={"Cache-Control": "no-store"})
async def users(request):
return web.json_response([{"id": k, **v} for k, v in load_users().items()], headers={"Cache-Control": "no-store"})
@admin_only
async def people(request):
"""Everyone with a Home Assistant login, for choosing what they see."""
seen = load_users()
try:
users = await ha_users(request.app)
except Exception as err: # noqa: BLE001
return web.json_response({"error": f"Couldn't read Home Assistant's users: {err}"}, status=502)
out = [{**u, "seen": seen.get(u["id"], {}).get("seen")} for u in users.values() if u["active"]]
return web.json_response(sorted(out, key=lambda u: (not u["admin"], u["name"].lower())), headers={"Cache-Control": "no-store"})
SAFE_DEVICE = re.compile(r"^[\w-]{4,64}$")
async def device_seen(request):
"""Every device reports in when it opens, so an admin can give it a dashboard."""
try:
body = await request.json()
except json.JSONDecodeError:
return web.Response(status=400)
dev = str(body.get("device", ""))
if not SAFE_DEVICE.match(dev):
return web.Response(status=400)
me = who(request) or {}
devices = load_json(DEVICES_FILE, {})
devices[dev] = {"user": me.get("id"), "display": me.get("display", ""), "seen": time.time(),
"app": "haven" if body.get("app") == "haven" else "browser", "kind": str(body.get("kind", ""))[:20]}
if len(devices) > 200: # forget the longest-unseen ones
devices = dict(sorted(devices.items(), key=lambda kv: kv[1].get("seen", 0))[-200:])
save_json(DEVICES_FILE, devices)
return web.json_response({"ok": True})
@admin_only
async def get_devices(request):
return web.json_response(load_json(DEVICES_FILE, {}), headers={"Cache-Control": "no-store"})
@admin_only
async def forget_device(request):
dev = request.match_info["device"]
devices = load_json(DEVICES_FILE, {})
devices.pop(dev, None)
save_json(DEVICES_FILE, devices)
return web.json_response({"ok": True})
async def changelog(request):
@@ -234,17 +349,46 @@ async def service_worker(request):
return web.Response(text=js, content_type="application/javascript", headers={"Cache-Control": "no-store"})
async def android_app(request):
"""The Android app (built from android/ in the repository), to download on a phone."""
return web.FileResponse(WEB_DIR / "dashboard-studio.apk", headers={
async def haven_app(request):
"""Haven, the Android app for dashboards (built from haven/ in the repository), to download on a phone."""
return web.FileResponse(WEB_DIR / "haven.apk", headers={
"Content-Type": "application/vnd.android.package-archive",
"Content-Disposition": 'attachment; filename="Dashboard Studio.apk"', "Cache-Control": "no-store"})
"Content-Disposition": 'attachment; filename="Haven.apk"', "Cache-Control": "no-store"})
def access_of(store, uid):
"""What a person may see: {"all": true} or {"dashboards": [...]}, plus the one to open first ("home")."""
acc = store.get("access") or {}
mine = (acc.get("users") or {}).get(uid) if uid else None
return mine if isinstance(mine, dict) else acc.get("everyone") or {"all": True}
def visible_ids(store, uid, device):
ids = [d.get("id") for d in store.get("dashboards", []) if isinstance(d, dict)]
entry = access_of(store, uid)
seen = ids if entry.get("all") else [i for i in ids if i in (entry.get("dashboards") or [])]
dev = ((store.get("assign") or {}).get("devices") or {}).get(device or "") or {}
if dev.get("dashboard") in ids and dev["dashboard"] not in seen: # an admin put it on this screen
seen.append(dev["dashboard"])
return seen, dev
async def get_dashboards(request):
return web.json_response(load_store())
store = load_store()
if await role_of(request) == "admin":
return web.json_response({**store, "role": "admin"}, headers={"Cache-Control": "no-store"})
me = who(request) or {}
ids, dev = visible_ids(store, me.get("id"), request.query.get("device"))
return web.json_response({
"version": store.get("version", 1),
"dashboards": [d for d in store.get("dashboards", []) if d.get("id") in ids],
"assign": {"devices": {request.query["device"]: dev} if dev else {}, "users": {}},
"access": {"me": access_of(store, me.get("id"))},
"role": "basic",
}, headers={"Cache-Control": "no-store"})
@admin_only
async def put_dashboards(request):
try:
data = await request.json()
@@ -252,8 +396,9 @@ async def put_dashboards(request):
return web.json_response({"error": "invalid json"}, status=400)
if not isinstance(data, dict) or not isinstance(data.get("dashboards"), list):
return web.json_response({"error": "expected {dashboards: [...]}"}, status=400)
data.pop("role", None)
try:
record_history(load_store(), data)
record_history(load_store(), data, (who(request) or {}).get("display", ""))
except OSError:
LOG.exception("Could not write the version history")
save_store(data)
@@ -264,17 +409,19 @@ def widget_count(d):
return sum(len(t.get("widgets", [])) for t in d.get("tabs", []) if isinstance(t, dict))
@admin_only
async def get_history(request):
dash_id = request.match_info["dash_id"]
if not SAFE_ID.match(dash_id):
return web.Response(status=400)
entries = load_history(dash_id)
return web.json_response([
{"at": e["at"], "name": e["data"].get("name", ""), "tabs": len(e["data"].get("tabs", [])), "widgets": widget_count(e["data"])}
{"at": e["at"], "name": e["data"].get("name", ""), "tabs": len(e["data"].get("tabs", [])), "widgets": widget_count(e["data"]), "by": e.get("by", "")}
for e in reversed(entries)
], headers={"Cache-Control": "no-store"})
@admin_only
async def get_history_version(request):
dash_id = request.match_info["dash_id"]
if not SAFE_ID.match(dash_id):
@@ -289,6 +436,19 @@ async def get_history_version(request):
return web.Response(status=404)
@admin_only
async def activity(request):
"""Who changed which dashboard, newest first, from the version history."""
names = {d.get("id"): d.get("name", "") for d in load_store().get("dashboards", []) if isinstance(d, dict)}
rows = []
for f in HISTORY_DIR.glob("*.json") if HISTORY_DIR.is_dir() else []:
for e in load_json(f, [])[1:]: # the first entry is how it was before history started
rows.append({"at": e["at"], "dashboard": f.stem, "name": names.get(f.stem) or e["data"].get("name", ""),
"by": e.get("by", ""), "deleted": f.stem not in names})
rows.sort(key=lambda r: r["at"], reverse=True)
return web.json_response(rows[:60], headers={"Cache-Control": "no-store"})
async def camera(request):
entity_id = request.match_info["entity_id"]
if not entity_id.startswith("camera.") or "/" in entity_id:
@@ -309,12 +469,39 @@ async def camera(request):
)
# What people who aren't administrators may ask Home Assistant through this app: reading states and
# controlling devices, as they could in Home Assistant itself, but nothing that changes its setup.
BASIC_WS = {"get_states", "get_config", "get_services", "unsubscribe_events", "persistent_notification/subscribe",
"config/area_registry/list", "config/entity_registry/list", "config/device_registry/list", "todo/item/list",
"history/history_during_period", "recorder/statistics_during_period", "energy/solar_forecast", "ping"}
BASIC_EVENTS = {"state_changed", "dashboard_studio_announce"}
BLOCKED_DOMAINS = {"hassio", "recorder", "logger", "system_log", "backup", "lovelace", "frontend", "cloud", "auth", "zone", "person"}
def allowed_for_basic(req):
t = req.get("type")
if t == "subscribe_events":
return req.get("event_type") in BASIC_EVENTS
if t == "call_service":
domain, service = req.get("domain"), req.get("service")
if domain in BLOCKED_DOMAINS:
return False
return service in {"toggle", "turn_on", "turn_off", "update_entity"} if domain == "homeassistant" else True
return t in BASIC_WS
def denied(req):
return {"id": req.get("id"), "type": "result", "success": False,
"error": {"code": "unauthorized", "message": "Only Home Assistant administrators can do this"}}
async def websocket(request):
client = web.WebSocketResponse(heartbeat=30, max_msg_size=0)
await client.prepare(request)
allow = None if await role_of(request) == "admin" else allowed_for_basic
if MOCK:
await request.app["mock"].serve(client)
await request.app["mock"].serve(client, allow)
return client
if not SUPERVISOR_TOKEN:
await client.send_json({"type": "error", "message": "Dashboard Studio has no access to Home Assistant (SUPERVISOR_TOKEN is missing). Restart the app; if this stays, reinstall it."})
@@ -340,14 +527,22 @@ async def websocket(request):
return client
await client.send_json({"type": "ready", "ha_version": msg.get("ha_version")})
async def pump(src, dst):
async def pump(src, dst, check=None):
async for m in src:
if m.type == aiohttp.WSMsgType.TEXT:
if check:
try:
req = json.loads(m.data)
except json.JSONDecodeError:
continue
if not isinstance(req, dict) or not check(req):
await client.send_json(denied(req if isinstance(req, dict) else {}))
continue
await dst.send_str(m.data)
elif m.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE):
break
tasks = [asyncio.create_task(pump(client, upstream)), asyncio.create_task(pump(upstream, client))]
tasks = [asyncio.create_task(pump(client, upstream, allow)), asyncio.create_task(pump(upstream, client))]
_, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
t.cancel()
@@ -379,10 +574,14 @@ def make_app():
app.router.add_get("/api/version", version)
app.router.add_get("/manifest.webmanifest", manifest)
app.router.add_get("/sw.js", service_worker)
app.router.add_get("/android.apk", android_app)
app.router.add_get("/haven.apk", haven_app)
app.router.add_get("/api/changelog", changelog)
app.router.add_get("/api/whoami", whoami)
app.router.add_get("/api/users", users)
app.router.add_get("/api/people", people)
app.router.add_post("/api/devices/seen", device_seen)
app.router.add_get("/api/devices", get_devices)
app.router.add_delete("/api/devices/{device}", forget_device)
app.router.add_get("/api/activity", activity)
app.router.add_get("/api/dashboards", get_dashboards)
app.router.add_put("/api/dashboards", put_dashboards)
app.router.add_get("/api/history/{dash_id}", get_history)