passkey/fastapi/api.py

@@ -30,67 +30,78 @@ def register_api_routes(app: FastAPI):
     @app.post("/auth/validate")
     async def validate_token(response: Response, auth=Cookie(None)):
         """Lightweight token validation endpoint."""
-        s = await get_session(auth)
+        try:
-        return {
+            s = await get_session(auth)
-            "valid": True,
+            return {
-            "user_uuid": str(s.user_uuid),
+                "valid": True,
-        }
+                "user_uuid": str(s.user_uuid),
+            }
+        except ValueError:
+            response.status_code = 401
+            return {"valid": False}
 
     @app.post("/auth/user-info")
     async def api_user_info(response: Response, auth=Cookie(None)):
         """Get full user information for the authenticated user."""
-        reset = passphrase.is_well_formed(auth)
+        try:
-        s = await (get_reset if reset else get_session)(auth)
+            reset = passphrase.is_well_formed(auth)
-        u = await db.instance.get_user_by_uuid(s.user_uuid)
+            s = await (get_reset if reset else get_session)(auth)
-        # Get all credentials for the user
+            u = await db.instance.get_user_by_uuid(s.user_uuid)
-        credential_ids = await db.instance.get_credentials_by_user_uuid(s.user_uuid)
+            # Get all credentials for the user
+            credential_ids = await db.instance.get_credentials_by_user_uuid(s.user_uuid)
 
-        credentials = []
+            credentials = []
-        user_aaguids = set()
+            user_aaguids = set()
 
-        for cred_id in credential_ids:
+            for cred_id in credential_ids:
-            c = await db.instance.get_credential_by_id(cred_id)
+                c = await db.instance.get_credential_by_id(cred_id)
 
-            # Convert AAGUID to string format
+                # Convert AAGUID to string format
-            aaguid_str = str(c.aaguid)
+                aaguid_str = str(c.aaguid)
-            user_aaguids.add(aaguid_str)
+                user_aaguids.add(aaguid_str)
 
-            # Check if this is the current session credential
+                # Check if this is the current session credential
-            is_current_session = s.credential_uuid == c.uuid
+                is_current_session = s.credential_uuid == c.uuid
 
-            credentials.append(
+                credentials.append(
-                {
+                    {
-                    "credential_uuid": str(c.uuid),
+                        "credential_uuid": str(c.uuid),
-                    "aaguid": aaguid_str,
+                        "aaguid": aaguid_str,
-                    "created_at": c.created_at.isoformat(),
+                        "created_at": c.created_at.isoformat(),
-                    "last_used": c.last_used.isoformat() if c.last_used else None,
+                        "last_used": c.last_used.isoformat() if c.last_used else None,
-                    "last_verified": c.last_verified.isoformat()
+                        "last_verified": c.last_verified.isoformat()
-                    if c.last_verified
+                        if c.last_verified
-                    else None,
+                        else None,
-                    "sign_count": c.sign_count,
+                        "sign_count": c.sign_count,
-                    "is_current_session": is_current_session,
+                        "is_current_session": is_current_session,
-                }
+                    }
-            )
+                )
 
-        # Get AAGUID information for only the AAGUIDs that the user has
+            # Get AAGUID information for only the AAGUIDs that the user has
-        aaguid_info = aaguid.filter(user_aaguids)
+            aaguid_info = aaguid.filter(user_aaguids)
 
-        # Sort credentials by creation date (earliest first, most recently created last)
+            # Sort credentials by creation date (earliest first, most recently created last)
-        credentials.sort(key=lambda cred: cred["created_at"])
+            credentials.sort(key=lambda cred: cred["created_at"])
 
-        return {
+            return {
-            "authenticated": not reset,
+                "authenticated": not reset,
-            "session_type": s.info["type"],
+                "session_type": s.info["type"],
-            "user": {
+                "user": {
-                "user_uuid": str(u.uuid),
+                    "user_uuid": str(u.uuid),
-                "user_name": u.display_name,
+                    "user_name": u.display_name,
-                "created_at": u.created_at.isoformat() if u.created_at else None,
+                    "created_at": u.created_at.isoformat() if u.created_at else None,
-                "last_seen": u.last_seen.isoformat() if u.last_seen else None,
+                    "last_seen": u.last_seen.isoformat() if u.last_seen else None,
-                "visits": u.visits,
+                    "visits": u.visits,
-            },
+                },
-            "credentials": credentials,
+                "credentials": credentials,
-            "aaguid_info": aaguid_info,
+                "aaguid_info": aaguid_info,
-        }
+            }
+        except ValueError as e:
+            response.status_code = 400
+            return {"detail": f"Failed to get user info: {e}"}
+        except Exception:
+            response.status_code = 500
+            return {"detail": "Failed to get user info"}
 
     @app.post("/auth/logout")
     async def api_logout(response: Response, auth=Cookie(None)):
@@ -108,20 +119,36 @@ def register_api_routes(app: FastAPI):
     @app.post("/auth/set-session")
     async def api_set_session(response: Response, auth=Depends(bearer_auth)):
         """Set session cookie from Authorization header. Fetched after login by WebSocket."""
-        user = await get_session(auth.credentials)
+        try:
-        if not user:
+            user = await get_session(auth.credentials)
-            raise ValueError("Invalid Authorization header.")
+            if not user:
-        session.set_session_cookie(response, auth.credentials)
+                raise ValueError("Invalid Authorization header.")
+            session.set_session_cookie(response, auth.credentials)
 
-        return {
+            return {
-            "message": "Session cookie set successfully",
+                "message": "Session cookie set successfully",
-            "user_uuid": str(user.user_uuid),
+                "user_uuid": str(user.user_uuid),
-        }
+            }
+
+        except ValueError as e:
+            response.status_code = 400
+            return {"detail": str(e)}
+        except Exception:
+            response.status_code = 500
+            return {"detail": "Failed to set session"}
 
     @app.delete("/auth/credential/{uuid}")
     async def api_delete_credential(
         response: Response, uuid: UUID, auth: str = Cookie(None)
     ):
         """Delete a specific credential for the current user."""
-        await delete_credential(uuid, auth)
+        try:
-        return {"message": "Credential deleted successfully"}
+            await delete_credential(uuid, auth)
+            return {"message": "Credential deleted successfully"}
+
+        except ValueError as e:
+            response.status_code = 400
+            return {"detail": str(e)}
+        except Exception:
+            response.status_code = 500
+            return {"detail": "Failed to delete credential"}

passkey/fastapi/mainapp.py

@@ -1,10 +1,9 @@
 import contextlib
-import logging
 from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import Cookie, FastAPI, Request, Response
-from fastapi.responses import FileResponse, JSONResponse
+from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 
 from ..authsession import get_session
@@ -31,21 +30,6 @@ async def lifespan(app: FastAPI):
 
 app = FastAPI(lifespan=lifespan)
 
-
-# Global exception handlers
-@app.exception_handler(ValueError)
-async def value_error_handler(request: Request, exc: ValueError):
-    """Handle ValueError exceptions globally with 400 status code."""
-    return JSONResponse(status_code=400, content={"detail": str(exc)})
-
-
-@app.exception_handler(Exception)
-async def general_exception_handler(request: Request, exc: Exception):
-    """Handle all other exceptions globally with 500 status code."""
-    logging.exception("Internal Server Error")
-    return JSONResponse(status_code=500, content={"detail": "Internal server error"})
-
-
 # Mount the WebSocket subapp
 app.mount("/auth/ws", ws.app)
 

passkey/fastapi/reset.py

@@ -15,27 +15,35 @@ def register_reset_routes(app):
     @app.post("/auth/create-link")
     async def api_create_link(request: Request, response: Response, auth=Cookie(None)):
         """Create a device addition link for the authenticated user."""
-        # Require authentication
+        try:
-        s = await get_session(auth)
+            # Require authentication
+            s = await get_session(auth)
 
-        # Generate a human-readable token
+            # Generate a human-readable token
-        token = passphrase.generate()  # e.g., "cross.rotate.yin.note.evoke"
+            token = passphrase.generate()  # e.g., "cross.rotate.yin.note.evoke"
-        await db.instance.create_session(
+            await db.instance.create_session(
-            user_uuid=s.user_uuid,
+                user_uuid=s.user_uuid,
-            key=tokens.reset_key(token),
+                key=tokens.reset_key(token),
-            expires=expires(),
+                expires=expires(),
-            info=session.infodict(request, "device addition"),
+                info=session.infodict(request, "device addition"),
-        )
+            )
 
-        # Generate the device addition link with pretty URL
+            # Generate the device addition link with pretty URL
-        path = request.url.path.removesuffix("create-link") + token
+            path = request.url.path.removesuffix("create-link") + token
-        url = f"{request.headers['origin']}{path}"
+            url = f"{request.headers['origin']}{path}"
 
-        return {
+            return {
-            "message": "Registration link generated successfully",
+                "message": "Registration link generated successfully",
-            "url": url,
+                "url": url,
-            "expires": expires().isoformat(),
+                "expires": expires().isoformat(),
-        }
+            }
+
+        except ValueError:
+            response.status_code = 401
+            return {"detail": "Authentication required"}
+        except Exception as e:
+            response.status_code = 500
+            return {"detail": f"Failed to create registration link: {str(e)}"}
 
     @app.get("/auth/{reset_token}")
     async def reset_authentication(

passkey/fastapi/ws.py

@@ -1,6 +1,15 @@
+"""
+WebSocket handlers for passkey authentication operations.
+
+This module contains all WebSocket endpoints for:
+- User registration
+- Adding credentials to existing users
+- Device credential addition via token
+- Authentication
+"""
+
 import logging
 from datetime import datetime
-from functools import wraps
 from uuid import UUID
 
 import uuid7
@@ -14,25 +23,6 @@ from ..util import passphrase
 from ..util.tokens import create_token, session_key
 from .session import infodict
 
-
-# WebSocket error handling decorator
-def websocket_error_handler(func):
-    @wraps(func)
-    async def wrapper(ws: WebSocket, *args, **kwargs):
-        try:
-            await ws.accept()
-            return await func(ws, *args, **kwargs)
-        except WebSocketDisconnect:
-            pass
-        except (ValueError, InvalidAuthenticationResponse) as e:
-            await ws.send_json({"detail": str(e)})
-        except Exception:
-            logging.exception("Internal Server Error")
-            await ws.send_json({"detail": "Internal Server Error"})
-
-    return wrapper
-
-
 # Create a FastAPI subapp for WebSocket endpoints
 app = FastAPI()
 
@@ -63,104 +53,131 @@ async def register_chat(
 
 
 @app.websocket("/register")
-@websocket_error_handler
 async def websocket_register_new(
     ws: WebSocket, user_name: str = Query(""), auth=Cookie(None)
 ):
     """Register a new user and with a new passkey credential."""
-    origin = ws.headers["origin"]
+    await ws.accept()
-    user_uuid = uuid7.create()
+    origin = ws.headers.get("origin")
-    # WebAuthn registration
+    try:
-    credential = await register_chat(ws, user_uuid, user_name, origin=origin)
+        user_uuid = uuid7.create()
+        # WebAuthn registration
+        credential = await register_chat(ws, user_uuid, user_name, origin=origin)
 
-    # Store the user and credential in the database
+        # Store the user and credential in the database
-    await db.instance.create_user_and_credential(
+        await db.instance.create_user_and_credential(
-        User(user_uuid, user_name, created_at=datetime.now()),
+            User(user_uuid, user_name, created_at=datetime.now()),
-        credential,
+            credential,
-    )
+        )
-    # Create a session token for the new user
+        # Create a session token for the new user
-    token = create_token()
+        token = create_token()
-    await db.instance.create_session(
+        await db.instance.create_session(
-        user_uuid=user_uuid,
+            user_uuid=user_uuid,
-        key=session_key(token),
+            key=session_key(token),
-        expires=datetime.now() + EXPIRES,
+            expires=datetime.now() + EXPIRES,
-        info=infodict(ws, "authenticated"),
+            info=infodict(ws, "authenticated"),
-        credential_uuid=credential.uuid,
+            credential_uuid=credential.uuid,
-    )
+        )
 
-    await ws.send_json(
+        await ws.send_json(
-        {
+            {
-            "user_uuid": str(user_uuid),
+                "user_uuid": str(user_uuid),
-            "session_token": token,
+                "session_token": token,
-        }
+            }
-    )
+        )
+    except ValueError as e:
+        await ws.send_json({"detail": str(e)})
+    except WebSocketDisconnect:
+        pass
+    except Exception:
+        logging.exception("Internal Server Error")
+        await ws.send_json({"detail": "Internal Server Error"})
 
 
 @app.websocket("/add_credential")
-@websocket_error_handler
 async def websocket_register_add(ws: WebSocket, auth=Cookie(None)):
     """Register a new credential for an existing user."""
+    await ws.accept()
     origin = ws.headers["origin"]
-    # Try to get either a regular session or a reset session
+    try:
-    reset = passphrase.is_well_formed(auth)
+        # Try to get either a regular session or a reset session
-    s = await (get_reset if reset else get_session)(auth)
+        reset = passphrase.is_well_formed(auth)
-    user_uuid = s.user_uuid
+        s = await (get_reset if reset else get_session)(auth)
+        user_uuid = s.user_uuid
 
-    # Get user information to get the user_name
+        # Get user information to get the user_name
-    user = await db.instance.get_user_by_uuid(user_uuid)
+        user = await db.instance.get_user_by_uuid(user_uuid)
-    user_name = user.display_name
+        user_name = user.display_name
-    challenge_ids = await db.instance.get_credentials_by_user_uuid(user_uuid)
+        challenge_ids = await db.instance.get_credentials_by_user_uuid(user_uuid)
 
-    # WebAuthn registration
+        # WebAuthn registration
-    credential = await register_chat(ws, user_uuid, user_name, challenge_ids, origin)
+        credential = await register_chat(
-    if reset:
+            ws, user_uuid, user_name, challenge_ids, origin
-        # Replace reset session with a new session
-        await db.instance.delete_session(s.key)
-        token = await create_session(
-            user_uuid, credential.uuid, infodict(ws, "authenticated")
         )
-    else:
+        if reset:
-        token = auth
+            # Replace reset session with a new session
-    assert isinstance(token, str) and len(token) == 16
+            await db.instance.delete_session(s.key)
-    # Store the new credential in the database
+            token = await create_session(
-    await db.instance.create_credential(credential)
+                user_uuid, credential.uuid, infodict(ws, "authenticated")
+            )
+        else:
+            token = auth
+        assert isinstance(token, str) and len(token) == 16
+        # Store the new credential in the database
+        await db.instance.create_credential(credential)
 
-    await ws.send_json(
+        await ws.send_json(
-        {
+            {
-            "user_uuid": str(user.uuid),
+                "user_uuid": str(user.uuid),
-            "credential_uuid": str(credential.uuid),
+                "credential_uuid": str(credential.uuid),
-            "session_token": token,
+                "session_token": token,
-            "message": "New credential added successfully",
+                "message": "New credential added successfully",
-        }
+            }
-    )
+        )
+    except ValueError as e:
+        await ws.send_json({"detail": str(e)})
+    except WebSocketDisconnect:
+        pass
+    except Exception:
+        logging.exception("Internal Server Error")
+        await ws.send_json({"detail": "Internal Server Error"})
 
 
 @app.websocket("/authenticate")
-@websocket_error_handler
 async def websocket_authenticate(ws: WebSocket):
+    await ws.accept()
     origin = ws.headers["origin"]
-    options, challenge = passkey.auth_generate_options()
+    try:
-    await ws.send_json(options)
+        options, challenge = passkey.auth_generate_options()
-    # Wait for the client to use his authenticator to authenticate
+        await ws.send_json(options)
-    credential = passkey.auth_parse(await ws.receive_json())
+        # Wait for the client to use his authenticator to authenticate
-    # Fetch from the database by credential ID
+        credential = passkey.auth_parse(await ws.receive_json())
-    stored_cred = await db.instance.get_credential_by_id(credential.raw_id)
+        # Fetch from the database by credential ID
-    # Verify the credential matches the stored data
+        stored_cred = await db.instance.get_credential_by_id(credential.raw_id)
-    passkey.auth_verify(credential, challenge, stored_cred, origin=origin)
+        # Verify the credential matches the stored data
-    # Update both credential and user's last_seen timestamp
+        passkey.auth_verify(credential, challenge, stored_cred, origin=origin)
-    await db.instance.login(stored_cred.user_uuid, stored_cred)
+        # Update both credential and user's last_seen timestamp
+        await db.instance.login(stored_cred.user_uuid, stored_cred)
 
-    # Create a session token for the authenticated user
+        # Create a session token for the authenticated user
-    assert stored_cred.uuid is not None
+        assert stored_cred.uuid is not None
-    token = await create_session(
+        token = await create_session(
-        user_uuid=stored_cred.user_uuid,
+            user_uuid=stored_cred.user_uuid,
-        info=infodict(ws, "auth"),
+            info=infodict(ws, "auth"),
-        credential_uuid=stored_cred.uuid,
+            credential_uuid=stored_cred.uuid,
-    )
+        )
 
-    await ws.send_json(
+        await ws.send_json(
-        {
+            {
-            "user_uuid": str(stored_cred.user_uuid),
+                "user_uuid": str(stored_cred.user_uuid),
-            "session_token": token,
+                "session_token": token,
-        }
+            }
-    )
+        )
+    except (ValueError, InvalidAuthenticationResponse) as e:
+        logging.exception("ValueError")
+        await ws.send_json({"detail": str(e)})
+    except WebSocketDisconnect:
+        pass
+    except Exception:
+        logging.exception("Internal Server Error")
+        await ws.send_json({"detail": "Internal Server Error"})