A non-functional draft, saving to allow reverts.

passkeyauth/.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+dist/
+.*
+!.gitignore
+*.lock
+*.db

passkeyauth/TODO.txt

@@ -0,0 +1,6 @@
+Practical checks:
+- Check how user verification works
+
+Details:
+- Extract authenticator type
+- Extract user verified flag (present always required)

passkeyauth/__init__.py

@@ -0,0 +1 @@
+# passkeyauth package

passkeyauth/db.py

@@ -0,0 +1,149 @@
+import sqlite3
+
+DB_PATH = "webauthn.db"
+
+
+def init_database():
+    """Initialize the SQLite database with required tables"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+
+    # Create users table
+    cursor.execute(
+        """
+        CREATE TABLE IF NOT EXISTS users (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT UNIQUE NOT NULL,
+            user_id BLOB NOT NULL,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    )
+
+    # Create credentials table
+    cursor.execute(
+        """
+        CREATE TABLE IF NOT EXISTS credentials (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id INTEGER NOT NULL,
+            credential_id BLOB NOT NULL,
+            public_key BLOB NOT NULL,
+            sign_count INTEGER DEFAULT 0,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            FOREIGN KEY (user_id) REFERENCES users (id),
+            UNIQUE(credential_id)
+        )
+        """
+    )
+
+    conn.commit()
+    conn.close()
+
+
+def get_user_by_username(username: str) -> dict | None:
+    """Get user record by username"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(
+        "SELECT id, username, user_id FROM users WHERE username = ?", (username,)
+    )
+    row = cursor.fetchone()
+    conn.close()
+
+    if row:
+        return {"id": row[0], "username": row[1], "user_id": row[2]}
+    return None
+
+
+def get_user_by_user_id(user_id: bytes) -> dict | None:
+    """Get user record by WebAuthn user ID"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(
+        "SELECT id, username, user_id FROM users WHERE user_id = ?", (user_id,)
+    )
+    row = cursor.fetchone()
+    conn.close()
+
+    if row:
+        return {"id": row[0], "username": row[1], "user_id": row[2]}
+    return None
+
+
+def create_user(username: str, user_id: bytes) -> int:
+    """Create a new user and return the user ID"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(
+        "INSERT INTO users (username, user_id) VALUES (?, ?)", (username, user_id)
+    )
+    user_db_id = cursor.lastrowid
+    conn.commit()
+    conn.close()
+    if user_db_id is None:
+        raise RuntimeError("Failed to create user")
+    return user_db_id
+
+
+def store_credential(user_db_id: int, credential_id: bytes, public_key: bytes) -> None:
+    """Store a credential for a user"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(
+        "INSERT INTO credentials (user_id, credential_id, public_key) VALUES (?, ?, ?)",
+        (user_db_id, credential_id, public_key),
+    )
+    conn.commit()
+    conn.close()
+
+
+def get_credential_by_id(credential_id: bytes) -> dict | None:
+    """Get credential by credential ID"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(
+        """
+        SELECT c.public_key, c.sign_count, u.username
+        FROM credentials c
+        JOIN users u ON c.user_id = u.id
+        WHERE c.credential_id = ?
+        """,
+        (credential_id,),
+    )
+    row = cursor.fetchone()
+    conn.close()
+
+    if row:
+        return {"public_key": row[0], "sign_count": row[1], "username": row[2]}
+    return None
+
+
+def get_user_credentials(username: str) -> list[bytes]:
+    """Get all credential IDs for a user"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(
+        """
+        SELECT c.credential_id
+        FROM credentials c
+        JOIN users u ON c.user_id = u.id
+        WHERE u.username = ?
+        """,
+        (username,),
+    )
+    rows = cursor.fetchall()
+    conn.close()
+
+    return [row[0] for row in rows]
+
+
+def update_credential_sign_count(credential_id: bytes, sign_count: int) -> None:
+    """Update the sign count for a credential"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(
+        "UPDATE credentials SET sign_count = ? WHERE credential_id = ?",
+        (sign_count, credential_id),
+    )
+    conn.commit()
+    conn.close()

passkeyauth/main.py

@@ -0,0 +1,106 @@
+"""
+Minimal FastAPI WebAuthn server with WebSocket support for passkey registration and authentication.
+
+This module provides a simple WebAuthn implementation that:
+- Uses WebSocket for real-time communication
+- Supports Resident Keys (discoverable credentials) for passwordless authentication
+- Maintains challenges locally per connection
+- Uses SQLite database for persistent storage of users and credentials
+- Enables true passwordless authentication where users don't need to enter a username
+"""
+
+from pathlib import Path
+
+import db
+import uuid7
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.responses import FileResponse
+from fastapi.staticfiles import StaticFiles
+
+from passkeyauth.passkey import Passkey
+
+STATIC_DIR = Path(__file__).parent.parent / "static"
+
+passkey = Passkey(
+    rp_id="localhost",
+    rp_name="Passkey Auth",
+    origin="http://localhost:8000",
+)
+
+app = FastAPI(title="Passkey Auth")
+
+
+@app.websocket("/ws/new_user_registration")
+async def websocket_register_new(ws: WebSocket):
+    """Register a new user and with a new passkey credential."""
+    await ws.accept()
+    try:
+        form = await ws.receive_json()
+        user_id = uuid7.create().bytes
+        user_name = form["user_name"]
+        await register_chat(ws, user_id, username)
+        # Store the user in the database
+        await db.create_user(user_name, user_id)
+        await ws.send_json({"status": "success", "user_id": user_id.hex()})
+    except WebSocketDisconnect:
+        pass
+
+
+async def register_chat(ws: WebSocket, user_id: bytes, username: str):
+    """Generate registration options and send them to the client."""
+    options, challenge = passkey.reg_generate_options(
+        user_id=user_id,
+        username=username,
+    )
+    await ws.send_text(options)
+    # Wait for the client to use his authenticator to register
+    credential = passkey.reg_credential(await ws.receive_json())
+    passkey.reg_verify(credential, challenge)
+
+
+@app.websocket("/ws/authenticate")
+async def websocket_authenticate(ws: WebSocket):
+    await ws.accept()
+    try:
+        options = passkey.auth_generate_options()
+        await ws.send_json(options)
+        # Wait for the client to use his authenticator to authenticate
+        credential = passkey.auth_credential(await ws.receive_json())
+        # Fetch from the database by credential ID
+        stored_cred = await db.fetch_credential(credential.raw_id)
+        # Verify the credential matches the stored data, that is also updated
+        passkey.auth_verify(credential, stored_cred)
+        # Update the credential in the database
+        await db.update_credential(stored_cred)
+    except WebSocketDisconnect:
+        pass
+
+
+# Serve static files
+app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
+
+
+@app.get("/")
+async def get_index():
+    """Serve the main HTML page"""
+    return FileResponse(STATIC_DIR / "index.html")
+
+
+def main():
+    """Entry point for the application"""
+    import uvicorn
+
+    uvicorn.run(
+        "passkeyauth.main:app",
+        host="0.0.0.0",
+        port=8000,
+        reload=True,
+        log_level="info",
+    )
+
+
+# Initialize database on startup
+db.init_database()
+
+if __name__ == "__main__":
+    main()

passkeyauth/passkey.py

@@ -0,0 +1,166 @@
+"""
+WebAuthn handler class that combines registration and authentication functionality.
+
+This module provides a unified interface for WebAuthn operations including:
+- Registration challenge generation and verification
+- Authentication challenge generation and verification
+- Credential validation
+"""
+
+from webauthn import (
+    generate_authentication_options,
+    generate_registration_options,
+    verify_authentication_response,
+    verify_registration_response,
+)
+from webauthn.helpers import (
+    options_to_json,
+    parse_authentication_credential_json,
+    parse_registration_credential_json,
+)
+from webauthn.helpers.cose import COSEAlgorithmIdentifier
+from webauthn.helpers.structs import (
+    AuthenticationCredential,
+    AuthenticatorSelectionCriteria,
+    RegistrationCredential,
+    ResidentKeyRequirement,
+    UserVerificationRequirement,
+)
+from webauthn.registration.verify_registration_response import VerifiedRegistration
+
+
+class Passkey:
+    """WebAuthn handler for registration and authentication operations."""
+
+    def __init__(
+        self,
+        rp_id: str,
+        rp_name: str,
+        origin: str,
+        supported_pub_key_algs: list[COSEAlgorithmIdentifier] | None = None,
+    ):
+        """
+        Initialize the WebAuthn handler.
+
+        Args:
+            rp_id: The relying party identifier
+            rp_name: The relying party name
+            origin: The origin URL of the application
+            supported_pub_key_algs: List of supported COSE algorithms
+        """
+        self.rp_id = rp_id
+        self.rp_name = rp_name
+        self.origin = origin
+        self.supported_pub_key_algs = supported_pub_key_algs or [
+            COSEAlgorithmIdentifier.EDDSA,
+            # COSEAlgorithmIdentifier.ECDSA_SHA_256,
+            # COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_256,
+        ]
+
+    ### Registration Methods ###
+
+    def reg_generate_options(
+        self, user_id: bytes, username: str, display_name="", **regopts
+    ) -> tuple[str, bytes]:
+        """
+        Generate registration options for WebAuthn registration.
+
+        Args:
+            user_id: The user ID as bytes
+            username: The username
+            display_name: The display name (defaults to username if empty)
+
+        Returns:
+            JSON string containing registration options
+        """
+        options = generate_registration_options(
+            rp_id=self.rp_id,
+            rp_name=self.rp_name,
+            user_id=user_id,
+            user_name=username,
+            user_display_name=display_name or username,
+            authenticator_selection=AuthenticatorSelectionCriteria(
+                resident_key=ResidentKeyRequirement.REQUIRED,
+                user_verification=UserVerificationRequirement.PREFERRED,
+            ),
+            supported_pub_key_algs=self.supported_pub_key_algs,
+            **regopts,
+        )
+        return options_to_json(options), options.challenge
+
+    @staticmethod
+    def reg_credential(credential: dict | str) -> RegistrationCredential:
+        return parse_registration_credential_json(credential)
+
+    def reg_verify(
+        self,
+        credential: RegistrationCredential,
+        expected_challenge: bytes,
+    ) -> VerifiedRegistration:
+        """
+        Verify registration response.
+
+        Args:
+            credential: The credential response from the client
+            expected_challenge: The expected challenge bytes
+
+        Returns:
+            Registration verification result
+        """
+        registration = verify_registration_response(
+            credential=credential,
+            expected_challenge=expected_challenge,
+            expected_origin=self.origin,
+            expected_rp_id=self.rp_id,
+        )
+        return registration
+
+    ### Authentication Methods ###
+
+    async def auth_generate_options(
+        self, user_verification_required=False, **kwopts
+    ) -> str:
+        """
+        Generate authentication options for WebAuthn authentication.
+
+        Args:
+            user_verification_required: The user will have to re-enter PIN or use biometrics for this operation. Useful when accessing security settings etc.
+        Returns:
+            JSON string containing authentication options
+        """
+        options = generate_authentication_options(
+            rp_id=self.rp_id,
+            user_verification=(
+                UserVerificationRequirement.REQUIRED
+                if user_verification_required
+                else UserVerificationRequirement.PREFERRED
+            ),
+            **kwopts,
+        )
+        return options_to_json(options)
+
+    @staticmethod
+    def auth_credential(credential: dict | str) -> AuthenticationCredential:
+        """Convert the authentication credential from JSON to a dataclass instance."""
+        return parse_authentication_credential_json(credential)
+
+    async def auth_verify(
+        self,
+        credential: AuthenticationCredential,
+        expected_challenge: bytes,
+        stored_cred: dict,
+    ):
+        """
+        Verify authentication response against locally stored credential data.
+        """
+        # Verify the authentication response
+        verification = verify_authentication_response(
+            credential=credential,
+            expected_challenge=expected_challenge,
+            expected_origin=self.origin,
+            expected_rp_id=self.rp_id,
+            credential_public_key=stored_cred["public_key"],
+            credential_current_sign_count=stored_cred["sign_count"],
+        )
+        stored_cred["sign_count"] = verification.new_sign_count
+        return verification