162 lines
5.1 KiB
Python
162 lines
5.1 KiB
Python
"""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 . import hostutil, permutil, tokens
|
|
|
|
|
|
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": tokens.encode_session_key(entry.key),
|
|
"credential_uuid": str(entry.credential_uuid),
|
|
"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},
|
|
}
|