The_miro
/
SmartestHome
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
SmartestHome/identity/server.py

1842 lines
85 KiB
Python
Executable File
Raw Blame History

"""identity — the household's person <-> BLE-identifier registry, from
docs/project-plan.md Phase 6.
Solves two problems, both from the same design: a person owning **multiple phones**
(private + work) and defending registration against **MAC spoofing/randomization**.
THE MODEL: a person has zero or more *identifiers*. An identifier is a Home Assistant
entity_id that resolves presence for one physical device. Multi-phone support falls
out of this for free — the same person just accumulates a second identifier the
second time they register with their other phone in hand.
ANTI-SPOOFING, the actual point of this module: a raw Bluetooth MAC address —
especially a randomized one, which is the default on iOS/Android — is never accepted
as an identifier by itself. Only entity_ids matching `TRUSTED_ENTITY_PREFIXES` are
eligible candidates at registration time, and that allowlist is meant to contain only
HA's own IRK-resolved Private BLE Device entities (survive MAC rotation because HA
resolved them cryptographically, not by matching a MAC string) and manually
provisioned fixed-MAC tag entities. An attacker spoofing an arbitrary advertised MAC
never produces a trusted candidate; spoofing a specific person's *resolved* identity
would require their device's actual IRK secret, a materially harder bar. This is a
defense against passive/opportunistic spoofing, not a claim of cryptographic
non-repudiation — see README.md's threat-model section for the honest boundary.
NEVER AUTO-COMMIT ON AMBIGUITY: if registration finds zero, more than one, or an
already-claimed candidate, nothing is written — the caller gets back a reason and (for
the ambiguous case) the candidate list, and a human has to disambiguate via the
touchscreen (POST /register again with an explicit entity_id). The one case that DOES
commit in a single call is the clean one (exactly one trusted, unclaimed candidate),
because the spoken "register me as <name>" command *is* the human confirmation —
requiring a second round-trip for the unambiguous case would just be friction for no
safety benefit. This mirrors, rather than weakens, this project's existing "an
identity merge must never auto-commit silently" rule (docs/project-plan.md's Identity
store row): ambiguity is exactly the case that still requires a human.
NICKNAMES ARE FOR HUMANS TO SAY, NEVER FOR THE MACHINE TO SAY BACK: a person may have
a `nickname` that household members use to refer to them, and /resolve accepts it as
an *input* alias so "is Bibi home?" works. But every payload this service emits also
carries `speak_name`, which is ALWAYS the person's real name — voice/TTS consumers are
required to read `speak_name`, never `nickname`. The asymmetry is deliberate and is
the whole point of the field: a nickname is something people grant each other, not
something a machine should presume to use.
VISITS ARE SAMPLED, NOT REPORTED: nothing pushes an arrival/departure event at this
service — `_presence_sampler_loop()` polls presence() on a timer and writes visit rows
on transitions. That makes the history honest about its own resolution (you know when
someone was *observed* home, to within one poll interval) and means "who was home when,
with whom" needs no separate table: co-presence is an interval-overlap query over
`visits`, not a second copy of the same truth that could drift from it.
DEVICE RIGHTS ARE AN ANSWER, NEVER AN ACTION: `device_grants` records that a person may
operate a specific HA entity (the "let my cousin unlock the front door" case), and
GET /device-access answers yes/no with a reason. This service never calls a lock, opens
a door, or talks to a device — HA asks, HA acts, exactly the "HA mediates, nothing
auto-acts" rule the rest of this project follows. Every check is logged to
`device_access_events` whether it was allowed or denied, because for a door lock the
denied ones are the interesting ones.
Endpoints:
- POST /register/photo raw image bytes -> stored as an audit artifact only,
NOT run through any face-matching (see README.md)
- POST /register the main call, described above
- GET /people admin/audit list of registered people + identifiers
- POST /people/<id> edit any editable field on a person (the admin panel)
- DELETE /people/<id>/identifiers/<id> revoke a mistaken/compromised identifier
- GET /presence who's currently home, resolved from registered identifiers
- GET /resolve spoken name OR nickname -> the canonical person
- GET /people/<id>/visits, GET /visits, GET /co-presence the visit history
- GET /prune/candidates, POST /people/prune last-visited-before cleanup
- GET /device-access, POST/DELETE /people/<id>/device-grants per-device rights
- GET/POST /people/<id>/chore-assignments who owes which chore
- GET /weather proxies the household smarthome/weather/current MQTT topic
"""
from __future__ import annotations
import json
import logging
import os
import re
import sqlite3
import sys
import threading
import time
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
import paho.mqtt.client as mqtt
LOG = logging.getLogger("identity")
TOKEN = os.environ.get("IDENTITY_TOKEN", "")
DB_PATH = Path(os.environ.get("IDENTITY_DB_PATH", "/data/identity.db"))
PHOTO_DIR = Path(os.environ.get("IDENTITY_PHOTO_DIR", "/data/photos"))
HA_URL = os.environ.get("HA_URL", "http://homeassistant:8123").rstrip("/")
HA_TOKEN = os.environ.get("HA_TOKEN", "")
# Comma-separated entity_id PREFIXES that are trusted as registration candidates.
# MUST be edited to match your real HA entity IDs (Developer Tools -> States) — these
# defaults are plausible-looking placeholders, not confirmed against a real HA
# instance. Only put Private BLE Device / fixed-tag entities here — never a raw
# bluetooth_le_tracker/device_tracker entity backed by an unresolved random MAC.
TRUSTED_ENTITY_PREFIXES = tuple(
p.strip() for p in os.environ.get(
"TRUSTED_ENTITY_PREFIXES", "device_tracker.pble_,device_tracker.bletag_"
).split(",") if p.strip()
)
# HA states considered "this device is present/nearby right now." device_tracker's
# own vocabulary is home/not_home; Bermuda-style room-level sensors may report a room
# name instead — PRESENT_STATES intentionally does not try to guess a room-level
# state string, since which room a device is in isn't needed for registration, only
# whether it's near the registering endpoint at all. Extend this list if your trusted
# entities use different state values (VERIFY against your instance).
PRESENT_STATES = {"home"}
# Floor-plan groundwork (docs/project-plan.md Phase 6 open decisions): the attribute
# key Bermuda (or whatever room-presence integration you use) attaches to a trusted
# entity's state to say which room/Area it's currently closest to — VERIFY the exact
# name against your own instance's Developer Tools -> States (Bermuda's own area-
# reporting attribute name has not been confirmed here; this is a plausible guess,
# same honesty rule as TRUSTED_ENTITY_PREFIXES). presence() reads this opportunistically
# and degrades to `room: null` if it's missing or unset, never breaking the payload —
# this is what lets a future floor-plan UI show "which room," not just "home/away,"
# without this service needing to change again once that UI actually gets built. The
# rendering itself (a floor-plan image, room<->coordinate mapping) is deliberately NOT
# built here — same "don't build against a guess" rule as everywhere else in this
# project; there's no floor plan or room list to design against yet.
AREA_ATTRIBUTE = os.environ.get("AREA_ATTRIBUTE", "area_id")
MAX_IMAGE_BYTES = int(os.environ.get("IDENTITY_MAX_IMAGE_MB", "15")) * 1024 * 1024
MAX_JSON_BYTES = 32 * 1024
NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 '.-]{0,63}$")
PHOTO_ID_RE = re.compile(r"^[a-f0-9]{16}$")
# Same character class as a name — a nickname is a name, just not the one on the
# birth certificate. Also matched against NAME_RE's length cap for the same reason.
NICKNAME_RE = NAME_RE
ENTITY_ID_RE = re.compile(r"^[a-z_]+\.[a-z0-9_]+$")
# --- Visit sampling (see the module docstring's "VISITS ARE SAMPLED") ---------------
# How often the sampler asks presence() who's home. 60s is deliberately coarse: this
# is a household log ("was Amir home Tuesday evening"), not a security audit trail,
# and every sample is a full HA /api/states fetch.
PRESENCE_POLL_SECONDS = int(os.environ.get("PRESENCE_POLL_SECONDS", "60"))
# How long a person has to read as NOT home before their visit is closed. BLE presence
# flaps — a phone in a pocket in the far corner of the flat drops off and comes back —
# and without this every such flap would end one visit and start another, turning one
# evening at home into forty "visits". Closing uses the LAST time they were actually
# seen, not the moment the grace expired, so the recorded departure stays honest.
DEPARTURE_GRACE_SECONDS = int(os.environ.get("DEPARTURE_GRACE_SECONDS", "900"))
# Safety net for a visit that never gets a definite "not home" to close it — someone
# device-less who was marked home by hand and never marked away, or a person whose
# identifier disappeared from HA entirely. Closed with close_reason='timed_out' rather
# than 'departed' so the record never claims to have observed a departure it didn't.
VISIT_MAX_OPEN_HOURS = float(os.environ.get("VISIT_MAX_OPEN_HOURS", "72"))
_sampler_stop = threading.Event()
_db_lock = threading.Lock()
_weather_lock = threading.Lock()
_last_weather: dict = {"available": False}
# Frigate face-recognition as a SECOND, corroborating presence signal (docs/
# project-plan.md Phase 20 — Tapo pan/tilt cameras) — never a registration signal
# (see the module docstring: the camera is audit-only for /register, always).
# `frigate/events`' recognized-name field is written from Frigate's documented
# `sub_label` shape (`["name", confidence]`), NOT verified against a real Frigate
# 0.16+ face-recognition deployment — VERIFY before trusting this. Degrades to
# simply never firing (BLE/manual presence still work) if the topic/shape is wrong.
FRIGATE_EVENTS_TOPIC = os.environ.get("FRIGATE_EVENTS_TOPIC", "frigate/events")
FACE_PRESENCE_WINDOW_SECONDS = int(os.environ.get("FACE_PRESENCE_WINDOW_SECONDS", "600"))
_face_lock = threading.Lock()
_last_face_seen: dict[str, float] = {} # lowercased name -> unix timestamp
def _now() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def _db() -> sqlite3.Connection:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(DB_PATH, timeout=10)
conn.execute("PRAGMA foreign_keys = ON")
conn.row_factory = sqlite3.Row
return conn
def _ensure_column(conn: sqlite3.Connection, table: str, column: str, ddl: str) -> None:
"""Add a column to an existing table if it isn't there yet. `CREATE TABLE IF NOT
EXISTS` only ever creates a table that's missing entirely — it silently does
nothing to a table that already exists with an older shape, so a deployment that
has been running since before a column was added would never gain it. Cheap enough
to run unconditionally at every startup (one PRAGMA per column).
"""
existing = {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")}
if column not in existing:
LOG.info("identity: migrating — adding %s.%s", table, column)
conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}")
def init_db() -> None:
with _db_lock, _db() as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS people (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE COLLATE NOCASE,
created_at TEXT NOT NULL,
photo_path TEXT,
-- Chore-system settings (see chores/README.md). Owned here, not in
-- chores/ itself, because a person's identity/household-standing is
-- this service's job — chores just reads /people and /presence.
-- chore_exempt: household member who's still tracked (a frequent
-- guest, e.g. a cousin) but never nudged about chores in general.
-- Litter is the deliberate exception (see chores/check.py's
-- _EXEMPTIONS_DONT_APPLY) — everyone is responsible for putting
-- trash they left in the bin, exempt or not.
chore_exempt INTEGER NOT NULL DEFAULT 0,
-- chore_reminder_style: free text, e.g. "be assertive, don't let up"
-- or "be gentle, give me a few minutes of grace" — passed to the
-- LLM that phrases (never decides) chore reminders. NULL/empty means
-- the plain, un-styled template.
chore_reminder_style TEXT,
-- nickname: what the household CALLS this person. An input alias for
-- /resolve only — never what a voice assistant says back, see the
-- module docstring and speak_name below. Deliberately not UNIQUE at
-- the SQL level: the constraint that actually matters is "must not
-- collide with anyone else's name OR nickname," which spans two
-- columns and is enforced in _nickname_conflict().
nickname TEXT,
-- note: free-text admin scratchpad ("Sarah's cousin, visits at
-- Christmas"). Never parsed by anything, never shown to the LLM.
note TEXT
);
CREATE TABLE IF NOT EXISTS visits (
id INTEGER PRIMARY KEY,
person_id INTEGER NOT NULL REFERENCES people(id) ON DELETE CASCADE,
arrived_at TEXT NOT NULL,
-- NULL means "still here" — exactly one open row per person at a time.
departed_at TEXT,
-- The most recent sample that still said "home." Departure is recorded
-- as of THIS, not as of when the grace window expired, so a visit never
-- claims someone was home during the 15 minutes we were waiting to be
-- sure they'd gone.
last_seen_at TEXT NOT NULL,
-- Which signal opened the visit: 'ble' | 'face' | 'manual'. Kept
-- because they are not equally trustworthy and a history that mixes
-- them without saying so would be quietly misleading.
source TEXT NOT NULL,
-- 'departed' (observed a definite not-home) | 'timed_out' (never got
-- one, closed by VISIT_MAX_OPEN_HOURS). Never conflated — see that
-- constant's comment.
close_reason TEXT
);
CREATE INDEX IF NOT EXISTS visits_person_arrived
ON visits (person_id, arrived_at);
-- Partial index over open visits only: the sampler's hottest query is
-- "who has a visit open right now," once per person per poll.
CREATE INDEX IF NOT EXISTS visits_open
ON visits (person_id) WHERE departed_at IS NULL;
CREATE TABLE IF NOT EXISTS device_grants (
id INTEGER PRIMARY KEY,
person_id INTEGER NOT NULL REFERENCES people(id) ON DELETE CASCADE,
ha_entity_id TEXT NOT NULL,
-- Free-form verb, defaulting to 'operate'. Not an enum because what
-- "operate" means is the caller's business (HA's), not this registry's
-- — it stores who may do what, it does not model device capabilities.
permission TEXT NOT NULL DEFAULT 'operate',
granted_at TEXT NOT NULL,
-- NULL = open-ended. A set expiry is the "cousin has the front door
-- for the weekend" case; it is checked at answer time, never by a
-- sweep, so a grant that lapses does so instantly rather than whenever
-- a cleanup job next happens to run.
expires_at TEXT,
note TEXT,
UNIQUE (person_id, ha_entity_id, permission)
);
CREATE TABLE IF NOT EXISTS device_access_events (
id INTEGER PRIMARY KEY,
-- Nullable on purpose: a check for an unknown person_id is exactly the
-- kind of thing worth having a row for.
person_id INTEGER,
ha_entity_id TEXT NOT NULL,
permission TEXT NOT NULL,
allowed INTEGER NOT NULL,
reason TEXT NOT NULL,
requested_via TEXT,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS device_access_events_created
ON device_access_events (created_at);
CREATE TABLE IF NOT EXISTS chore_assignments (
id INTEGER PRIMARY KEY,
person_id INTEGER NOT NULL REFERENCES people(id) ON DELETE CASCADE,
chore_type TEXT NOT NULL,
created_at TEXT NOT NULL,
UNIQUE (person_id, chore_type)
);
CREATE TABLE IF NOT EXISTS identifiers (
id INTEGER PRIMARY KEY,
person_id INTEGER NOT NULL REFERENCES people(id) ON DELETE CASCADE,
ha_entity_id TEXT NOT NULL UNIQUE,
registered_at TEXT NOT NULL,
registered_via_device TEXT
);
CREATE TABLE IF NOT EXISTS registration_events (
id INTEGER PRIMARY KEY,
person_id INTEGER,
device_id TEXT,
photo_path TEXT,
outcome TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS manual_presence (
person_id INTEGER PRIMARY KEY REFERENCES people(id) ON DELETE CASCADE,
home INTEGER NOT NULL,
updated_at TEXT NOT NULL
);
"""
)
# Columns added to `people` after the table first shipped — see
# _ensure_column()'s docstring for why the CREATE TABLE above isn't enough.
_ensure_column(conn, "people", "nickname", "TEXT")
_ensure_column(conn, "people", "note", "TEXT")
def _ha_get(path: str):
if not HA_TOKEN:
raise RuntimeError("HA_TOKEN is not configured")
req = urllib.request.Request(f"{HA_URL}{path}")
req.add_header("Authorization", f"Bearer {HA_TOKEN}")
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
def _trusted_present_candidates() -> list[dict]:
"""Entity_ids matching TRUSTED_ENTITY_PREFIXES whose current HA state indicates
presence right now. This is the ONLY source of registration candidates — see the
module docstring for why raw/unresolved MACs never reach this list.
"""
states = _ha_get("/api/states")
candidates = []
for entity in states:
entity_id = entity.get("entity_id", "")
if not entity_id.startswith(TRUSTED_ENTITY_PREFIXES):
continue
if entity.get("state") not in PRESENT_STATES:
continue
candidates.append(
{
"entity_id": entity_id,
"friendly_name": (entity.get("attributes") or {}).get("friendly_name", entity_id),
}
)
return candidates
def _already_claimed_by(conn: sqlite3.Connection, entity_id: str) -> sqlite3.Row | None:
return conn.execute(
"""
SELECT people.id, people.name FROM identifiers
JOIN people ON people.id = identifiers.person_id
WHERE identifiers.ha_entity_id = ?
""",
(entity_id,),
).fetchone()
def _match_by_name_or_nickname(conn: sqlite3.Connection, spoken: str) -> list[sqlite3.Row]:
"""Every person whose real name OR nickname matches `spoken`, case-insensitively.
Returns a LIST, not a single row, because the interesting case is when it has more
than one entry: someone's nickname colliding with someone else's real name is
exactly the ambiguity this project refuses to guess its way through. Callers must
decide what to do with >1 rather than silently taking the first.
"""
return conn.execute(
"SELECT * FROM people WHERE name = ? COLLATE NOCASE OR nickname = ? COLLATE NOCASE",
(spoken, spoken),
).fetchall()
def _nickname_conflict(conn: sqlite3.Connection, nickname: str, person_id: int) -> str | None:
"""The name a nickname must not steal: anyone ELSE's real name or nickname. Returns
the conflicting person's name, or None if the nickname is free. Enforced here in
Python rather than as a SQL constraint because it spans two columns across all
rows — see the `nickname` column comment.
"""
row = conn.execute(
"SELECT name FROM people WHERE id != ? "
"AND (name = ? COLLATE NOCASE OR nickname = ? COLLATE NOCASE)",
(person_id, nickname, nickname),
).fetchone()
return row["name"] if row else None
def _find_or_create_person(conn: sqlite3.Connection, name: str) -> tuple[int, bool]:
"""Resolves an existing person by real name or NICKNAME before creating one — so
"register me as Bibi" attaches Bibi's second phone to the person already recorded
as Linus rather than minting a duplicate. Raises AmbiguousName if the spoken string
matches two different people (see _match_by_name_or_nickname); callers turn that
into the same refuse-and-ask-a-human response as an ambiguous BLE candidate.
"""
matches = _match_by_name_or_nickname(conn, name)
if len(matches) > 1:
raise AmbiguousName([m["name"] for m in matches])
if matches:
return matches[0]["id"], False
cur = conn.execute("INSERT INTO people (name, created_at) VALUES (?, ?)", (name, _now()))
assert cur.lastrowid is not None
return cur.lastrowid, True
class AmbiguousName(Exception):
"""A spoken name matched more than one person (one by real name, another by
nickname). Never resolved by picking one — see the module docstring's
NEVER AUTO-COMMIT ON AMBIGUITY rule, which this is a second instance of.
"""
def __init__(self, names: list[str]) -> None:
super().__init__(", ".join(names))
self.names = names
def _ambiguous_name_response(exc: AmbiguousName) -> dict:
"""The refusal shape for AmbiguousName, shared by every registration path. Names
the colliding people out loud — the human standing at the panel is the one who can
fix this (by picking, or by changing a nickname), and they can't if the message
just says "ambiguous".
"""
return {
"ok": False,
"reason": "ambiguous_name",
"candidates": [{"name": n} for n in exc.names],
"message": (
f"That name matches more than one person here ({', '.join(exc.names)}) — "
"say a full name, or clear up the nickname in the admin panel."
),
}
def _set_profile_photo(conn: sqlite3.Connection, person_id: int, photo_path: str | None) -> None:
"""Every successful registration that captured a photo updates the person's
profile picture to it — "most recent registration photo wins" rather than
"first one wins," a simple, defensible default with no extra UI needed to pick
one. A no-op if this call didn't have a photo (camera unavailable, etc.) —
doesn't clear an existing profile photo just because a later re-registration
happened to skip the camera.
"""
if photo_path is None:
return
conn.execute("UPDATE people SET photo_path = ? WHERE id = ?", (photo_path, person_id))
def register(
name: str, device_id: str, photo_path: str | None, forced_entity_id: str | None, no_device: bool = False
) -> dict:
name = name.strip()
if not NAME_RE.match(name):
return {"ok": False, "reason": "bad_name", "message": "That name doesn't look valid."}
# THE "MY GRANDMOTHER DOESN'T HAVE A PHONE" CASE: registration doesn't have to
# find a device at all. A person with zero identifiers is a fully valid record —
# just one /presence can never resolve automatically (see presence()'s "unknown",
# not "away", handling below, and set_manual_presence() for the hand-operated
# alternative). This is an explicit opt-in flag, not a fallback for "no candidate
# found" — those stay separate cases (a real phone that just isn't nearby right
# now is a different situation from someone who was never going to have one).
if no_device:
with _db_lock, _db() as conn:
try:
person_id, is_new_person = _find_or_create_person(conn, name)
except AmbiguousName as exc:
return _ambiguous_name_response(exc)
_set_profile_photo(conn, person_id, photo_path)
_log_event(conn, person_id, device_id, photo_path, "registered_no_device")
return {
"ok": True,
"person_id": person_id,
"person_name": name,
"entity_id": None,
"is_new_person": is_new_person,
"is_new_device": False,
"message": (
f"Welcome, {name}! You're registered without a device — "
"use the door panel to mark yourself home or away by hand."
) if is_new_person else f"{name} is already registered.",
}
try:
candidates = _trusted_present_candidates()
except (urllib.error.URLError, urllib.error.HTTPError, RuntimeError) as exc:
LOG.warning("identity: could not reach HA for candidates", exc_info=True)
return {"ok": False, "reason": "ha_unreachable", "message": f"Could not reach Home Assistant: {exc}"}
with _db_lock, _db() as conn:
if forced_entity_id:
matches = [c for c in candidates if c["entity_id"] == forced_entity_id]
if not matches:
return {
"ok": False,
"reason": "not_a_candidate",
"message": "That device isn't showing as nearby right now.",
}
chosen = matches[0]
elif len(candidates) == 0:
_log_event(conn, None, device_id, photo_path, "no_candidate")
return {
"ok": False,
"reason": "no_candidate",
"message": "I couldn't find your phone nearby — make sure Bluetooth is on and try again.",
}
elif len(candidates) > 1:
_log_event(conn, None, device_id, photo_path, "ambiguous")
return {
"ok": False,
"reason": "ambiguous",
"candidates": candidates,
"message": "I found more than one device nearby — pick yours on the screen.",
}
else:
chosen = candidates[0]
# Whether this device is already someone's is compared against the RESOLVED
# person, not the raw spoken string — otherwise registering a second phone as
# "Bibi" when the first is filed under Linus (whose nickname is Bibi) would
# look like a stranger claiming Linus's device and get refused.
spoken_matches = _match_by_name_or_nickname(conn, name)
if len(spoken_matches) > 1:
return _ambiguous_name_response(AmbiguousName([m["name"] for m in spoken_matches]))
spoken_person_id = spoken_matches[0]["id"] if spoken_matches else None
claimed_by = _already_claimed_by(conn, chosen["entity_id"])
if claimed_by is not None and claimed_by["id"] != spoken_person_id:
_log_event(conn, claimed_by["id"], device_id, photo_path, "already_claimed")
return {
"ok": False,
"reason": "already_claimed",
"message": "That device is already registered to someone else.",
}
if claimed_by is not None:
# Same person re-registering the same device — a harmless no-op, not an
# error (e.g. re-running the flow after a network hiccup).
_set_profile_photo(conn, claimed_by["id"], photo_path)
_log_event(conn, claimed_by["id"], device_id, photo_path, "already_registered")
return {
"ok": True,
"person_id": claimed_by["id"],
"person_name": claimed_by["name"],
"entity_id": chosen["entity_id"],
"is_new_person": False,
"is_new_device": False,
"message": f"You're already registered, {claimed_by['name']}.",
}
try:
person_id, is_new_person = _find_or_create_person(conn, name)
except AmbiguousName as exc:
return _ambiguous_name_response(exc)
conn.execute(
"INSERT INTO identifiers (person_id, ha_entity_id, registered_at, registered_via_device) "
"VALUES (?, ?, ?, ?)",
(person_id, chosen["entity_id"], _now(), device_id),
)
_set_profile_photo(conn, person_id, photo_path)
_log_event(conn, person_id, device_id, photo_path, "registered")
device_count = conn.execute(
"SELECT COUNT(*) AS n FROM identifiers WHERE person_id = ?", (person_id,)
).fetchone()["n"]
message = (
f"Welcome, {name}!" if is_new_person
else f"Got it — that's device number {device_count} for {name}."
)
return {
"ok": True,
"person_id": person_id,
"person_name": name,
"entity_id": chosen["entity_id"],
"is_new_person": is_new_person,
"is_new_device": True,
"device_count": device_count,
"message": message,
}
def _log_event(conn, person_id, device_id, photo_path, outcome) -> None:
conn.execute(
"INSERT INTO registration_events (person_id, device_id, photo_path, outcome, created_at) "
"VALUES (?, ?, ?, ?, ?)",
(person_id, device_id, photo_path, outcome, _now()),
)
def _person_payload(conn: sqlite3.Connection, person: sqlite3.Row) -> dict:
"""One person's full admin-facing record. Shared by list_people() and the
single-person reads so the admin panel never has to reconcile two shapes.
"""
identifiers = conn.execute(
"SELECT id, ha_entity_id, registered_at, registered_via_device "
"FROM identifiers WHERE person_id = ? ORDER BY registered_at",
(person["id"],),
).fetchall()
grants = conn.execute(
"SELECT id, ha_entity_id, permission, granted_at, expires_at, note "
"FROM device_grants WHERE person_id = ? ORDER BY ha_entity_id",
(person["id"],),
).fetchall()
assignments = conn.execute(
"SELECT chore_type FROM chore_assignments WHERE person_id = ? ORDER BY chore_type",
(person["id"],),
).fetchall()
stats = conn.execute(
"SELECT COUNT(*) AS n, MAX(arrived_at) AS last_arrived FROM visits WHERE person_id = ?",
(person["id"],),
).fetchone()
open_visit = conn.execute(
"SELECT arrived_at FROM visits WHERE person_id = ? AND departed_at IS NULL",
(person["id"],),
).fetchone()
return {
"id": person["id"],
"name": person["name"],
"nickname": person["nickname"],
# THE FIELD VOICE/TTS CONSUMERS MUST USE. Always the real name, never the
# nickname, no matter which of the two the human said to get here — see the
# module docstring. Emitted as its own key rather than leaving callers to
# "just use name" because a caller that has both fields in front of it will
# eventually reach for the friendlier-looking one.
"speak_name": person["name"],
"note": person["note"],
"created_at": person["created_at"],
# The path itself is never exposed — an internal container
# filesystem detail — just whether GET /people/<id>/photo has
# anything to serve.
"has_photo": person["photo_path"] is not None,
"chore_exempt": bool(person["chore_exempt"]),
"chore_reminder_style": person["chore_reminder_style"],
"chore_assignments": [a["chore_type"] for a in assignments],
"identifiers": [dict(i) for i in identifiers],
"device_grants": [dict(g) for g in grants],
"visit_count": stats["n"],
# Falls back to created_at for someone with no recorded visits at all — a
# person registered before visit sampling existed, or one who has genuinely
# never been seen home. Pruning needs SOME date for everyone or the never-
# visited (the most prunable people of all) would silently never be selected.
"last_visit_at": stats["last_arrived"] or person["created_at"],
"last_visit_is_estimated": stats["last_arrived"] is None,
"currently_home_since": open_visit["arrived_at"] if open_visit else None,
}
def list_people() -> list[dict]:
with _db_lock, _db() as conn:
people = conn.execute("SELECT * FROM people ORDER BY name COLLATE NOCASE").fetchall()
return [_person_payload(conn, person) for person in people]
def set_chore_settings(person_id: int, chore_exempt: bool | None, reminder_style: str | None) -> bool:
"""Either field left as None leaves that column untouched — lets a caller update
just one of the two without needing to know the other's current value first.
"""
with _db_lock, _db() as conn:
exists = conn.execute("SELECT 1 FROM people WHERE id = ?", (person_id,)).fetchone()
if not exists:
return False
if chore_exempt is not None:
conn.execute("UPDATE people SET chore_exempt = ? WHERE id = ?", (int(chore_exempt), person_id))
if reminder_style is not None:
conn.execute(
"UPDATE people SET chore_reminder_style = ? WHERE id = ?",
(reminder_style.strip() or None, person_id),
)
return True
def update_person(person_id: int, fields: dict) -> dict:
"""The admin panel's "edit every field" call. Only keys actually present in
`fields` are touched — a partial edit never blanks the fields it didn't mention,
which is what makes the frontend able to send one changed input rather than
round-tripping the whole record and racing anyone else editing it.
`created_at` is deliberately NOT editable: it records when this person entered the
household's records, which is a fact about what happened, not a preference. Neither
is `id`. Everything a human might actually want to change is here, including
renaming a "Guest 4" into a real person once you learn who they are — that's the
intended promotion path, not a separate endpoint.
"""
with _db_lock, _db() as conn:
person = conn.execute("SELECT * FROM people WHERE id = ?", (person_id,)).fetchone()
if person is None:
return {"ok": False, "reason": "not_found", "message": "No such person."}
updates: list[tuple[str, object]] = []
if "name" in fields:
name = str(fields["name"] or "").strip()
if not NAME_RE.match(name):
return {"ok": False, "reason": "bad_name", "message": "That name doesn't look valid."}
clash = conn.execute(
"SELECT name FROM people WHERE id != ? "
"AND (name = ? COLLATE NOCASE OR nickname = ? COLLATE NOCASE)",
(person_id, name, name),
).fetchone()
if clash:
return {
"ok": False,
"reason": "name_taken",
"message": f"{name} already belongs to someone here.",
}
updates.append(("name", name))
if "nickname" in fields:
raw = fields["nickname"]
nickname = str(raw or "").strip()
if not nickname:
updates.append(("nickname", None)) # clearing it is a legitimate edit
elif not NICKNAME_RE.match(nickname):
return {"ok": False, "reason": "bad_nickname", "message": "That nickname doesn't look valid."}
else:
clash = _nickname_conflict(conn, nickname, person_id)
if clash:
return {
"ok": False,
"reason": "nickname_taken",
"message": f"“{nickname}” would collide with {clash} — pick another.",
}
updates.append(("nickname", nickname))
if "chore_exempt" in fields:
updates.append(("chore_exempt", int(bool(fields["chore_exempt"]))))
for text_field in ("chore_reminder_style", "note"):
if text_field in fields:
value = fields[text_field]
if value is not None and not isinstance(value, str):
return {"ok": False, "reason": "bad_field", "message": f"'{text_field}' must be text."}
updates.append((text_field, (value or "").strip() or None))
# Clearing the profile picture is an edit like any other. The file itself is
# left on disk on purpose — it's also a registration_events audit artifact,
# and "stop showing this photo" is a different request from "destroy the
# record that this photo was taken."
if fields.get("clear_photo"):
updates.append(("photo_path", None))
for column, value in updates:
conn.execute(f"UPDATE people SET {column} = ? WHERE id = ?", (value, person_id))
updated = conn.execute("SELECT * FROM people WHERE id = ?", (person_id,)).fetchone()
return {"ok": True, "person": _person_payload(conn, updated), "changed": [c for c, _ in updates]}
def add_identifier(person_id: int, entity_id: str) -> dict:
"""Attach an identifier by hand, for the case registration can't cover: a fixed-MAC
BLE tag provisioned for someone before they (or the tag) are anywhere near the
door panel.
STILL BEHIND THE ANTI-SPOOFING BOUNDARY: the entity_id must match
TRUSTED_ENTITY_PREFIXES, exactly like a registration candidate. What this endpoint
relaxes is only the "must be present RIGHT NOW" requirement — being in the room is
a convenience check, whereas the trusted-prefix allowlist is the actual security
property (see the module docstring), and an admin panel is not a reason to hand out
an exception to it.
"""
entity_id = entity_id.strip()
if not ENTITY_ID_RE.match(entity_id):
return {"ok": False, "reason": "bad_entity_id", "message": "That doesn't look like an entity_id."}
if not entity_id.startswith(TRUSTED_ENTITY_PREFIXES):
return {
"ok": False,
"reason": "untrusted_entity",
"message": (
"That entity isn't in TRUSTED_ENTITY_PREFIXES — only IRK-resolved "
"Private BLE Device or fixed-tag entities can be identifiers."
),
}
with _db_lock, _db() as conn:
if conn.execute("SELECT 1 FROM people WHERE id = ?", (person_id,)).fetchone() is None:
return {"ok": False, "reason": "not_found", "message": "No such person."}
claimed = _already_claimed_by(conn, entity_id)
if claimed is not None:
return {
"ok": False,
"reason": "already_claimed",
"message": f"That device already belongs to {claimed['name']}.",
}
conn.execute(
"INSERT INTO identifiers (person_id, ha_entity_id, registered_at, registered_via_device) "
"VALUES (?, ?, ?, ?)",
(person_id, entity_id, _now(), "admin-panel"),
)
return {"ok": True, "message": f"Added {entity_id}."}
def resolve_name(spoken: str) -> dict:
"""Spoken name or nickname -> the canonical person. The point of the whole nickname
feature on the input side: household members say "is Bibi home?", this turns that
into Linus's record. The response always carries `speak_name` (the real name) so
the assistant that asked can phrase its answer with the right one — see the module
docstring's NICKNAMES ARE FOR HUMANS TO SAY rule.
"""
spoken = (spoken or "").strip()
if not spoken:
return {"matched": False, "reason": "empty"}
with _db_lock, _db() as conn:
matches = _match_by_name_or_nickname(conn, spoken)
if not matches:
return {"matched": False, "reason": "unknown", "query": spoken}
if len(matches) > 1:
return {
"matched": False,
"reason": "ambiguous_name",
"query": spoken,
"candidates": [{"id": m["id"], "name": m["name"]} for m in matches],
}
person = matches[0]
return {
"matched": True,
"query": spoken,
"id": person["id"],
"name": person["name"],
"nickname": person["nickname"],
"speak_name": person["name"],
"matched_on": "name" if (person["name"] or "").lower() == spoken.lower() else "nickname",
}
def get_person_photo(person_id: int) -> bytes | None:
with _db_lock, _db() as conn:
row = conn.execute("SELECT photo_path FROM people WHERE id = ?", (person_id,)).fetchone()
if row is None or row["photo_path"] is None:
return None
path = Path(row["photo_path"])
if not path.is_file():
return None
return path.read_bytes()
def delete_identifier(person_id: int, identifier_id: int) -> bool:
with _db_lock, _db() as conn:
cur = conn.execute(
"DELETE FROM identifiers WHERE id = ? AND person_id = ?", (identifier_id, person_id)
)
return cur.rowcount > 0
def delete_person(person_id: int) -> bool:
"""Mainly for cleaning up stale/anonymous Guest records (see register_guest) —
equally usable for any person, ON DELETE CASCADE takes their identifiers with
them.
"""
with _db_lock, _db() as conn:
cur = conn.execute("DELETE FROM people WHERE id = ?", (person_id,))
return cur.rowcount > 0
def register_guest(device_id: str, photo_path: str | None) -> dict:
"""The "doesn't need to know who it is" path — no name, no BLE candidate lookup
at all, just a household record that someone who isn't a registered resident is
around. Always creates a NEW person ("Guest 1", "Guest 2", ...) rather than
reusing one — unlike named registration, two guest visits are not assumed to be
the same person the way two registrations of "Amir" are. See delete_person() for
cleaning up a stale guest entry afterwards; nothing here expires them
automatically.
"""
with _db_lock, _db() as conn:
existing = conn.execute("SELECT COUNT(*) AS n FROM people WHERE name LIKE 'Guest %'").fetchone()["n"]
name = f"Guest {existing + 1}"
person_id, _ = _find_or_create_person(conn, name)
_set_profile_photo(conn, person_id, photo_path)
_log_event(conn, person_id, device_id, photo_path, "registered_guest")
return {
"ok": True,
"person_id": person_id,
"person_name": name,
"entity_id": None,
"is_new_person": True,
"is_new_device": False,
"message": f"Added {name}.",
}
def set_manual_presence(person_id: int, home: bool) -> bool:
"""The hand-operated alternative to BLE presence, for anyone with zero
identifiers (grandmother, a guest) — see presence()'s handling below. Silently
accepted for a person who *does* have identifiers too (harmless, just ignored by
presence() in that case) rather than rejected, since there's no safety reason to
forbid it and one fewer edge case to special-case in the caller.
"""
with _db_lock, _db() as conn:
exists = conn.execute("SELECT 1 FROM people WHERE id = ?", (person_id,)).fetchone()
if not exists:
return False
conn.execute(
"INSERT INTO manual_presence (person_id, home, updated_at) VALUES (?, ?, ?) "
"ON CONFLICT(person_id) DO UPDATE SET home = excluded.home, updated_at = excluded.updated_at",
(person_id, int(home), _now()),
)
return True
def presence() -> dict:
"""Who's home. For a person with at least one registered identifier, resolved
from that identifier's current HA state, same as before. For a person with
ZERO identifiers (grandmother, a guest — see register()'s no_device path and
register_guest()) there is nothing to resolve automatically, so this reports
`home: null` ("unknown") rather than `false` ("away") unless a manual override
has been set via set_manual_presence() — reporting them as away by default would
be actively wrong, not just uninformative, the moment they're actually home.
Does not assume or require pre-existing HA `person.*` entities, since this
service is itself the source of truth for name<->identifier mapping. Degrades to
an empty list (never an error page) if nobody is registered yet or HA is
unreachable, same "degrade, don't blank" rule as every other renderer in this
project.
"""
people = list_people()
if not people:
return {"people": [], "generated_at": _now()}
try:
states = {s["entity_id"]: s for s in _ha_get("/api/states")}
ha_ok = True
except (urllib.error.URLError, urllib.error.HTTPError, RuntimeError):
LOG.warning("identity: could not reach HA for presence", exc_info=True)
states = {}
ha_ok = False
with _db_lock, _db() as conn:
manual = {
row["person_id"]: bool(row["home"])
for row in conn.execute("SELECT person_id, home FROM manual_presence")
}
with _face_lock:
seen_recently = {
name for name, ts in _last_face_seen.items() if time.time() - ts <= FACE_PRESENCE_WINDOW_SECONDS
}
result = []
for person in people:
room = None
face_seen = person["name"].strip().lower() in seen_recently
if person["identifiers"] and ha_ok:
entity_states = [states[i["ha_entity_id"]] for i in person["identifiers"] if i["ha_entity_id"] in states]
# A second, corroborating signal — see FRIGATE_EVENTS_TOPIC's
# module-level comment. Camera face recognition is never a registration
# signal (see the module docstring), only ever an OR-ed-in presence one.
home = face_seen or any(s.get("state") in PRESENT_STATES for s in entity_states)
# Floor-plan groundwork — see AREA_ATTRIBUTE's module-level comment.
# First identifier that actually reports one wins; a person with two
# phones in two different rooms is a real but rare edge case not worth
# more than "pick one" for a v1 that has no map to show it on yet anyway.
for s in entity_states:
area = (s.get("attributes") or {}).get(AREA_ATTRIBUTE)
if area:
room = area
break
elif person["identifiers"]:
home = True if face_seen else None # HA unreachable, but a camera sighting still counts
else:
# Device-less (grandmother, a guest): face recognition can confirm
# "home" even though nothing here can ever confirm "away" from a
# camera alone (not seen recently just means not seen, not proven
# absent) — so a recent sighting overrides a stale/never-set manual
# flag, but a manual "away" is never second-guessed by a camera miss.
home = True if face_seen else manual.get(person["id"])
result.append(
{
"id": person["id"],
"name": person["name"],
"nickname": person["nickname"],
# Always the real name — see the module docstring. `chores/` phrases
# reminders off this, so a nudge never addresses someone by a nickname
# the machine wasn't given.
"speak_name": person["speak_name"],
"home": home,
"room": room,
"has_device": bool(person["identifiers"]),
"has_photo": person["has_photo"],
"face_seen_recently": face_seen,
# Chore-system settings, straight passthrough — see chores/README.md
# for how these are used (never anything presence-related itself).
"chore_exempt": person["chore_exempt"],
"chore_reminder_style": person["chore_reminder_style"],
"chore_assignments": person["chore_assignments"],
}
)
payload = {"people": result, "generated_at": _now()}
if not ha_ok:
payload["error"] = "ha_unreachable"
return payload
# --- Visit history: who was home when, and with whom -------------------------------
# See the module docstring's "VISITS ARE SAMPLED, NOT REPORTED" for the shape of this
# and why co-presence is a query rather than a table.
def _parse_iso(value: str) -> datetime:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
def _record_presence_sample(sample: dict) -> None:
"""One sampler pass: turn "who reads as home right now" into open/closed visit
rows. Three states in, three different behaviours out, and the third is the one
that matters:
home is True -> open a visit, or extend the open one's last_seen_at
home is False -> close the open visit, but only once DEPARTURE_GRACE_SECONDS
of not-home have passed (BLE flaps; see that constant)
home is None -> DO NOTHING AT ALL. "Unknown" is not "away". A device-less
person nobody has toggled, or an HA outage, must never write a
departure into the history — an inferred absence recorded as an
observed one is a lie the history can never be un-told.
"""
if sample.get("error") == "ha_unreachable":
# Every BLE-backed person reads as unknown during an outage, so there's nothing
# trustworthy to record. Skipping the whole pass (rather than just the unknowns)
# also keeps the VISIT_MAX_OPEN_HOURS sweep from counting an outage against a
# visit it can't currently see the state of.
return
now = _now()
with _db_lock, _db() as conn:
for person in sample.get("people", []):
person_id = person["id"]
open_visit = conn.execute(
"SELECT * FROM visits WHERE person_id = ? AND departed_at IS NULL", (person_id,)
).fetchone()
if person.get("home") is True:
if open_visit is None:
source = "face" if person.get("face_seen_recently") else (
"ble" if person.get("has_device") else "manual"
)
conn.execute(
"INSERT INTO visits (person_id, arrived_at, last_seen_at, source) "
"VALUES (?, ?, ?, ?)",
(person_id, now, now, source),
)
LOG.info("identity: %s arrived (source: %s)", person["name"], source)
else:
conn.execute("UPDATE visits SET last_seen_at = ? WHERE id = ?", (now, open_visit["id"]))
elif person.get("home") is False and open_visit is not None:
gone_for = (_parse_iso(now) - _parse_iso(open_visit["last_seen_at"])).total_seconds()
if gone_for >= DEPARTURE_GRACE_SECONDS:
conn.execute(
"UPDATE visits SET departed_at = ?, close_reason = 'departed' WHERE id = ?",
(open_visit["last_seen_at"], open_visit["id"]),
)
LOG.info("identity: %s departed (last seen %s)", person["name"], open_visit["last_seen_at"])
# The stuck-visit sweep — see VISIT_MAX_OPEN_HOURS.
cutoff = (datetime.now(timezone.utc) - timedelta(hours=VISIT_MAX_OPEN_HOURS)).isoformat().replace("+00:00", "Z")
stale = conn.execute(
"UPDATE visits SET departed_at = last_seen_at, close_reason = 'timed_out' "
"WHERE departed_at IS NULL AND last_seen_at < ?",
(cutoff,),
)
if stale.rowcount:
LOG.info("identity: closed %d visit(s) that went stale past %sh", stale.rowcount, VISIT_MAX_OPEN_HOURS)
def _presence_sampler_loop() -> None:
"""The only thing in this service that runs on its own initiative. Deliberately a
plain daemon thread rather than a systemd timer + separate entrypoint (like
chores/): the visit log is this service's own state, sampled from its own
presence() — handing that to an external scheduler would mean a second process
writing the same SQLite file for no gain.
Never lets an exception kill the thread: a sampler that dies quietly would leave a
permanently frozen visit history that still *looks* fine from the outside, the
worst possible failure mode for a log.
"""
while not _sampler_stop.wait(PRESENCE_POLL_SECONDS):
try:
_record_presence_sample(presence())
except Exception:
LOG.warning("identity: presence sampler pass failed, will retry", exc_info=True)
def list_visits(person_id: int | None = None, since: str | None = None, limit: int = 200) -> list[dict]:
clauses, params = [], []
if person_id is not None:
clauses.append("visits.person_id = ?")
params.append(person_id)
if since:
clauses.append("visits.arrived_at >= ?")
params.append(since)
where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
params.append(max(1, min(limit, 1000)))
with _db_lock, _db() as conn:
rows = conn.execute(
f"""
SELECT visits.*, people.name, people.nickname
FROM visits JOIN people ON people.id = visits.person_id
{where}
ORDER BY visits.arrived_at DESC
LIMIT ?
""",
params,
).fetchall()
return [
{
"id": r["id"],
"person_id": r["person_id"],
"name": r["name"],
"speak_name": r["name"],
"nickname": r["nickname"],
"arrived_at": r["arrived_at"],
"departed_at": r["departed_at"],
"last_seen_at": r["last_seen_at"],
"source": r["source"],
"close_reason": r["close_reason"],
"ongoing": r["departed_at"] is None,
}
for r in rows
]
def co_presence(person_id: int | None = None, since: str | None = None, limit: int = 200) -> dict:
""""Who was home when, with whom" — computed by overlapping visit intervals, never
stored. Two visits overlap iff each starts before the other ends, with an open
visit (departed_at IS NULL) treated as ending now.
Deriving it means it can never disagree with the visit log it came from, and it
costs nothing to keep correct when a visit is later closed or corrected. The
tradeoff is honest: this is O(visits²) inside the window, which is fine for a
household's worth of rows and would not be for a venue's.
"""
visits = list_visits(person_id=None, since=since, limit=1000)
now = _now()
def overlap(a: dict, b: dict) -> tuple[str, str] | None:
start = max(a["arrived_at"], b["arrived_at"])
end = min(a["departed_at"] or now, b["departed_at"] or now)
return (start, end) if start < end else None
subject = [v for v in visits if person_id is None or v["person_id"] == person_id]
overlaps = []
for visit in subject:
for other in visits:
if other["person_id"] == visit["person_id"] or other["id"] == visit["id"]:
continue
# Without this, every pair shows up twice (once from each side) whenever
# the caller didn't narrow to one person.
if person_id is None and other["id"] < visit["id"]:
continue
window = overlap(visit, other)
if window is None:
continue
overlaps.append(
{
"from": window[0],
"until": window[1],
"ongoing": visit["departed_at"] is None and other["departed_at"] is None,
"people": [
{"id": visit["person_id"], "name": visit["name"]},
{"id": other["person_id"], "name": other["name"]},
],
}
)
overlaps.sort(key=lambda o: o["from"], reverse=True)
return {"overlaps": overlaps[: max(1, min(limit, 1000))], "generated_at": now}
# --- Pruning: last-visited-before ---------------------------------------------------
def prune_candidates(last_visit_before: str) -> dict:
"""Everyone whose most recent visit predates `last_visit_before` — the admin
panel's "select all that have last visited before <date>" checkbox filter. A READ.
It selects; it never deletes. See prune_people() for why those are separate calls.
Someone with no recorded visits at all falls back to their created_at (see
_person_payload) — otherwise a person registered once and never seen since, the
single most prunable record there is, would be the one this filter could never
find.
"""
people = list_people()
matching = [p for p in people if p["last_visit_at"] < last_visit_before]
return {
"last_visit_before": last_visit_before,
"candidates": matching,
"count": len(matching),
"generated_at": _now(),
}
def prune_people(person_ids: list[int]) -> dict:
"""Bulk-delete BY EXPLICIT ID, never by filter. The frontend runs prune_candidates()
to fill in the checkboxes and then sends back the ids the human actually looked at
and confirmed.
That indirection is the point: a filter re-evaluated at delete time could quietly
take someone who came home in the seconds between the preview and the click, and
"the list I approved is the list that got deleted" is worth one extra round trip
for an irreversible operation on people's records.
"""
deleted, missing = [], []
with _db_lock, _db() as conn:
for person_id in person_ids:
row = conn.execute("SELECT name FROM people WHERE id = ?", (person_id,)).fetchone()
if row is None:
missing.append(person_id)
continue
conn.execute("DELETE FROM people WHERE id = ?", (person_id,))
deleted.append({"id": person_id, "name": row["name"]})
if deleted:
LOG.info("identity: pruned %d person record(s): %s", len(deleted), ", ".join(d["name"] for d in deleted))
return {"ok": True, "deleted": deleted, "not_found": missing}
# --- Per-device rights --------------------------------------------------------------
# See the module docstring's "DEVICE RIGHTS ARE AN ANSWER, NEVER AN ACTION".
def grant_device(person_id: int, entity_id: str, permission: str, expires_at: str | None, note: str | None) -> dict:
entity_id = (entity_id or "").strip()
if not ENTITY_ID_RE.match(entity_id):
return {"ok": False, "reason": "bad_entity_id", "message": "That doesn't look like an entity_id."}
permission = (permission or "operate").strip() or "operate"
if expires_at:
try:
_parse_iso(expires_at)
except ValueError:
return {"ok": False, "reason": "bad_expiry", "message": "expires_at must be an ISO-8601 timestamp."}
with _db_lock, _db() as conn:
if conn.execute("SELECT 1 FROM people WHERE id = ?", (person_id,)).fetchone() is None:
return {"ok": False, "reason": "not_found", "message": "No such person."}
conn.execute(
"INSERT INTO device_grants (person_id, ha_entity_id, permission, granted_at, expires_at, note) "
"VALUES (?, ?, ?, ?, ?, ?) "
"ON CONFLICT(person_id, ha_entity_id, permission) DO UPDATE SET "
"expires_at = excluded.expires_at, note = excluded.note, granted_at = excluded.granted_at",
(person_id, entity_id, permission, _now(), expires_at or None, (note or "").strip() or None),
)
LOG.info("identity: granted %s on %s to person %d", permission, entity_id, person_id)
return {"ok": True, "message": f"Granted {permission} on {entity_id}."}
def revoke_device_grant(person_id: int, grant_id: int) -> bool:
with _db_lock, _db() as conn:
cur = conn.execute("DELETE FROM device_grants WHERE id = ? AND person_id = ?", (grant_id, person_id))
return cur.rowcount > 0
def check_device_access(person_id: int, entity_id: str, permission: str, requested_via: str | None) -> dict:
""""May this person operate this device?" — the call HA makes before it unlocks
anything. Answers, logs, and returns; it never touches the device itself.
DENY IS THE DEFAULT AND THE ONLY FALLBACK. Every path that isn't an unexpired
matching grant returns allowed: false, including a person who doesn't exist and a
grant whose expiry has passed. This is the one place in this service where failing
closed matters more than degrading gracefully: everything else here would rather
report "unknown" than guess, but a lock has no useful "unknown" — the door is
either opened or it isn't, and the safe half of that is "isn't".
"""
permission = (permission or "operate").strip() or "operate"
entity_id = (entity_id or "").strip()
now = _now()
with _db_lock, _db() as conn:
person = conn.execute("SELECT name FROM people WHERE id = ?", (person_id,)).fetchone()
if person is None:
allowed, reason, grant = False, "unknown_person", None
else:
grant = conn.execute(
"SELECT * FROM device_grants WHERE person_id = ? AND ha_entity_id = ? AND permission = ?",
(person_id, entity_id, permission),
).fetchone()
if grant is None:
allowed, reason = False, "no_grant"
elif grant["expires_at"] and grant["expires_at"] <= now:
allowed, reason = False, "expired"
else:
allowed, reason = True, "granted"
conn.execute(
"INSERT INTO device_access_events "
"(person_id, ha_entity_id, permission, allowed, reason, requested_via, created_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(person_id, entity_id, permission, int(allowed), reason, requested_via, now),
)
LOG.info(
"identity: device-access %s — person %d on %s (%s): %s",
"ALLOW" if allowed else "DENY", person_id, entity_id, permission, reason,
)
return {
"allowed": allowed,
"reason": reason,
"person_id": person_id,
"person_name": person["name"] if person else None,
"entity_id": entity_id,
"permission": permission,
"expires_at": grant["expires_at"] if grant else None,
"checked_at": now,
}
def list_device_access_events(limit: int = 100) -> list[dict]:
with _db_lock, _db() as conn:
rows = conn.execute(
"SELECT device_access_events.*, people.name FROM device_access_events "
"LEFT JOIN people ON people.id = device_access_events.person_id "
"ORDER BY device_access_events.created_at DESC LIMIT ?",
(max(1, min(limit, 1000)),),
).fetchall()
return [{**dict(r), "allowed": bool(r["allowed"])} for r in rows]
# --- Chore assignments --------------------------------------------------------------
def set_chore_assignments(person_id: int, chore_types: list[str]) -> dict:
"""Replaces this person's whole assignment set — the admin panel edits it as a row
of checkboxes, so "what's ticked now" is the natural unit, not per-type add/remove.
`chores/` treats an assignment as a strong PREFERENCE, not a lock (see
chores/README.md): the household principle is still "I don't care who does it as
long as it gets done," so if the assignee isn't home the nudge falls through to
whoever is, unless CHORE_ASSIGNMENT_STRICT is on over there. Litter ignores
assignments entirely, for the same reason it ignores exemptions — putting your own
trash in the bin was never a chore anyone could be assigned.
"""
cleaned = sorted({str(t).strip().lower() for t in chore_types if str(t).strip()})
with _db_lock, _db() as conn:
if conn.execute("SELECT 1 FROM people WHERE id = ?", (person_id,)).fetchone() is None:
return {"ok": False, "reason": "not_found", "message": "No such person."}
conn.execute("DELETE FROM chore_assignments WHERE person_id = ?", (person_id,))
for chore_type in cleaned:
conn.execute(
"INSERT INTO chore_assignments (person_id, chore_type, created_at) VALUES (?, ?, ?)",
(person_id, chore_type, _now()),
)
return {"ok": True, "chore_assignments": cleaned}
def chore_assignments() -> dict:
"""chore_type -> the people assigned to it. The shape `chores/` actually wants;
/people carries the same facts per-person for the admin panel's benefit.
"""
with _db_lock, _db() as conn:
rows = conn.execute(
"SELECT chore_assignments.chore_type, people.id, people.name "
"FROM chore_assignments JOIN people ON people.id = chore_assignments.person_id "
"ORDER BY chore_assignments.chore_type, people.name COLLATE NOCASE"
).fetchall()
by_type: dict[str, list[dict]] = {}
for row in rows:
by_type.setdefault(row["chore_type"], []).append({"id": row["id"], "name": row["name"]})
return {"assignments": by_type, "generated_at": _now()}
def _on_weather_message(_client, _userdata, message) -> None:
try:
payload = json.loads(message.payload.decode("utf-8", "replace"))
except ValueError:
return
with _weather_lock:
global _last_weather
_last_weather = {**payload, "available": True}
def _on_frigate_event(_client, _userdata, message) -> None:
"""Best-effort only — see FRIGATE_EVENTS_TOPIC's module-level comment. A
malformed or unexpected payload just means this particular event contributes no
presence signal, never an error that could take the weather subscription (or
anything else) down with it.
"""
try:
payload = json.loads(message.payload.decode("utf-8", "replace"))
sub_label = ((payload.get("after") or {}).get("sub_label"))
name = sub_label[0] if isinstance(sub_label, list) and sub_label else None
if not name:
return
with _face_lock:
_last_face_seen[str(name).strip().lower()] = time.time()
except Exception:
pass
def _on_mqtt_message(client, userdata, message) -> None:
if message.topic == "smarthome/weather/current":
_on_weather_message(client, userdata, message)
elif message.topic == FRIGATE_EVENTS_TOPIC:
_on_frigate_event(client, userdata, message)
def start_mqtt(broker_host: str, broker_port: int, username: str, password: str) -> None:
if not broker_host:
LOG.warning("identity: MQTT_BROKER_HOST not set — /weather will always report unavailable")
return
callback_api = getattr(mqtt, "CallbackAPIVersion", None)
client = mqtt.Client(callback_api.VERSION1) if callback_api is not None else mqtt.Client()
if username:
client.username_pw_set(username, password or None)
def on_connect(c, _userdata, _flags, rc):
if rc == 0:
c.subscribe("smarthome/weather/current", qos=1)
c.subscribe(FRIGATE_EVENTS_TOPIC, qos=0)
client.on_connect = on_connect
client.on_message = _on_mqtt_message
client.connect_async(broker_host, broker_port, keepalive=60)
client.loop_start()
class Handler(BaseHTTPRequestHandler):
server_version = "identity/1"
def log_message(self, format, *args): # noqa: A002
LOG.info("%s - %s", self.address_string(), format % args)
def _authorized(self) -> bool:
if not TOKEN:
return False
return self.headers.get("Authorization", "") == f"Bearer {TOKEN}"
def _respond(self, status: HTTPStatus, payload) -> None:
body = json.dumps(payload).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(body)
def _read_body(self, max_bytes: int) -> bytes:
try:
length = int(self.headers.get("Content-Length") or 0)
except ValueError:
raise ValueError("missing or invalid Content-Length") from None
if length <= 0:
return b""
if length > max_bytes:
raise ValueError(f"body too large ({length} > {max_bytes} bytes)")
return self.rfile.read(length)
def do_OPTIONS(self): # noqa: N802
self.send_response(HTTPStatus.NO_CONTENT)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type")
self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
self.end_headers()
def do_GET(self): # noqa: N802
if not self._authorized():
self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"})
return
split = urlsplit(self.path)
path = split.path
query = parse_qs(split.query)
def q(key: str, default: str = "") -> str:
return (query.get(key) or [default])[0]
def q_int(key: str, default: int) -> int:
try:
return int(q(key, str(default)))
except ValueError:
return default
photo_match = re.match(r"^/people/(\d+)/photo$", path)
visits_match = re.match(r"^/people/(\d+)/visits$", path)
assignments_match = re.match(r"^/people/(\d+)/chore-assignments$", path)
if path == "/people":
self._respond(HTTPStatus.OK, {"people": list_people()})
elif path == "/presence":
self._respond(HTTPStatus.OK, presence())
elif path == "/resolve":
self._respond(HTTPStatus.OK, resolve_name(q("q") or q("name")))
elif path == "/visits":
self._respond(HTTPStatus.OK, {"visits": list_visits(since=q("since") or None, limit=q_int("limit", 200))})
elif visits_match:
self._respond(
HTTPStatus.OK,
{"visits": list_visits(person_id=int(visits_match.group(1)), since=q("since") or None,
limit=q_int("limit", 200))},
)
elif path == "/co-presence":
person_id = q_int("person_id", 0) or None
self._respond(HTTPStatus.OK, co_presence(person_id, q("since") or None, q_int("limit", 200)))
elif path == "/prune/candidates":
self._handle_prune_candidates(q("last_visit_before"))
elif path == "/device-access":
self._handle_device_access(q_int("person_id", 0), q("entity_id"), q("permission", "operate"), q("via") or None)
elif path == "/device-access/events":
self._respond(HTTPStatus.OK, {"events": list_device_access_events(q_int("limit", 100))})
elif path == "/chore-assignments":
self._respond(HTTPStatus.OK, chore_assignments())
elif assignments_match:
people = [p for p in list_people() if p["id"] == int(assignments_match.group(1))]
if not people:
self._respond(HTTPStatus.NOT_FOUND, {"error": "no such person"})
else:
self._respond(HTTPStatus.OK, {"chore_assignments": people[0]["chore_assignments"]})
elif path == "/weather":
with _weather_lock:
self._respond(HTTPStatus.OK, dict(_last_weather))
elif photo_match:
self._handle_get_photo(int(photo_match.group(1)))
else:
self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"})
def _handle_prune_candidates(self, last_visit_before: str) -> None:
if not last_visit_before:
self._respond(
HTTPStatus.BAD_REQUEST,
{"error": "'last_visit_before' (an ISO-8601 date or timestamp) is required"},
)
return
# A bare date ("2026-01-01") is what a browser's <input type="date"> hands over,
# and comparing it against a full timestamp string would silently include
# everyone who last visited ON that date. Widening it to the end of the day
# keeps "before 2026-01-01" meaning what a person reading it expects.
if re.match(r"^\d{4}-\d{2}-\d{2}$", last_visit_before):
last_visit_before = f"{last_visit_before}T00:00:00Z"
self._respond(HTTPStatus.OK, prune_candidates(last_visit_before))
def _handle_device_access(self, person_id: int, entity_id: str, permission: str, via: str | None) -> None:
if not person_id or not entity_id:
self._respond(
HTTPStatus.BAD_REQUEST,
{"allowed": False, "error": "'person_id' and 'entity_id' are required"},
)
return
self._respond(HTTPStatus.OK, check_device_access(person_id, entity_id, permission, via))
def _handle_get_photo(self, person_id: int) -> None:
data = get_person_photo(person_id)
if data is None:
self._respond(HTTPStatus.NOT_FOUND, {"error": "no photo for this person"})
return
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "image/jpeg")
self.send_header("Content-Length", str(len(data)))
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Cache-Control", "no-cache")
self.end_headers()
self.wfile.write(data)
def do_POST(self): # noqa: N802
if not self._authorized():
self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"})
return
path = urlsplit(self.path).path
chore_settings_match = re.match(r"^/people/(\d+)/chore-settings$", path)
assignments_match = re.match(r"^/people/(\d+)/chore-assignments$", path)
grants_match = re.match(r"^/people/(\d+)/device-grants$", path)
identifiers_match = re.match(r"^/people/(\d+)/identifiers$", path)
person_match = re.match(r"^/people/(\d+)$", path)
# /people/prune is checked before the bare /people/<id> edit route so it is
# never parsed as a person id (it can't be — it's not digits — but the ordering
# makes the intent explicit rather than incidental).
if path == "/register/photo":
self._handle_register_photo()
elif path == "/register":
self._handle_register()
elif path == "/register/guest":
self._handle_register_guest()
elif path == "/presence/manual":
self._handle_presence_manual()
elif path == "/people/prune":
self._handle_prune()
elif chore_settings_match:
self._handle_chore_settings(int(chore_settings_match.group(1)))
elif assignments_match:
self._handle_set_assignments(int(assignments_match.group(1)))
elif grants_match:
self._handle_grant_device(int(grants_match.group(1)))
elif identifiers_match:
self._handle_add_identifier(int(identifiers_match.group(1)))
elif person_match:
self._handle_update_person(int(person_match.group(1)))
else:
self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"})
def _json_body(self) -> dict | None:
"""Returns the parsed body, or None after having already sent a 400 — callers
check for None and return immediately. Replaces the identical try/except that
every handler below used to repeat.
"""
try:
raw = self._read_body(MAX_JSON_BYTES)
payload = json.loads(raw or b"{}")
except (ValueError, json.JSONDecodeError) as exc:
self._respond(HTTPStatus.BAD_REQUEST, {"error": f"bad request body: {exc}"})
return None
if not isinstance(payload, dict):
self._respond(HTTPStatus.BAD_REQUEST, {"error": "body must be a JSON object"})
return None
return payload
def _handle_update_person(self, person_id: int) -> None:
payload = self._json_body()
if payload is None:
return
result = update_person(person_id, payload)
if result.get("ok"):
self._respond(HTTPStatus.OK, result)
else:
self._respond(
HTTPStatus.NOT_FOUND if result.get("reason") == "not_found" else HTTPStatus.CONFLICT,
result,
)
def _handle_add_identifier(self, person_id: int) -> None:
payload = self._json_body()
if payload is None:
return
result = add_identifier(person_id, str(payload.get("entity_id", "")))
self._respond(HTTPStatus.OK if result.get("ok") else HTTPStatus.CONFLICT, result)
def _handle_set_assignments(self, person_id: int) -> None:
payload = self._json_body()
if payload is None:
return
types = payload.get("chore_types")
if not isinstance(types, list):
self._respond(HTTPStatus.BAD_REQUEST, {"error": "'chore_types' must be a list of strings"})
return
result = set_chore_assignments(person_id, types)
self._respond(HTTPStatus.OK if result.get("ok") else HTTPStatus.NOT_FOUND, result)
def _handle_grant_device(self, person_id: int) -> None:
payload = self._json_body()
if payload is None:
return
result = grant_device(
person_id,
str(payload.get("entity_id", "")),
str(payload.get("permission", "operate")),
payload.get("expires_at"),
payload.get("note"),
)
self._respond(
HTTPStatus.OK if result.get("ok")
else (HTTPStatus.NOT_FOUND if result.get("reason") == "not_found" else HTTPStatus.BAD_REQUEST),
result,
)
def _handle_prune(self) -> None:
payload = self._json_body()
if payload is None:
return
ids = payload.get("person_ids")
# Explicit ids only, never a filter — see prune_people()'s docstring for why
# this endpoint deliberately can't be handed the same query /prune/candidates
# takes.
if not isinstance(ids, list) or not all(isinstance(i, int) for i in ids):
self._respond(HTTPStatus.BAD_REQUEST, {"error": "'person_ids' must be a list of integers"})
return
if not ids:
self._respond(HTTPStatus.BAD_REQUEST, {"error": "'person_ids' is empty — nothing to prune"})
return
self._respond(HTTPStatus.OK, prune_people(ids))
def do_DELETE(self): # noqa: N802
if not self._authorized():
self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"})
return
path = urlsplit(self.path).path
id_match = re.match(r"^/people/(\d+)/identifiers/(\d+)$", path)
if id_match:
ok = delete_identifier(int(id_match.group(1)), int(id_match.group(2)))
self._respond(
HTTPStatus.OK if ok else HTTPStatus.NOT_FOUND,
{"ok": True} if ok else {"error": "no such identifier"},
)
return
grant_match = re.match(r"^/people/(\d+)/device-grants/(\d+)$", path)
if grant_match:
ok = revoke_device_grant(int(grant_match.group(1)), int(grant_match.group(2)))
self._respond(
HTTPStatus.OK if ok else HTTPStatus.NOT_FOUND,
{"ok": True} if ok else {"error": "no such grant"},
)
return
person_match = re.match(r"^/people/(\d+)$", path)
if person_match:
ok = delete_person(int(person_match.group(1)))
self._respond(
HTTPStatus.OK if ok else HTTPStatus.NOT_FOUND,
{"ok": True} if ok else {"error": "no such person"},
)
return
self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"})
def _handle_register_photo(self) -> None:
try:
data = self._read_body(MAX_IMAGE_BYTES)
except ValueError as exc:
self._respond(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": str(exc)})
return
if not data:
self._respond(HTTPStatus.BAD_REQUEST, {"error": "empty image body"})
return
photo_id = f"{int(time.time())}{os.getpid() % 10000:04d}"
PHOTO_DIR.mkdir(parents=True, exist_ok=True)
photo_path = PHOTO_DIR / f"{photo_id}.jpg"
photo_path.write_bytes(data)
LOG.info("identity: stored registration photo %s (%d bytes)", photo_id, len(data))
self._respond(HTTPStatus.OK, {"photo_id": photo_id})
def _handle_register(self) -> None:
payload = self._json_body()
if payload is None:
return
name = str(payload.get("name", ""))
device_id = str(payload.get("device_id", "unknown"))
photo_id = payload.get("photo_id")
forced_entity_id = payload.get("entity_id")
no_device = bool(payload.get("no_device", False))
ok, photo_path = self._resolve_photo_path(photo_id)
if not ok:
return # error response already sent
result = register(name, device_id, photo_path, forced_entity_id, no_device)
self._respond(HTTPStatus.OK if result.get("ok") else HTTPStatus.CONFLICT, result)
def _resolve_photo_path(self, photo_id) -> tuple[bool, str | None]:
"""Shared by /register and /register/guest. `(True, None)` if no photo_id
was given, `(True, path)` if it resolves to a real stored photo, `(False,
None)` if photo_id was present but invalid — callers must check the first
element and return early on False, since the error response is already sent
by the time this returns it.
"""
if not photo_id:
return True, None
if not PHOTO_ID_RE.match(str(photo_id)):
self._respond(HTTPStatus.BAD_REQUEST, {"error": "invalid photo_id"})
return False, None
candidate_path = PHOTO_DIR / f"{photo_id}.jpg"
return True, (str(candidate_path) if candidate_path.exists() else None)
def _handle_register_guest(self) -> None:
payload = self._json_body()
if payload is None:
return
device_id = str(payload.get("device_id", "unknown"))
ok, photo_path = self._resolve_photo_path(payload.get("photo_id"))
if not ok:
return
result = register_guest(device_id, photo_path)
self._respond(HTTPStatus.OK, result)
def _handle_presence_manual(self) -> None:
payload = self._json_body()
if payload is None:
return
try:
person_id = int(payload.get("person_id"))
home = bool(payload.get("home"))
except (TypeError, ValueError):
self._respond(HTTPStatus.BAD_REQUEST, {"error": "'person_id' (int) and 'home' (bool) are required"})
return
ok = set_manual_presence(person_id, home)
self._respond(HTTPStatus.OK if ok else HTTPStatus.NOT_FOUND, {"ok": ok})
def _handle_chore_settings(self, person_id: int) -> None:
payload = self._json_body()
if payload is None:
return
chore_exempt = payload.get("chore_exempt")
if chore_exempt is not None:
chore_exempt = bool(chore_exempt)
reminder_style = payload.get("chore_reminder_style")
if reminder_style is not None and not isinstance(reminder_style, str):
self._respond(HTTPStatus.BAD_REQUEST, {"error": "'chore_reminder_style' must be a string"})
return
ok = set_chore_settings(person_id, chore_exempt, reminder_style)
self._respond(HTTPStatus.OK if ok else HTTPStatus.NOT_FOUND, {"ok": ok})
def main() -> int:
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
)
if not TOKEN:
LOG.error("IDENTITY_TOKEN is not set — every request will be rejected until it is.")
if not HA_TOKEN:
LOG.warning("HA_TOKEN is not set — /register and /presence will fail until it is configured.")
init_db()
start_mqtt(
os.environ.get("MQTT_BROKER_HOST", ""),
int(os.environ.get("MQTT_BROKER_PORT") or 1883),
os.environ.get("MQTT_USERNAME", ""),
os.environ.get("MQTT_PASSWORD", ""),
)
# The visit-history sampler — see _presence_sampler_loop(). Daemon, so a Ctrl-C /
# container stop doesn't wait out a poll interval before exiting.
threading.Thread(target=_presence_sampler_loop, name="presence-sampler", daemon=True).start()
LOG.info(
"identity: sampling presence every %ds (departure grace %ds)",
PRESENCE_POLL_SECONDS, DEPARTURE_GRACE_SECONDS,
)
port = int(os.environ.get("IDENTITY_PORT", "8097"))
server = ThreadingHTTPServer(("0.0.0.0", port), Handler)
LOG.info("identity listening on :%d (HA: %s, db: %s)", port, HA_URL, DB_PATH)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink