attempt to anonymize transcripts

Author: thoraxe

src/app/endpoints/conversations.py

@@ -20,6 +20,7 @@
 from app.database import get_session
 from utils.endpoints import check_configuration_loaded, validate_conversation_ownership
 from utils.suid import check_suid
+from utils.user_anonymization import get_anonymous_user_id
 
 logger = logging.getLogger("app.endpoints.handlers")
 router = APIRouter(tags=["conversations"])
@@ -154,13 +155,22 @@ def get_conversations_list_endpoint_handler(
 
     user_id, _, _ = auth
 
-    logger.info("Retrieving conversations for user %s", user_id)
+    # Get anonymous user ID for database lookup
+    anonymous_user_id = get_anonymous_user_id(user_id)
+
+    logger.info(
+        "Retrieving conversations for user %s (anonymous: %s)",
+        user_id[:8] + "..." if len(user_id) > 8 else user_id,
+        anonymous_user_id,
+    )
 
     with get_session() as session:
         try:
-            # Get all conversations for this user
+            # Get all conversations for this user using anonymous ID
             user_conversations = (
-                session.query(UserConversation).filter_by(user_id=user_id).all()
+                session.query(UserConversation)
+                .filter_by(anonymous_user_id=anonymous_user_id)
+                .all()
             )
 
             # Return conversation summaries with metadata

src/app/endpoints/query.py

@@ -24,16 +24,17 @@
 from configuration import configuration
 from app.database import get_session
 import metrics
+import constants
 from models.database.conversations import UserConversation
 from models.responses import QueryResponse, UnauthorizedResponse, ForbiddenResponse
 from models.requests import QueryRequest, Attachment
-import constants
 from utils.endpoints import (
     check_configuration_loaded,
     get_agent,
     get_system_prompt,
     validate_conversation_ownership,
 )
+from utils.user_anonymization import get_anonymous_user_id
 from utils.mcp_headers import mcp_headers_dependency, handle_mcp_headers_with_toolgroups
 from utils.transcripts import store_transcript
 from utils.types import TurnSummary
@@ -76,25 +77,31 @@ def is_transcripts_enabled() -> bool:
 def persist_user_conversation_details(
     user_id: str, conversation_id: str, model: str, provider_id: str
 ) -> None:
-    """Associate conversation to user in the database."""
+    """Associate conversation to user in the database using anonymous user ID."""
+    # Get anonymous user ID for database storage
+    anonymous_user_id = get_anonymous_user_id(user_id)
+
     with get_session() as session:
         existing_conversation = (
             session.query(UserConversation)
-            .filter_by(id=conversation_id, user_id=user_id)
+            .filter_by(id=conversation_id, anonymous_user_id=anonymous_user_id)
             .first()
         )
 
         if not existing_conversation:
             conversation = UserConversation(
                 id=conversation_id,
-                user_id=user_id,
+                anonymous_user_id=anonymous_user_id,
                 last_used_model=model,
                 last_used_provider=provider_id,
                 message_count=1,
             )
             session.add(conversation)
             logger.debug(
-                "Associated conversation %s to user %s", conversation_id, user_id
+                "Associated conversation %s to anonymous user %s (original: %s)",
+                conversation_id,
+                anonymous_user_id,
+                user_id[:8] + "..." if len(user_id) > 8 else user_id,
             )
         else:
             existing_conversation.last_used_model = model

src/models/database/conversations.py

@@ -16,8 +16,8 @@ class UserConversation(Base):  # pylint: disable=too-few-public-methods
     # The conversation ID
     id: Mapped[str] = mapped_column(primary_key=True)
 
-    # The user ID associated with the conversation
-    user_id: Mapped[str] = mapped_column(index=True)
+    # The anonymous user ID associated with the conversation
+    anonymous_user_id: Mapped[str] = mapped_column(index=True)
 
     # The last provider/model used in the conversation
     last_used_model: Mapped[str] = mapped_column()

src/models/database/user_mapping.py

@@ -0,0 +1,28 @@
+"""User ID anonymization mapping model."""
+
+from datetime import datetime
+
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import DateTime, func, Index
+
+from models.database.base import Base
+
+
+class UserMapping(Base):  # pylint: disable=too-few-public-methods
+    """Model for mapping real user IDs to anonymous UUIDs."""
+
+    __tablename__ = "user_mapping"
+
+    # Anonymous UUID used for all storage/analytics (primary key)
+    anonymous_id: Mapped[str] = mapped_column(primary_key=True)
+
+    # Original user ID from authentication (hashed for security)
+    user_id_hash: Mapped[str] = mapped_column(index=True, unique=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),  # pylint: disable=not-callable
+    )
+
+    # Index for efficient lookups
+    __table_args__ = (Index("ix_user_mapping_hash_lookup", "user_id_hash"),)

src/utils/endpoints.py

@@ -13,6 +13,7 @@
 from configuration import AppConfig
 from utils.suid import get_suid
 from utils.types import GraniteToolParser
+from utils.user_anonymization import get_anonymous_user_id
 
 
 logger = logging.getLogger("utils.endpoints")
@@ -21,11 +22,14 @@
 def validate_conversation_ownership(
     user_id: str, conversation_id: str
 ) -> UserConversation | None:
-    """Validate that the conversation belongs to the user."""
+    """Validate that the conversation belongs to the user using anonymous ID lookup."""
+    # Get anonymous user ID for database lookup
+    anonymous_user_id = get_anonymous_user_id(user_id)
+
     with get_session() as session:
         conversation = (
             session.query(UserConversation)
-            .filter_by(id=conversation_id, user_id=user_id)
+            .filter_by(id=conversation_id, anonymous_user_id=anonymous_user_id)
             .first()
         )
         return conversation

src/utils/transcripts.py

@@ -14,15 +14,16 @@
 from models.requests import Attachment, QueryRequest
 from utils.suid import get_suid
 from utils.types import TurnSummary
+from utils.user_anonymization import get_anonymous_user_id
 
 logger = logging.getLogger("utils.transcripts")
 
 
-def construct_transcripts_path(user_id: str, conversation_id: str) -> Path:
-    """Construct path to transcripts."""
+def construct_transcripts_path(anonymous_user_id: str, conversation_id: str) -> Path:
+    """Construct path to transcripts using anonymous user ID."""
     # these two normalizations are required by Snyk as it detects
     # this Path sanitization pattern
-    uid = os.path.normpath("/" + user_id).lstrip("/")
+    uid = os.path.normpath("/" + anonymous_user_id).lstrip("/")
     cid = os.path.normpath("/" + conversation_id).lstrip("/")
     file_path = (
         configuration.user_data_collection_configuration.transcripts_storage or ""
@@ -46,7 +47,7 @@ def store_transcript(  # pylint: disable=too-many-arguments,too-many-positional-
     """Store transcript in the local filesystem.
 
     Args:
-        user_id: The user ID (UUID).
+        user_id: The original user ID from authentication (will be anonymized).
         conversation_id: The conversation ID (UUID).
         query_is_valid: The result of the query validation.
         query: The query (without attachments).
@@ -56,7 +57,15 @@ def store_transcript(  # pylint: disable=too-many-arguments,too-many-positional-
         truncated: The flag indicating if the history was truncated.
         attachments: The list of `Attachment` objects.
     """
-    transcripts_path = construct_transcripts_path(user_id, conversation_id)
+    # Get anonymous user ID for storage
+    anonymous_user_id = get_anonymous_user_id(user_id)
+    logger.debug(
+        "Anonymized user %s to %s for transcript storage",
+        user_id[:8] + "..." if len(user_id) > 8 else user_id,
+        anonymous_user_id,
+    )
+
+    transcripts_path = construct_transcripts_path(anonymous_user_id, conversation_id)
     transcripts_path.mkdir(parents=True, exist_ok=True)
 
     data_to_store = {
@@ -65,7 +74,7 @@ def store_transcript(  # pylint: disable=too-many-arguments,too-many-positional-
             "model": model_id,
             "query_provider": query_request.provider,
             "query_model": query_request.model,
-            "user_id": user_id,
+            "anonymous_user_id": anonymous_user_id,  # Store anonymous ID only
             "conversation_id": conversation_id,
             "timestamp": datetime.now(UTC).isoformat(),
         },

src/utils/user_anonymization.py

@@ -0,0 +1,122 @@
+"""User ID anonymization utilities."""
+
+import hashlib
+import logging
+from typing import Optional
+
+from sqlalchemy.exc import IntegrityError
+
+from models.database.user_mapping import UserMapping
+from app.database import get_session
+from utils.suid import get_suid
+
+logger = logging.getLogger("utils.user_anonymization")
+
+
+def _hash_user_id(user_id: str) -> str:
+    """
+    Create a consistent hash of the user ID for mapping purposes.
+
+    Uses SHA-256 with a fixed salt to ensure consistent hashing
+    while preventing rainbow table attacks.
+    """
+    # Use a fixed salt - in production, this should be configurable
+    salt = "lightspeed_user_anonymization_salt_v1"
+    hash_input = f"{salt}:{user_id}".encode("utf-8")
+    return hashlib.sha256(hash_input).hexdigest()
+
+
+def get_anonymous_user_id(auth_user_id: str) -> str:
+    """
+    Get or create an anonymous UUID for a user ID from authentication.
+
+    This function:
+    1. Hashes the original user ID for secure storage
+    2. Looks up existing anonymous mapping
+    3. Creates new anonymous UUID if none exists
+    4. Returns the anonymous UUID for use in storage/analytics
+
+    Args:
+        auth_user_id: The original user ID from authentication
+
+    Returns:
+        Anonymous UUID string for this user
+    """
+    user_id_hash = _hash_user_id(auth_user_id)
+
+    with get_session() as session:
+        # Try to find existing mapping
+        existing_mapping = (
+            session.query(UserMapping).filter_by(user_id_hash=user_id_hash).first()
+        )
+
+        if existing_mapping:
+            logger.debug(
+                "Found existing anonymous ID for user hash %s", user_id_hash[:8] + "..."
+            )
+            return existing_mapping.anonymous_id
+
+        # Create new anonymous mapping
+        anonymous_id = get_suid()
+        new_mapping = UserMapping(anonymous_id=anonymous_id, user_id_hash=user_id_hash)
+
+        try:
+            session.add(new_mapping)
+            session.commit()
+            logger.info(
+                "Created new anonymous ID %s for user hash %s",
+                anonymous_id,
+                user_id_hash[:8] + "...",
+            )
+            return anonymous_id
+
+        except IntegrityError as e:
+            session.rollback()
+            # Race condition - another thread created the mapping
+            logger.warning("Race condition creating user mapping: %s", e)
+
+            # Try to fetch the mapping created by the other thread
+            existing_mapping = (
+                session.query(UserMapping).filter_by(user_id_hash=user_id_hash).first()
+            )
+
+            if existing_mapping:
+                return existing_mapping.anonymous_id
+
+            # If we still can't find it, something is wrong
+            logger.error(
+                "Failed to create or retrieve user mapping for hash %s",
+                user_id_hash[:8] + "...",
+            )
+            raise RuntimeError("Unable to create or retrieve anonymous user ID") from e
+
+
+def get_user_count() -> int:
+    """
+    Get the total number of unique users in the system.
+
+    Returns:
+        Total count of unique anonymous users
+    """
+    with get_session() as session:
+        return session.query(UserMapping).count()
+
+
+def find_anonymous_user_id(auth_user_id: str) -> Optional[str]:
+    """
+    Find existing anonymous ID for a user without creating a new one.
+
+    Args:
+        auth_user_id: The original user ID from authentication
+
+    Returns:
+        Anonymous UUID if found, None otherwise
+    """
+    user_id_hash = _hash_user_id(auth_user_id)
+
+    with get_session() as session:
+        existing_mapping = (
+            session.query(UserMapping).filter_by(user_id_hash=user_id_hash).first()
+        )
+
+        return existing_mapping.anonymous_id if existing_mapping else None