From 342f48042cf0ab7833715bb1e225fa6d712d2bd6 Mon Sep 17 00:00:00 2001 From: Helmut Merz Date: Mon, 9 Nov 2026 11:36:18 +0100 Subject: [PATCH] api.auth: HTTP Basic Auth working --- scopes/api/auth/user.py | 35 ++++++++++++++++++++++++++++------- scopes/api/deps.py | 4 ++-- tests/test.env | 2 +- tests/test_api.py | 4 ---- 4 files changed, 31 insertions(+), 14 deletions(-) diff --git a/scopes/api/auth/user.py b/scopes/api/auth/user.py index 88634c0..014918f 100644 --- a/scopes/api/auth/user.py +++ b/scopes/api/auth/user.py @@ -1,14 +1,15 @@ # scopes.api.auth user from datetime import datetime -from fastapi import APIRouter, Body +from fastapi import APIRouter, Body, Depends, HTTPException +from fastapi.security import HTTPBasic, HTTPBasicCredentials import jwt from pwdlib import PasswordHash from pydantic import BaseModel, Field from typing import Annotated, List from scopes.api import deps -from scopes.config import settings +from scopes.api.auth import config from scopes.storage import user as st_user from scopes.storage.common import Storage @@ -33,6 +34,10 @@ class UserEdit(UserBase): # password handling, authentication +def dummy_auth() -> User: + return User(name='tester', login='tester', display_name='Tester') +config.auth_methods['testing'] = dummy_auth + pw_hash = PasswordHash.recommended() def hashPassword(pw): @@ -41,11 +46,27 @@ def hashPassword(pw): def verifyPassword(plain, hashed): return pw_hash.verify(plain, hashed) -def userFromToken(users: deps.Users, token: deps.AuthToken): - payload = jwt.decode(token, settings.jwt_key, algorithms=[settings.jwt_algorithm]) +security = HTTPBasic() + +type Credentials = Annotated[HTTPBasicCredentials, Depends(security)] + +def basicAuth(users: deps.Users, cred: Credentials) -> User: + tr = users.queryLast(login=cred.username) + users.storage.commit() + user = tr.toObject(User) + if verifyPassword(cred.password, user.pw_hash): + return user + raise HTTPException(status_code=401, detail='credentials', + headers={"WWW-Authenticate": "Basic"}) +config.auth_methods['basic'] = basicAuth + +def userFromToken(users: deps.Users, token: deps.AuthToken) -> User: + payload = jwt.decode(token, config.settings.jwt_key, + algorithms=[config.settings.jwt_algorithm]) userName = payload.get('sub') - user = users.queryLast(name=userName) - return user + tr = users.queryLast(name=userName) + users.storage.commit() + return tr.toObject(User) # routes @@ -59,8 +80,8 @@ async def get_users(users: deps.Users): @router.post('/users', response_model=User) async def post_user(user: Annotated[UserEdit, Body()], users: deps.Users, auth_user: deps.AuthUser): - print('*** post_user', auth_user) tr = st_user.User.fromObject(user) users.save(tr) + #print('*** post_user', auth_user, tr) return tr.toObject(User) diff --git a/scopes/api/deps.py b/scopes/api/deps.py index 2f9e4a8..fd9e8ad 100644 --- a/scopes/api/deps.py +++ b/scopes/api/deps.py @@ -13,7 +13,7 @@ from scopes.storage import user as st_user type Storage = Annotated[common.Storage, Depends(storage)] type Users = Annotated[st_user.Users, Depends(users)] -type AuthUser = Annotated[str, Depends(auth_user)] +type AuthUser = Annotated[str, Depends(auth_user())] type AuthToken = Annotated[str, Depends(auth_token)] # storage @@ -30,7 +30,7 @@ def users(storage: Storage): def auth_user(): auth = auth_config.auth_methods.get(auth_config.settings.auth_method) if auth is not None: - return auth() + return auth def auth_token(): return None diff --git a/tests/test.env b/tests/test.env index b8a1832..33d27e9 100644 --- a/tests/test.env +++ b/tests/test.env @@ -9,7 +9,7 @@ DBUSER=testuser DBPASSWORD=secret DBSCHEMA=testing -AUTH_METHOD=testing +AUTH_METHOD=basic # openssl rand -hex 32: JWT_KEY=391a521ee5ba2b6ce67b8fdbb9f0ca88f40a87d3bc1e82974381b647bb0fc776 diff --git a/tests/test_api.py b/tests/test_api.py index 3818ae0..001df7e 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -24,10 +24,6 @@ def storage(): storage.commit() return storage -def dummy_auth(): - return api_user.User(name='tester', login='tester', display_name='Tester') -auth_config.auth_methods['testing'] = dummy_auth - def async_client(): return AsyncClient(transport=ASGITransport(app=app), base_url='http://test')