LeoVasanko/passkey-auth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Major refactor: HTTP-only cookies, passkey management, and UI improvements

Browse Source 
- Refactor session management from WebSocket tokens to HTTP-only cookies
- Move user/credential endpoints from WebSocket to HTTP REST API
- Add comprehensive passkey management (add/delete with safety checks)
- Implement AAGUID-based authenticator info with icons and names
- Add human-readable date formatting and clean grid layout
- Create modular architecture with session_manager, api_handlers, aaguid_manager
This commit is contained in:
Leo Vasanko
2025-07-06 19:45:33 -06:00
parent 9c2b7cf450
commit eb56c000e8
11 changed files with 1380 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
passkeyauth/aaguid_manager.py Normal file
View File

@@ -0,0 +1,65 @@
"""
AAGUID (Authenticator Attestation GUID) management for WebAuthn credentials.
This module provides functionality to:
- Load AAGUID data from JSON file
- Look up authenticator information by AAGUID
- Return only relevant AAGUID data for user credentials
"""
import json
from pathlib import Path
from typing import Optional
# Path to the AAGUID JSON file
AAGUID_FILE = Path(__file__).parent / "combined_aaguid.json"
class AAGUIDManager:
"""Manages AAGUID data and lookups."""
def __init__(self):
self.aaguid_data: dict[str, dict] = {}
self.load_aaguid_data()
def load_aaguid_data(self) -> None:
"""Load AAGUID data from the JSON file."""
try:
with open(AAGUID_FILE, encoding="utf-8") as f:
self.aaguid_data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"Warning: Could not load AAGUID data: {e}")
self.aaguid_data = {}
def get_authenticator_info(self, aaguid: str) -> Optional[dict]:
"""Get authenticator information for a specific AAGUID."""
return self.aaguid_data.get(aaguid)
def get_relevant_aaguids(self, aaguids: set[str]) -> dict[str, dict]:
"""
Get AAGUID information only for the provided set of AAGUIDs.
Args:
aaguids: Set of AAGUID strings that the user has credentials for
Returns:
Dictionary mapping AAGUID to authenticator information for only
the AAGUIDs that the user has and that we have data for
"""
relevant = {}
for aaguid in aaguids:
if aaguid in self.aaguid_data:
relevant[aaguid] = self.aaguid_data[aaguid]
return relevant
# Global AAGUID manager instance
_aaguid_manager: Optional[AAGUIDManager] = None
def get_aaguid_manager() -> AAGUIDManager:
"""Get the global AAGUID manager instance."""
global _aaguid_manager
if _aaguid_manager is None:
_aaguid_manager = AAGUIDManager()
return _aaguid_manager

247
passkeyauth/api_handlers.py Normal file
View File

@@ -0,0 +1,247 @@
"""
API endpoints for user management and session handling.
This module contains all the HTTP API endpoints for:
- User information retrieval
- User credentials management
- Session token validation and refresh
- Login/logout functionality
"""
from fastapi import Request, Response
from .aaguid_manager import get_aaguid_manager
from .db import connect
from .jwt_manager import refresh_session_token, validate_session_token
from .session_manager import (
clear_session_cookie,
get_current_user,
get_session_token_from_auth_header_or_body,
get_session_token_from_request,
set_session_cookie,
)
async def get_user_info(request: Request) -> dict:
"""Get user information from session cookie."""
try:
user = await get_current_user(request)
if not user:
return {"error": "Not authenticated"}
return {
"status": "success",
"user": {
"user_id": str(user.user_id),
"user_name": user.user_name,
"created_at": user.created_at.isoformat() if user.created_at else None,
"last_seen": user.last_seen.isoformat() if user.last_seen else None,
},
}
except Exception as e:
return {"error": f"Failed to get user info: {str(e)}"}
async def get_user_credentials(request: Request) -> dict:
"""Get all credentials for a user using session cookie."""
try:
user = await get_current_user(request)
if not user:
return {"error": "Not authenticated"}
# Get current session credential ID
current_credential_id = None
session_token = get_session_token_from_request(request)
if session_token:
token_data = validate_session_token(session_token)
if token_data:
current_credential_id = token_data.get("credential_id")
async with connect() as db:
# Get all credentials for the user
credential_ids = await db.get_credentials_by_user_id(user.user_id.bytes)
credentials = []
user_aaguids = set()
for cred_id in credential_ids:
try:
stored_cred = await db.get_credential_by_id(cred_id)
# Convert AAGUID to string format
aaguid_str = str(stored_cred.aaguid)
user_aaguids.add(aaguid_str)
# Check if this is the current session credential
is_current_session = (
current_credential_id == stored_cred.credential_id
)
credentials.append(
{
"credential_id": stored_cred.credential_id.hex(),
"aaguid": aaguid_str,
"created_at": stored_cred.created_at.isoformat(),
"last_used": stored_cred.last_used.isoformat()
if stored_cred.last_used
else None,
"last_verified": stored_cred.last_verified.isoformat()
if stored_cred.last_verified
else None,
"sign_count": stored_cred.sign_count,
"is_current_session": is_current_session,
}
)
except ValueError:
# Skip invalid credentials
continue
# Get AAGUID information for only the AAGUIDs that the user has
aaguid_manager = get_aaguid_manager()
aaguid_info = aaguid_manager.get_relevant_aaguids(user_aaguids)
# Sort credentials by creation date (earliest first, most recently created last)
credentials.sort(key=lambda cred: cred["created_at"])
return {
"status": "success",
"credentials": credentials,
"aaguid_info": aaguid_info,
}
except Exception as e:
return {"error": f"Failed to get credentials: {str(e)}"}
async def refresh_token(request: Request, response: Response) -> dict:
"""Refresh the session token."""
try:
session_token = get_session_token_from_request(request)
if not session_token:
return {"error": "No session token found"}
# Validate and refresh the token
new_token = refresh_session_token(session_token)
if new_token:
set_session_cookie(response, new_token)
return {"status": "success", "refreshed": True}
else:
clear_session_cookie(response)
return {"error": "Invalid or expired session token"}
except Exception as e:
return {"error": f"Failed to refresh token: {str(e)}"}
async def validate_token(request: Request) -> dict:
"""Validate a session token and return user info."""
try:
session_token = get_session_token_from_request(request)
if not session_token:
return {"error": "No session token found"}
# Validate the session token
token_data = validate_session_token(session_token)
if not token_data:
return {"error": "Invalid or expired session token"}
return {
"status": "success",
"valid": True,
"user_id": str(token_data["user_id"]),
"credential_id": token_data["credential_id"].hex(),
"issued_at": token_data["issued_at"],
"expires_at": token_data["expires_at"],
}
except Exception as e:
return {"error": f"Failed to validate token: {str(e)}"}
async def logout(response: Response) -> dict:
"""Log out the current user by clearing the session cookie."""
clear_session_cookie(response)
return {"status": "success", "message": "Logged out successfully"}
async def set_session(request: Request, response: Response) -> dict:
"""Set session cookie using JWT token from request body or Authorization header."""
try:
session_token = await get_session_token_from_auth_header_or_body(request)
if not session_token:
return {"error": "No session token provided"}
# Validate the session token
token_data = validate_session_token(session_token)
if not token_data:
return {"error": "Invalid or expired session token"}
# Set the HTTP-only cookie
set_session_cookie(response, session_token)
return {
"status": "success",
"message": "Session cookie set successfully",
"user_id": str(token_data["user_id"]),
}
except Exception as e:
return {"error": f"Failed to set session: {str(e)}"}
async def delete_credential(request: Request) -> dict:
"""Delete a specific credential for the current user."""
try:
user = await get_current_user(request)
if not user:
return {"error": "Not authenticated"}
# Get the credential ID from the request body
try:
body = await request.json()
credential_id = body.get("credential_id")
if not credential_id:
return {"error": "credential_id is required"}
except Exception:
return {"error": "Invalid request body"}
# Convert credential_id from hex string to bytes
try:
credential_id_bytes = bytes.fromhex(credential_id)
except ValueError:
return {"error": "Invalid credential_id format"}
async with connect() as db:
# First, verify the credential belongs to the current user
try:
stored_cred = await db.get_credential_by_id(credential_id_bytes)
if stored_cred.user_id != user.user_id:
return {"error": "Credential not found or access denied"}
except ValueError:
return {"error": "Credential not found"}
# Check if this is the current session credential
session_token = get_session_token_from_request(request)
if session_token:
token_data = validate_session_token(session_token)
if (
token_data
and token_data.get("credential_id") == credential_id_bytes
):
return {"error": "Cannot delete current session credential"}
# Get user's remaining credentials count
remaining_credentials = await db.get_credentials_by_user_id(
user.user_id.bytes
)
if len(remaining_credentials) <= 1:
return {"error": "Cannot delete last remaining credential"}
# Delete the credential
await db.delete_credential(credential_id_bytes)
return {"status": "success", "message": "Credential deleted successfully"}
except Exception as e:
return {"error": f"Failed to delete credential: {str(e)}"}

1
passkeyauth/combined_aaguid.json Normal file
View File

File diff suppressed because one or more lines are too long

48
passkeyauth/db.py
View File

@@ -54,22 +54,11 @@ SQL_STORE_CREDENTIAL = """
"""
SQL_GET_CREDENTIAL_BY_ID = """
SELECT credential_id, user_id, aaguid, public_key, sign_count, created_at, last_used, last_verified
FROM credentials
WHERE credential_id = ?
SELECT * FROM credentials WHERE credential_id = ?
"""
SQL_GET_USER_CREDENTIALS = """
SELECT c.credential_id
FROM credentials c
JOIN users u ON c.user_id = u.user_id
WHERE u.user_id = ?
"""
SQL_UPDATE_CREDENTIAL_SIGN_COUNT = """
UPDATE credentials
SET sign_count = ?, last_used = CURRENT_TIMESTAMP
WHERE credential_id = ?
SELECT credential_id FROM credentials WHERE user_id = ?
"""
SQL_UPDATE_CREDENTIAL = """
@@ -78,11 +67,13 @@ SQL_UPDATE_CREDENTIAL = """
WHERE credential_id = ?
"""
SQL_DELETE_CREDENTIAL = """
DELETE FROM credentials WHERE credential_id = ?
"""
@dataclass
class User:
"""User data model."""
user_id: UUID
user_name: str
created_at: datetime | None = None
@@ -118,8 +109,8 @@ class DB:
return User(
user_id=UUID(bytes=row[0]),
user_name=row[1],
created_at=row[2],
last_seen=row[3],
created_at=_convert_datetime(row[2]),
last_seen=_convert_datetime(row[3]),
)
raise ValueError("User not found")
@@ -127,7 +118,12 @@ class DB:
"""Create a new user and return the User dataclass."""
await self.conn.execute(
SQL_CREATE_USER,
(user.user_id.bytes, user.user_name, user.created_at, user.last_seen),
(
user.user_id.bytes,
user.user_name,
user.created_at or datetime.now(),
user.last_seen,
),
)
async def create_credential(self, credential: StoredCredential) -> None:
@@ -159,9 +155,9 @@ class DB:
aaguid=UUID(bytes=row[2]),
public_key=row[3],
sign_count=row[4],
created_at=row[5],
last_used=row[6],
last_verified=row[7],
created_at=datetime.fromisoformat(row[5]),
last_used=_convert_datetime(row[6]),
last_verified=_convert_datetime(row[7]),
)
raise ValueError("Credential not registered")
@@ -192,3 +188,13 @@ class DB:
"UPDATE users SET last_seen = ? WHERE user_id = ?",
(credential.last_used, user_id),
)
async def delete_credential(self, credential_id: bytes) -> None:
"""Delete a credential by its ID."""
await self.conn.execute(SQL_DELETE_CREDENTIAL, (credential_id,))
await self.conn.commit()
def _convert_datetime(val):
"""Convert string from SQLite to datetime object (pass through None)."""
return val and datetime.fromisoformat(val)

132
passkeyauth/jwt_manager.py Normal file
View File

@@ -0,0 +1,132 @@
"""
JWT session management for WebAuthn authentication.
This module provides JWT token generation and validation for managing user sessions
after successful WebAuthn authentication. Tokens contain user ID and credential ID
for session validation.
"""
import secrets
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional
from uuid import UUID
import jwt
SECRET_FILE = Path("server-secret.bin")
def load_or_create_secret() -> bytes:
"""Load JWT secret from file or create a new one."""
if SECRET_FILE.exists():
return SECRET_FILE.read_bytes()
else:
# Generate a new 32-byte secret
secret = secrets.token_bytes(32)
SECRET_FILE.write_bytes(secret)
return secret
class JWTManager:
"""Manages JWT tokens for user sessions."""
def __init__(self, secret_key: bytes, algorithm: str = "HS256"):
self.secret_key = secret_key
self.algorithm = algorithm
self.token_expiry = timedelta(hours=24) # Tokens expire after 24 hours
def create_token(self, user_id: UUID, credential_id: bytes) -> str:
"""
Create a JWT token for a user session.
Args:
user_id: The user's UUID
credential_id: The credential ID used for authentication
Returns:
JWT token string
"""
now = datetime.utcnow()
payload = {
"user_id": str(user_id),
"credential_id": credential_id.hex(),
"iat": now,
"exp": now + self.token_expiry,
"iss": "passkeyauth",
}
return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
def validate_token(self, token: str) -> Optional[dict]:
"""
Validate a JWT token and return the payload.
Args:
token: JWT token string
Returns:
Dictionary with user_id and credential_id, or None if invalid
"""
try:
payload = jwt.decode(
token,
self.secret_key,
algorithms=[self.algorithm],
issuer="passkeyauth",
)
return {
"user_id": UUID(payload["user_id"]),
"credential_id": bytes.fromhex(payload["credential_id"]),
"issued_at": payload["iat"],
"expires_at": payload["exp"],
}
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
def refresh_token(self, token: str) -> Optional[str]:
"""
Refresh a JWT token if it's still valid.
Args:
token: Current JWT token
Returns:
New JWT token string, or None if the current token is invalid
"""
payload = self.validate_token(token)
if payload is None:
return None
return self.create_token(payload["user_id"], payload["credential_id"])
# Global JWT manager instance
_jwt_manager: Optional[JWTManager] = None
def get_jwt_manager() -> JWTManager:
"""Get the global JWT manager instance."""
global _jwt_manager
if _jwt_manager is None:
secret = load_or_create_secret()
_jwt_manager = JWTManager(secret)
return _jwt_manager
def create_session_token(user_id: UUID, credential_id: bytes) -> str:
"""Create a session token for a user."""
return get_jwt_manager().create_token(user_id, credential_id)
def validate_session_token(token: str) -> Optional[dict]:
"""Validate a session token."""
return get_jwt_manager().validate_token(token)
def refresh_session_token(token: str) -> Optional[str]:
"""Refresh a session token."""
return get_jwt_manager().refresh_token(token)

131
passkeyauth/main.py
View File

@@ -10,18 +10,31 @@ This module provides a simple WebAuthn implementation that:
"""
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from uuid import UUID, uuid4
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi import FastAPI, Request, Response, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from .api_handlers import (
delete_credential,
get_user_credentials,
get_user_info,
logout,
refresh_token,
set_session,
validate_token,
)
from .db import User, connect
from .jwt_manager import create_session_token
from .passkey import Passkey
from .session_manager import get_user_from_cookie_string
STATIC_DIR = Path(__file__).parent.parent / "static"
passkey = Passkey(
rp_id="localhost",
rp_name="Passkey Auth",
@@ -55,14 +68,66 @@ async def websocket_register_new(ws: WebSocket):
# Store the user in the database
async with connect() as db:
await db.conn.execute("BEGIN")
await db.create_user(User(user_id, user_name))
await db.create_user(User(user_id, user_name, created_at=datetime.now()))
await db.create_credential(credential)
await ws.send_json({"status": "success", "user_id": str(user_id)})
# Create a session token for the new user
session_token = create_session_token(user_id, credential.credential_id)
await ws.send_json(
{
"status": "success",
"user_id": str(user_id),
"session_token": session_token,
}
)
except ValueError as e:
await ws.send_json({"error": str(e)})
except WebSocketDisconnect:
pass
@app.websocket("/ws/add_credential")
async def websocket_register_add(ws: WebSocket):
"""Register a new credential for an existing user."""
await ws.accept()
try:
# Authenticate user via cookie
cookie_header = ws.headers.get("cookie", "")
user_id = await get_user_from_cookie_string(cookie_header)
if not user_id:
await ws.send_json({"error": "Authentication required"})
return
# Get user information to get the user_name
async with connect() as db:
user = await db.get_user_by_user_id(user_id.bytes)
user_name = user.user_name
# WebAuthn registration
credential = await register_chat(ws, user_id, user_name)
print(f"New credential for user {user_id}: {credential}")
# Store the new credential in the database
async with connect() as db:
await db.create_credential(credential)
await ws.send_json(
{
"status": "success",
"user_id": str(user_id),
"credential_id": credential.credential_id.hex(),
"message": "New credential added successfully",
}
)
except ValueError as e:
await ws.send_json({"error": str(e)})
except WebSocketDisconnect:
pass
except Exception as e:
await ws.send_json({"error": f"Server error: {str(e)}"})
async def register_chat(ws: WebSocket, user_id: UUID, user_name: str):
"""Generate registration options and send them to the client."""
options, challenge = passkey.reg_generate_options(
@@ -71,6 +136,7 @@ async def register_chat(ws: WebSocket, user_id: UUID, user_name: str):
)
await ws.send_json(options)
response = await ws.receive_json()
print(response)
return passkey.reg_verify(response, challenge, user_id)
@@ -87,14 +153,69 @@ async def websocket_authenticate(ws: WebSocket):
stored_cred = await db.get_credential_by_id(credential.raw_id)
# Verify the credential matches the stored data
await passkey.auth_verify(credential, challenge, stored_cred)
await db.update_credential(stored_cred)
await ws.send_json({"status": "success"})
# Update both credential and user's last_seen timestamp
await db.login(stored_cred.user_id.bytes, stored_cred)
# Create a session token for the authenticated user
session_token = create_session_token(
stored_cred.user_id, stored_cred.credential_id
)
await ws.send_json(
{
"status": "success",
"user_id": str(stored_cred.user_id),
"session_token": session_token,
}
)
except ValueError as e:
await ws.send_json({"error": str(e)})
except WebSocketDisconnect:
pass
@app.get("/api/user-info")
async def api_get_user_info(request: Request):
"""Get user information from session cookie."""
return await get_user_info(request)
@app.get("/api/user-credentials")
async def api_get_user_credentials(request: Request):
"""Get all credentials for a user using session cookie."""
return await get_user_credentials(request)
@app.post("/api/refresh-token")
async def api_refresh_token(request: Request, response: Response):
"""Refresh the session token."""
return await refresh_token(request, response)
@app.get("/api/validate-token")
async def api_validate_token(request: Request):
"""Validate a session token and return user info."""
return await validate_token(request)
@app.post("/api/logout")
async def api_logout(response: Response):
"""Log out the current user by clearing the session cookie."""
return await logout(response)
@app.post("/api/set-session")
async def api_set_session(request: Request, response: Response):
"""Set session cookie using JWT token from request body or Authorization header."""
return await set_session(request, response)
@app.post("/api/delete-credential")
async def api_delete_credential(request: Request):
"""Delete a specific credential for the current user."""
return await delete_credential(request)
# Serve static files
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")

2
passkeyauth/passkey.py
View File

@@ -28,6 +28,7 @@ from webauthn.helpers import (
)
from webauthn.helpers.cose import COSEAlgorithmIdentifier
from webauthn.helpers.structs import (
AttestationConveyancePreference,
AuthenticationCredential,
AuthenticatorSelectionCriteria,
PublicKeyCredentialDescriptor,
@@ -109,6 +110,7 @@ class Passkey:
rp_name=self.rp_name,
user_id=user_id.bytes,
user_name=user_name,
attestation=AttestationConveyancePreference.DIRECT,
authenticator_selection=AuthenticatorSelectionCriteria(
resident_key=ResidentKeyRequirement.REQUIRED,
user_verification=UserVerificationRequirement.PREFERRED,

107
passkeyauth/session_manager.py Normal file
View File

@@ -0,0 +1,107 @@
"""
Session management for WebAuthn authentication.
This module provides session management functionality including:
- Getting current user from session cookies
- Setting and clearing HTTP-only cookies
- Session validation and token handling
"""
from typing import Optional
from uuid import UUID
from fastapi import Request, Response
from .db import User, connect
from .jwt_manager import validate_session_token
COOKIE_NAME = "session_token"
COOKIE_MAX_AGE = 86400 # 24 hours
async def get_current_user(request: Request) -> Optional[User]:
"""Get the current user from the session cookie."""
session_token = request.cookies.get(COOKIE_NAME)
if not session_token:
return None
token_data = validate_session_token(session_token)
if not token_data:
return None
async with connect() as db:
try:
user = await db.get_user_by_user_id(token_data["user_id"].bytes)
return user
except Exception:
return None
def set_session_cookie(response: Response, session_token: str) -> None:
"""Set the session token as an HTTP-only cookie."""
response.set_cookie(
key=COOKIE_NAME,
value=session_token,
max_age=COOKIE_MAX_AGE,
httponly=True,
secure=False, # Set to True in production with HTTPS
samesite="lax",
)
def clear_session_cookie(response: Response) -> None:
"""Clear the session cookie."""
response.delete_cookie(key=COOKIE_NAME)
def get_session_token_from_request(request: Request) -> Optional[str]:
"""Extract session token from request cookies."""
return request.cookies.get(COOKIE_NAME)
async def validate_session_from_request(request: Request) -> Optional[dict]:
"""Validate session token from request and return token data."""
session_token = get_session_token_from_request(request)
if not session_token:
return None
return validate_session_token(session_token)
async def get_session_token_from_auth_header_or_body(request: Request) -> Optional[str]:
"""Extract session token from Authorization header or request body."""
# Try to get token from Authorization header first
auth_header = request.headers.get("Authorization")
if auth_header and auth_header.startswith("Bearer "):
return auth_header[7:] # Remove "Bearer " prefix
# Try to get from request body
try:
body = await request.json()
return body.get("session_token")
except Exception:
return None
async def get_user_from_cookie_string(cookie_header: str) -> Optional[UUID]:
"""Parse cookie header and return user ID if valid session exists."""
if not cookie_header:
return None
# Parse cookies from header (simple implementation)
cookies = {}
for cookie in cookie_header.split(";"):
cookie = cookie.strip()
if "=" in cookie:
name, value = cookie.split("=", 1)
cookies[name] = value
session_token = cookies.get(COOKIE_NAME)
if not session_token:
return None
token_data = validate_session_token(session_token)
if not token_data:
return None
return token_data["user_id"]