Refactor lengthy user info formatting to its own utility module that doesn't depend on FastAPI.

2025-12-02 14:30:31 +00:00
parent a62e8ddf1e
commit d4f8e97469
2 changed files with 179 additions and 161 deletions
@@ -0,0 +1,161 @@
+"""User information formatting and retrieval logic."""
+
+from datetime import timezone
+
+from passkey.util import useragent
+
+from .. import aaguid
+from ..authsession import session_key
+from ..globals import db
+from ..util import hostutil, permutil
+from ..util.tokens import encode_session_key
+
+
+def _format_datetime(dt):
+    """Format a datetime object to ISO 8601 string with UTC timezone."""
+    if dt is None:
+        return None
+    if dt.tzinfo:
+        return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+    else:
+        return dt.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+async def format_user_info(
+    *,
+    user_uuid,
+    auth: str,
+    session_record,
+    request_host: str | None,
+) -> dict:
+    """Format complete user information for authenticated users.
+
+    Args:
+        user_uuid: UUID of the user to fetch information for
+        auth: Authentication token
+        session_record: Current session record
+        request_host: Host header from the request
+
+    Returns:
+        Dictionary containing formatted user information including:
+        - User details
+        - Organization and role information
+        - Credentials list
+        - Sessions list
+        - Permissions
+    """
+    u = await db.instance.get_user_by_uuid(user_uuid)
+    ctx = await permutil.session_context(auth, request_host)
+
+    # Fetch and format credentials
+    credential_ids = await db.instance.get_credentials_by_user_uuid(user_uuid)
+    credentials: list[dict] = []
+    user_aaguids: set[str] = set()
+
+    for cred_id in credential_ids:
+        try:
+            c = await db.instance.get_credential_by_id(cred_id)
+        except ValueError:
+            continue
+
+        aaguid_str = str(c.aaguid)
+        user_aaguids.add(aaguid_str)
+        credentials.append(
+            {
+                "credential_uuid": str(c.uuid),
+                "aaguid": aaguid_str,
+                "created_at": _format_datetime(c.created_at),
+                "last_used": _format_datetime(c.last_used),
+                "last_verified": _format_datetime(c.last_verified),
+                "sign_count": c.sign_count,
+                "is_current_session": session_record.credential_uuid == c.uuid,
+            }
+        )
+
+    credentials.sort(key=lambda cred: cred["created_at"])
+    aaguid_info = aaguid.filter(user_aaguids)
+
+    # Format role and org information
+    role_info = None
+    org_info = None
+    effective_permissions: list[str] = []
+    is_global_admin = False
+    is_org_admin = False
+
+    if ctx:
+        role_info = {
+            "uuid": str(ctx.role.uuid),
+            "display_name": ctx.role.display_name,
+            "permissions": ctx.role.permissions,
+        }
+        org_info = {
+            "uuid": str(ctx.org.uuid),
+            "display_name": ctx.org.display_name,
+            "permissions": ctx.org.permissions,
+        }
+        effective_permissions = [p.id for p in (ctx.permissions or [])]
+        is_global_admin = "auth:admin" in (role_info["permissions"] or [])
+        is_org_admin = any(
+            p.startswith("auth:org:") for p in (role_info["permissions"] or [])
+        )
+
+    # Format sessions
+    normalized_request_host = hostutil.normalize_host(request_host)
+    session_records = await db.instance.list_sessions_for_user(user_uuid)
+    current_session_key = session_key(auth)
+    sessions_payload: list[dict] = []
+
+    for entry in session_records:
+        sessions_payload.append(
+            {
+                "id": encode_session_key(entry.key),
+                "host": entry.host,
+                "ip": entry.ip,
+                "user_agent": useragent.compact_user_agent(entry.user_agent),
+                "last_renewed": _format_datetime(entry.renewed),
+                "is_current": entry.key == current_session_key,
+                "is_current_host": bool(
+                    normalized_request_host
+                    and entry.host
+                    and entry.host == normalized_request_host
+                ),
+            }
+        )
+
+    return {
+        "authenticated": True,
+        "user": {
+            "user_uuid": str(u.uuid),
+            "user_name": u.display_name,
+            "created_at": _format_datetime(u.created_at),
+            "last_seen": _format_datetime(u.last_seen),
+            "visits": u.visits,
+        },
+        "org": org_info,
+        "role": role_info,
+        "permissions": effective_permissions,
+        "is_global_admin": is_global_admin,
+        "is_org_admin": is_org_admin,
+        "credentials": credentials,
+        "aaguid_info": aaguid_info,
+        "sessions": sessions_payload,
+    }
+
+
+async def format_reset_user_info(user_uuid, reset_token) -> dict:
+    """Format minimal user information for reset token requests.
+
+    Args:
+        user_uuid: UUID of the user
+        reset_token: Reset token record
+
+    Returns:
+        Dictionary with minimal user info for password reset flow
+    """
+    u = await db.instance.get_user_by_uuid(user_uuid)
+
+    return {
+        "authenticated": False,
+        "session_type": reset_token.token_type,
+        "user": {"user_uuid": str(u.uuid), "user_name": u.display_name},
+    }