File size: 2,770 Bytes
921d377 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """
Auth helper for the interactive service.
Thin wrapper over HomePilot's existing user-resolution logic. Lives
here (instead of reusing ``main._scoped_user_or_none``) so that:
- this subpackage has no import-time dependency on ``main``
- tests can monkeypatch a single entry point
- future interactive-only auth rules (e.g. per-experience share
links, viewer JWTs) slot in without touching the rest of the app
Resolution order matches the rest of HomePilot:
1. ``Authorization: Bearer <JWT>`` → validate as user token
2. ``homepilot_session`` cookie → validate as user token
3. Single-user install fallback → default user
Never raises on a failed lookup; returns ``None`` when no user
could be resolved. Callers decide whether to 401 / fall back.
"""
from __future__ import annotations
from typing import Any, Dict, Optional
def resolve_viewer_user_id(
authorization: str = "",
homepilot_session: Optional[str] = None,
) -> Optional[str]:
"""Resolve the current viewer to a HomePilot ``users.id``.
Returns the id as a string, or ``None`` if no user could be
identified (legitimate on a zero-user install; router decides
what to do next).
"""
try:
from ..users import (
ensure_users_tables,
_validate_token,
get_or_create_default_user,
count_users,
)
ensure_users_tables()
token = ""
if authorization and authorization.lower().startswith("bearer "):
token = authorization.split(" ", 1)[1].strip()
if not token and homepilot_session:
token = homepilot_session.strip()
user = _validate_token(token) if token else None
if user and user.get("id"):
return str(user["id"])
# Single-user fallback — mirrors ``_scoped_user_or_none``.
if count_users() <= 1:
default_user = get_or_create_default_user()
uid = default_user.get("id") if default_user else None
return str(uid) if uid else None
except Exception:
# Users subsystem unreachable → treat as anonymous.
return None
return None
def resolve_viewer(
authorization: str = "",
homepilot_session: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Like ``resolve_viewer_user_id`` but returns the full user row.
Used where the router needs the username / display_name (e.g.
personalization rule evaluation).
"""
uid = resolve_viewer_user_id(authorization, homepilot_session)
if not uid:
return None
try:
from ..users import list_users
for u in list_users():
if u.get("id") == uid:
return dict(u)
except Exception:
pass
return None
|