authentication programmatically

[auglotus]

I want to integrate FastAPI-Users with NiceGUI, there a problem is how to implement a Login page to click 'log in' button to execute user login auth.
Because NiceGUI is built upon FastAPI, so the integration is simple and the current_user dependency inject to NiceGUI route is working well, but except the login auth.

FastAPI-Users provide the auth route like POST /auth/jwt/login, but it seems that I can't trigger the route from FastAPI internally. If I use requests or fastapi.testclient to call the route, the call can be success but the call is another session not the main session.

FastAPI-Users provide a topic 'Create a user programmatically', follow it I can execute user management functions as well, so is there possibility to do user login programmatically?

[hgalytoby]

Since you didn't provide a minimal example, I don't know how you currently implement fastapi-users.

I don't know NiceGUI very well, but the login method I have tried to do.

fastapi-users refer to this https://fastapi-users.github.io/fastapi-users/latest/configuration/full-example/#sqlalchemy

NiceGUI refer to this https://github.com/zauberzeug/nicegui/blob/main/examples/authentication/main.py

@ui.page("/login")
async def login(
    user_manager: BaseUserManager = Depends(get_user_manager),
):
    async def try_login() ->
        credentials = OAuth2PasswordRequestForm(username=username.value, password=password.value)
        user = await user_manager.authenticate(credentials)

        if user is None or not user.is_active:
            ui.notify('Wrong username or password', color='negative')

        app.storage.user.update({'username': username.value, 'authenticated': True})
        ui.navigate.to(app.storage.user.get('referrer_path', '/'))

    if app.storage.user.get('authenticated', False):
        return RedirectResponse('/')

    with ui.card().classes('absolute-center'):
        username = ui.input('Username').on('keydown.enter', try_login)
        password = ui.input('Password', password=True, password_toggle_button=True).on('keydown.enter', try_login)
        ui.button('Log in', on_click=try_login)
    return None

I'm not sure there's a solution to your problem.

[auglotus]

@hgalytoby It's very greate! I'll try it.

My purpose for this issue are two: 1) can use full FastAPI-Users features in NiceGUI, 2) both WEB GUI and HTTP endpoints can be provided by one applicaiton or by integrated NiceGUI with an exist FastAPI app with ui.run_with.

[auglotus]

@hgalytoby Unfortunately, they failed, print(request.headers.get("Authorization")) get None.

request.headers._list.append((b'authorization', f "Bearer {app.storage.user.get('token')}".encode()))
or
request.headers.__dict__['_list'].append((b'authorization', f "Bearer {app.storage.user.get('token')}".encode()))

[hgalytoby]

I don't have any problem with it myself, maybe it's because we don't have the same code?

from typing import Optional

from fastapi.security import OAuth2PasswordRequestForm
from fastapi_users import BaseUserManager
from fastapi_users.authentication import Strategy
from starlette.datastructures import Headers
from fastapi import FastAPI, Request, Depends
from nicegui import Client, app, ui
from starlette.responses import RedirectResponse

from db import User
from users import current_active_user, get_user_manager, auth_backend


def init(fastapi_app: FastAPI) -> None:
    unrestricted_page_routes = {'/login'}

    @app.middleware("http")
    async def auth_middleware(request: Request, call_next):
        if not app.storage.user.get('authenticated', False):
            if request.url.path in Client.page_routes.values() and request.url.path not in unrestricted_page_routes:
                app.storage.user['referrer_path'] = request.url.path
                return RedirectResponse('/login')
        request._headers = Headers(scope=request.scope)
        request.headers.__dict__['_list'].append((b'authorization', f"Bearer {app.storage.user.get('token')}".encode()))
        print(f'auth_middleware auth: {request.headers.get("Authorization")}')
        return await call_next(request)

    @ui.page('/')
    async def main_page(request: Request, user: User = Depends(current_active_user)) -> None:
        print(f'main_page Auth: {request.headers.get("Authorization")}')
        print(f'main_page auth: {request.headers.get("authorization")}')
        print(f'main_page user: {user}')
        with ui.column().classes('absolute-center items-center'):
            ui.label(f'Hello {user.email}!').classes('text-2xl')
            ui.button(
                on_click=lambda: (app.storage.user.clear(), ui.navigate.to('/login')),
                icon='logout'
            ).props('outline round')

    @ui.page('/login')
    async def login(
        user_manager: BaseUserManager = Depends(get_user_manager),
        strategy: Strategy = Depends(auth_backend.get_strategy),
    ) -> Optional[RedirectResponse]:
        async def try_login() -> None:  # local function to avoid passing username and password as arguments
            credentials = OAuth2PasswordRequestForm(username=username.value, password=password.value)
            user = await user_manager.authenticate(credentials)
            if user is None or not user.is_active:
                ui.notify('Wrong username or password', color='negative')
            else:
                app.storage.user.update({
                    'token': await strategy.write_token(user),
                    'authenticated': True,
                })
                ui.navigate.to(app.storage.user.get('referrer_path', '/'))

        if app.storage.user.get('authenticated', False):
            return RedirectResponse('/')
        with ui.card().classes('absolute-center'):
            username = ui.input('Username').on('keydown.enter', try_login)
            password = ui.input(
                'Password', password=True, password_toggle_button=True
            ).on('keydown.enter', try_login)
            ui.button('Log in', on_click=try_login)
        return None

    ui.run_with(
        fastapi_app,
        storage_secret='pick your private secret here',
    )

2024-03-04.09.28.07.mov

[auglotus]

@hgalytoby Thanks for your help again! I found I mistakenly append 'authorization' into headers multiple times.

I summary the final solution below:

1. Config FastAPI-Users into NiceGUI following the official doc.

2. We need customize programmatically call functions (outside the dependency injection mechanism):

get_async_session_context = contextlib.asynccontextmanager(get_async_session)
get_user_db_context = contextlib.asynccontextmanager(get_user_db)
get_user_manager_context = contextlib.asynccontextmanager(get_user_manager)

async def user_authenticate(email: str, password: str) -> Optional[User]:
    try:
        async with get_async_session_context() as session:
            async with get_user_db_context(session) as user_db:
                async with get_user_manager_context(user_db) as user_manager:
                    user_logout()
                    credentials = OAuth2PasswordRequestForm(username=email, password=password) 
                    user = await user_manager.authenticate(credentials)
                    if user is None or not user.is_active:
                        return None
                    return user
    except:
        return None

async def user_create_token(user: User) -> Optional[str]:
    try:
        async with get_async_session_context() as session:
            async with get_user_db_context(session) as user_db:
                async with get_user_manager_context(user_db) as user_manager:
                    if user is None:
                        return None
                    strategy = get_jwt_strategy()
                    token = await strategy.write_token(user)
                    if token is not None:
                        return token
                    else:
                        return None
    except:
        return None    

async def user_check_token(token: str) -> bool:
    try:
        async with get_async_session_context() as session:
            async with get_user_db_context(session) as user_db:
                async with get_user_manager_context(user_db) as user_manager:
                    if token is None:
                        return False
                    strategy = get_jwt_strategy()
                    user = await strategy.read_token(token, user_manager)
                    if user is None or not user.is_active:
                        return False
                    else:
                        return True
    except:
        return False

3. Then define the Login Page, authenticate by email&password and save token to storage:

    @ui.page('/login')
    async def login_page() -> Optional[RedirectResponse]:
        async def try_login():
            user = await user_authenticate(email=email.value, password=password.value)
            token = await user_create_token(user)
            if token is not None:
                app.storage.user.update({'auth_token': token})
                ui.navigate.to(app.storage.user.get('referrer_path', '/'))  # go back to where the user wanted to go
            else:
                ui.notify('email or password wrong!', color='negative')

        if await user_check_token(app.storage.user.get('auth_token', None)):
            return RedirectResponse('/')
              
        with ui.card().classes('absolute-center'):
            email = ui.input('email').on('keydown.enter', try_login)
            password = ui.input('password', password=True, password_toggle_button=True).on('keydown.enter', try_login)
            ui.button('Login', on_click=try_login) 

4. And define the AuthMiddleware to intercept http route and inject token:

    # This middleware restricts access to all NiceGUI pages.
    # It redirects the user to the login page if they are not authenticated.
    @app.middleware("http")
    async def AuthMiddleware(request: Request, call_next):
        if not await user_check_token(app.storage.user.get('auth_token', None)):
            if request.url.path in Client.page_routes.values() and request.url.path not in unrestricted_page_routes:
                app.storage.user['referrer_path'] = request.url.path  # remember where the user wanted to go
                return RedirectResponse('/login')            
       
        # Remove original authorization header
        request.scope['headers'] = [e for e in request.scope['headers'] if not e[0] == b'authorization'] 
        # # add new authorization header
        request.scope['headers'].append((b'authorization', f"Bearer {app.storage.user.get('auth_token')}".encode()))
        
        return await call_next(request)

5. Using current_user (dependency injection mechanism):

    @ui.page('/')
    async def index_page(user: User = Depends(current_active_user)) -> None:
        ui.label("Hello, {user.email}")

6. Logout
   just app.storage.user.clear()

[MaiHoangViet1809]

app.storage.user

how did you make app.storage.user work with ui.run_with ? I try a lot way and nothing work, alway raise error:
RuntimeError: app.storage.user needs a storage_secret passed in ui.run()