Faster and simplified hash_secret() that directly produces urlsafe entries.

paskia/db/operations.py

@@ -11,7 +11,6 @@ import secrets
 from datetime import UTC, datetime, timedelta
 from uuid import UUID
 
-import base64url
 import uuid7
 
 from paskia import oidc_notify
@@ -588,7 +587,7 @@ def login(
     session = Session.create(
         user=user_uuid,
         credential=credential_uuid,
-        key=base64url.enc(hash_secret("cookie", token)),
+        key=hash_secret("cookie", token),
         host=host,
         ip=ip,
         user_agent=user_agent,
@@ -656,7 +655,7 @@ def create_credential_session(
 
     # Generate token and derive key
     token = secrets.token_urlsafe(12)
-    key = base64url.enc(hash_secret("cookie", token))
+    key = hash_secret("cookie", token)
 
     session = Session.create(
         user=user_uuid,

paskia/db/structs.py

@@ -5,7 +5,6 @@ import secrets
 from datetime import UTC, datetime
 from uuid import UUID
 
-import base64url
 import msgspec
 import uuid7
 
@@ -434,7 +433,7 @@ class Session(msgspec.Struct, dict=True, omit_defaults=True):
         """Create a new Session with the provided key.
 
         Args:
-            key: The base64url-encoded hashed session key (derived from secret via hash_secret then base64url.enc)
+            key: The hashed session key (derived from secret via hash_secret)
 
         Returns:
             Session object with key set
@@ -471,7 +470,7 @@ class ResetToken(msgspec.Struct, dict=True):
 
     def __post_init__(self):
         if not hasattr(self, "key"):
-            self.key: bytes = b""
+            self.key: str = ""
 
     @property
     def user(self) -> User:
@@ -487,15 +486,15 @@ class ResetToken(msgspec.Struct, dict=True):
         del db.data().reset_tokens[self.key]
 
     @staticmethod
-    def hash(passphrase: str) -> bytes:
+    def hash(passphrase: str) -> str:
-        """Hash a passphrase to bytes for reset token storage."""
+        """Hash a passphrase to string for reset token storage."""
         if not passphrase_util.is_well_formed(passphrase):
             raise ValueError(
                 "Trying to reset with a session token in place of a passphrase"
                 if len(passphrase) == 16
                 else "Invalid passphrase format"
             )
-        return hashlib.sha512(passphrase.encode()).digest()[:9]
+        return hash_secret("reset", passphrase)
 
     @classmethod
     def by_passphrase(cls, passphrase: str) -> ResetToken | None:
@@ -627,7 +626,7 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
     users: dict[UUID, User] = {}
     credentials: dict[UUID, Credential] = {}
     sessions: dict[str, Session] = {}
-    reset_tokens: dict[bytes, ResetToken] = {}
+    reset_tokens: dict[str, ResetToken] = {}
     # OIDC provider data
     oidc: OIDC = msgspec.field(default_factory=lambda: OIDC())
 
@@ -670,7 +669,7 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
             SessionContext if valid, None if session not found, expired, or host mismatch
         """
 
-        key = base64url.enc(hash_secret("cookie", session_secret))
+        key = hash_secret("cookie", session_secret)
         try:
             s = self.sessions[key]
         except KeyError:

paskia/fastapi/oid.py

@@ -40,7 +40,7 @@ def _oidc_session_by_token(
     token: str, client_uuid: UUID | None = None
 ) -> Session | None:
     """Look up an OIDC session by token (refresh token value)."""
-    key = base64url.enc(hash_secret("oidc", token))
+    key = hash_secret("oidc", token)
     s = db.data().sessions.get(key)
     if not s or s.client_uuid is None:
         return None

paskia/fastapi/ws.py

@@ -3,7 +3,6 @@ from datetime import UTC, datetime
 from urllib.parse import urlencode
 from uuid import UUID
 
-import base64url
 from fastapi import FastAPI, WebSocket
 
 from paskia import authcode, db
@@ -218,7 +217,7 @@ async def websocket_authenticate(
         session = Session.create(
             user=cred.user_uuid,
             credential=cred.uuid,
-            key=base64url.enc(hash_secret("oidc", token)),
+            key=hash_secret("oidc", token),
             host=normalized_host,
             ip=metadata["ip"],
             user_agent=metadata["user_agent"],

paskia/util/crypto.py

@@ -1,17 +1,15 @@
 import hashlib
 
+import base64url
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 
-def hash_secret(*data) -> bytes:
+def hash_secret(*data: str | bytes, length=12) -> str:
-    """A custom HMAC that securily combines and hashes the given data (context, secrets). The first argument should be a namespacing string."""
+    """A custom HMAC that securily combines and hashes the given data. The first argument should be a namespacing string."""
-    inner = bytearray(len(data).to_bytes(8, "big"))
+    p = [d.encode() if hasattr(d, "encode") else d for d in data]
-    for d in data:
+    p += [len(x).to_bytes(8, "little") for x in [p, *p]]
-        if isinstance(d, str):
+    return base64url.enc(hashlib.sha256(b"".join(p)).digest()[:length])
-            d = d.encode()
-        inner += hashlib.sha256(d).digest()
-    return hashlib.sha256(inner).digest()[:12]
 
 
 def secret_key() -> bytes:

tests/conftest.py

@@ -19,7 +19,6 @@ from collections.abc import AsyncGenerator
 from datetime import UTC, datetime, timedelta
 from uuid import UUID
 
-import base64url
 import httpx
 import pytest
 import pytest_asyncio
@@ -268,7 +267,7 @@ def create_test_session(
 
     # Generate token and derive key
     token = secrets.token_urlsafe(12)
-    key = base64url.enc(hash_secret("cookie", token))
+    key = hash_secret("cookie", token)
 
     session = Session.create(
         user=user_uuid,

tests/test_admin.py

@@ -16,7 +16,6 @@ import secrets
 from datetime import UTC, datetime
 from uuid import UUID
 
-import base64url
 import httpx
 import pytest
 import pytest_asyncio
@@ -1301,7 +1300,7 @@ class TestAdminSessions:
         test_user,
     ):
         """Admin can delete their own current session."""
-        session_db_key = base64url.enc(hash_secret("cookie", session_token))
+        session_db_key = hash_secret("cookie", session_token)
         response = await client.delete(
             f"/auth/api/admin/users/{test_user.uuid}/sessions/{session_db_key}",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},