Authentication for Custom API

[saimanoj1206]

I implemented a custom endpoint and expected it to be protected, but seems like the request headers are not receiving the Bearer token from frontend. Here is the code that I've used to implement endpoints and mount it to the chainlit app:

import logging
import os
from typing import Union

import chainlit as cl
import uvicorn
from chainlit.auth import authenticate_user
from chainlit.utils import mount_chainlit
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import RedirectResponse
from typing_extensions import Annotated

from systrans_ai_assistant.database.auth import PasswordHandler
from systrans_ai_assistant.database.db_utils import get_url
from systrans_ai_assistant.database.storage_client import (
    AzureStorageClient,
    FSStorageClient,
)
from systrans_ai_assistant.utils import create_logger, load_config

app = FastAPI()

ENV = os.environ.get("GOV_AI_ENV")
USER_CONFIG_PATH = f"configs/chatbot/{ENV}/user_config.yaml"
DATA_CONFIG_PATH = f"configs/chatbot/{ENV}/data_config.yaml"

user_config = load_config(USER_CONFIG_PATH)
data_config = load_config(DATA_CONFIG_PATH)

# Set up logger
create_logger(
    logger_name=data_config.gov_ai.logging.logger_name,
    level=data_config.gov_ai.logging.logger_level,
    log_file_path=data_config.gov_ai.logging.log_file_path,
)

# Get logger
logger = logging.getLogger(data_config.gov_ai.logging.logger_name)

# storage client
storage_client = AzureStorageClient(
    account_url=data_config.gov_ai.storage_client.account_url,
    container_name=data_config.gov_ai.storage_client.container_name,
    prefix=data_config.gov_ai.storage_client.prefix,
    chainlit_endpoint=data_config.gov_ai.chainlit.endpoint,
)



@app.get("/")
async def redirect_to_chainlit():
    logger.debug("Redirecting to Chainlit /chat")
    return RedirectResponse(url="/chat")


@app.get("/blob/data/{billing_guide_version}/pdf_splits/{blob_name}")
# async def pdf_view(billing_guide_version, blob_name):
async def pdf_view(
    request: Request,
    billing_guide_version,
    blob_name,
    current_user: Annotated[Union[cl.User], Depends(authenticate_user)],
):
    """
    Retrieve a PDF split from blob storage.

    Parameters
    ----------
    billing_guide_version : str
        The version of the billing guide to retrieve.
    blob_name : str
        The name of the blob file to retrieve.

    Returns
    -------
    StreamingResponse
        A response containing the requested PDF split from blob storage.
    """
    # cl.session
    print(request.headers)
    return await storage_client.get_blob(
        f"data/{billing_guide_version}/pdf_splits", f"{blob_name}"
    )

mount_chainlit(app=app, target="systrans_ai_assistant/app.py", path="/chat")

How do I make the headers come along with the request here?

Any guidance on this would be appreciated!

[dosubot]

I found a similar issue that might be helpful:

• Is it possible to access the full request object, in particular the headers? - The suggested solution is to use the @cl.header_auth_callback handler to access the headers, which may help you ensure the Bearer token is included in your requests.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[willydouhard]

So it always fail right? It can't authenticate any user? What authentication is setup? Oauth? password?

[saimanoj1206]

Yes the authentication for those custom endpoints is not working at all. I am using password authentication, here is the code for the same:

@cl.password_auth_callback
async def password_auth_callback(username: str, password: str) -> Optional[cl.User]:
    """
    Authenticates a user based on the provided username and password.

    Parameters
    ----------
    username : str
        The username of the user attempting to authenticate.
    password : str
        The password of the user attempting to authenticate.

    Returns
    -------
    Optional[cl.User]
        A User object if authentication is successful; otherwise, None.
    """
    user = await password_db.login(username, password)
    if user["user"]:
        return cl.User(identifier=user["user"])
    return None

The login page and all other functionalities, such as the main page and resume, are working fine. However, only the custom endpoints are not receiving the required headers. When I remove the request from the input, it works, but it is not secure, as users can share the PDF links, which we would like to prevent.

[willydouhard]

How do you call the custom endpoint? With fetch, I had to add credentials: include

[saimanoj1206]

I am using the endpoint for pdf element, setting the endpoint URL as pdf URL. I figured out the authorization headers are not coming from frontend. How do I fix this? Any guidance on this would be helpful

[willydouhard]

It is working on my end.

Chainlit app

import chainlit as cl

@cl.password_auth_callback
def auth_callback(username: str, password: str):
    if (username, password) == ("admin", "admin"):
        return cl.User(
            identifier="Admin", metadata={}
        )
    else:
        return None

@cl.on_chat_start
async def main():
    pdf = cl.Pdf(url="/app")
    await cl.Message(content="Hello World", elements=[pdf]).send()

Fast api app

from fastapi import FastAPI, Depends
from chainlit.auth import get_current_user
from chainlit.utils import mount_chainlit
from chainlit.server import GenericUser
from typing_extensions import Annotated

app = FastAPI()

UserParam = Annotated[GenericUser, Depends(get_current_user)]


@app.get("/app")
def read_main(current_user: UserParam):
    print(current_user)
    return {"message": "Hello World from main app"}

mount_chainlit(app=app, target="my_cl_app.py", path="/chainlit")

The chainlit app requires auth, then sends a message containing a pdf element. The src of the PDF is the /app endpoint protected by auth. It correctly prints the current user