Repository: All-Hands-AI/OpenHands
File: openhands/server/utils.py (Python)

Code before:
    if '\x00' in conversation_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Conversation ID contains invalid characters',
        )

    # Check for path traversal attempts
    if '..' in conversation_id or '/' in conversation_id or '\\' in conversation_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Conversation ID contains invalid path characters',
        )

    # Check for control characters and newlines
    if any(ord(c) < 32 for c in conversation_id):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Conversation ID contains control characters',
        )

    return conversation_id


# Custom type annotation for validated conversation IDs
# This combines FastAPI's Path parameter with automatic validation
ValidatedConversationId = Annotated[
    str,
    Path(
        ...,
        description='Conversation ID (max 100 characters, alphanumeric and hyphens only)',
        max_length=100,
    ),
    Depends(validate_conversation_id),
]


async def get_conversation_store(request: Request) -> ConversationStore | None:
    conversation_store: ConversationStore | None = getattr(
        request.state, 'conversation_store', None
    )
    if conversation_store:
        return conversation_store
    user_id = await get_user_id(request)
    conversation_store = await ConversationStoreImpl.get_instance(config, user_id)
    request.state.conversation_store = conversation_store
    return conversation_store


async def generate_unique_conversation_id(
    conversation_store: ConversationStore,
) -> str:

Review comment (suggestion):
Maybe we don’t need a custom type?

I wonder, as long as the Depends was on the endpoint, it would be nice to continue working with a str if we can

Code after:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Conversation ID contains invalid characters',
        )

    # Check for path traversal attempts
    if '..' in conversation_id or '/' in conversation_id or '\\' in conversation_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Conversation ID contains invalid path characters',
        )

    # Check for control characters and newlines
    if any(ord(c) < 32 for c in conversation_id):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='Conversation ID contains control characters',
        )

    return conversation_id


async def get_conversation_store(request: Request) -> ConversationStore | None:
    conversation_store: ConversationStore | None = getattr(
        request.state, 'conversation_store', None
    )
    if conversation_store:
        return conversation_store
    user_id = await get_user_id(request)
    conversation_store = await ConversationStoreImpl.get_instance(config, user_id)
    request.state.conversation_store = conversation_store
    return conversation_store


async def generate_unique_conversation_id(
    conversation_store: ConversationStore,
) -> str:
    conversation_id = uuid.uuid4().hex
    while await conversation_store.exists(conversation_id):
        conversation_id = uuid.uuid4().hex
    return conversation_id


async def get_conversation_metadata(
    conversation_id: str,
    conversation_store: ConversationStore = Depends(get_conversation_store),
) -> ConversationMetadata:
    """Get conversation metadata and validate user access without requiring an active conversation."""
    try:
        metadata = await conversation_store.get_metadata(conversation_id)
        return metadata