scopes/api/auth/config.py

@@ -9,7 +9,7 @@ class Settings(BaseSettings):
     #, env_prefix='SCOPES_')
     auth_method: str = 'oauth2'
     jwt_key: str | None = None
-    jwt_algorithm: str = 'HS256'
+    jwt_atlgorithm: str = 'HS256'
 
 settings = Settings()
 

scopes/api/auth/user.py

@@ -1,15 +1,14 @@
 # scopes.api.auth user
 
 from datetime import datetime
-from fastapi import APIRouter, Body, Depends, HTTPException
+from fastapi import APIRouter, Body
-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.api.auth import config
+from scopes.config import settings
 from scopes.storage import user as st_user
 from scopes.storage.common import Storage
 
@@ -25,7 +24,7 @@ class User(UserBase):
     timeStamp: datetime | None = None
     pw_hash: str | None = Field(exclude=True, default=None)
 
-class UserEdit(UserBase):
+class UserIn(UserBase):
     password: str | None = None
 
     def hashPassword(self) -> str | None:
@@ -34,10 +33,6 @@ 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):
@@ -46,42 +41,31 @@ def hashPassword(pw):
 def verifyPassword(plain, hashed):
     return pw_hash.verify(plain, hashed)
 
-security = HTTPBasic()
+def getUser(storage: Storage, userName: str) -> User:
+    users = storage.create(st_user.Users)
+    user = users.queryLast(name=userName)
+    return user.toObject(User)
 
-type Credentials = Annotated[HTTPBasicCredentials, Depends(security)]
+async def authenticate(storage: deps.Storage, token: deps.Auth):
-
+    payload = jwt.decode(token, settings.jwt_key, algorithms=[settings.jwt_algorithm])
-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')
-    tr = users.queryLast(name=userName)
+    user = getUser(storage, userName)
-    users.storage.commit()
+    return user
-    return tr.toObject(User)
                        
 # routes
 
 router = APIRouter()
 
 @router.get('/users', response_model=list[User])
-async def get_users(users: deps.Users):
+async def get_users(storage: deps.Storage):
+    users = storage.create(st_user.Users)
     result = [tr.toObject(User) for tr in users.query()]
     return result
 
 @router.post('/users', response_model=User)
-async def post_user(user: Annotated[UserEdit, Body()], users: deps.Users,
+async def post_user(user: Annotated[UserIn, Body()], storage: deps.Storage):
-                    auth_user: deps.AuthUser):
+    users = storage.create(st_user.Users)
     tr = st_user.User.fromObject(user)
     users.save(tr)
-    #print('*** post_user', auth_user, tr)
     return tr.toObject(User)
 

scopes/api/deps.py

@@ -8,30 +8,15 @@ from typing import Annotated
 from scopes import config
 from scopes.api.auth import config as auth_config
 from scopes.storage import common
-from scopes.storage import user as st_user
 
 type Storage = Annotated[common.Storage, Depends(storage)]
-type Users = Annotated[st_user.Users, Depends(users)]
+type Auth = Annotated[str, Depends(auth_method())]
-
-type AuthUser = Annotated[str, Depends(auth_user())]
-type AuthToken = Annotated[str, Depends(auth_token)]
-
-# storage
 
 def storage():
     stf = config.StorageFactory(config.settings)
     return stf(config.settings.dbschema)
 
-def users(storage: Storage):
+def auth_method():
-    return storage.create(st_user.Users)
-
-# auth
-
-def auth_user():
     auth = auth_config.auth_methods.get(auth_config.settings.auth_method)
-    if auth is not None:
     return auth
 
-def auth_token():
-    return None
-

tests/test.env

@@ -9,7 +9,7 @@ DBUSER=testuser
 DBPASSWORD=secret
 DBSCHEMA=testing
 
-AUTH_METHOD=basic
+AUTH_METHOD=testing
 
 # openssl rand -hex 32:
 JWT_KEY=391a521ee5ba2b6ce67b8fdbb9f0ca88f40a87d3bc1e82974381b647bb0fc776

tests/test_api.py

@@ -3,15 +3,14 @@
 """Tests for the fastapi- (and asyncio-) based stuff"""
 
 from scopes import config
-from scopes.api.auth import config as auth_config
 config.settings = config.Settings(_env_file='tests/test.env')
-auth_config.settings = auth_config.Settings(_env_file='tests/test.env')
 
 import pytest
 from fastapi.testclient import TestClient
-from httpx2 import ASGITransport, AsyncClient, BasicAuth
+from httpx2 import ASGITransport, AsyncClient
 
 from scopes.api import main
+from scopes.api.auth import config as auth_config
 from scopes.api.main import app
 from scopes.storage import tracking, user as st_user
 from scopes.api.auth import user as api_user
@@ -24,6 +23,10 @@ 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')
 
@@ -70,14 +73,13 @@ def test_user(storage, caplog):
     storage.commit()
     with TestClient(app) as clt:
         caplog.set_level(config.settings.log_level)
-        auth = BasicAuth(username='jim', password='secret')
         users = clt.get('/users').json()
         assert len(users) == 1
         u01i = users[0]
         assert u01i.get(pw_hash) is None
         inp = {'name': 'jane', 'login': 'jane', 'display_name': 'Jane', 
                'email': 'jane@xmpl.org', 'password': 'secret'}
-        resp = clt.post('/users', json=inp, auth=auth)
+        resp = clt.post('/users', json=inp)
         assert resp.status_code == 200
         u02 = resp.json()
         assert u02['uid'] == 'user-2'
@@ -85,7 +87,7 @@ def test_user(storage, caplog):
         assert ts02 is not None
         del inp['password']
         inp['display_name'] = 'Jane Doe'
-        resp = clt.post('/users', json=inp, auth=auth)
+        resp = clt.post('/users', json=inp)
         assert resp.status_code == 200
         u02a = resp.json()
         assert u02a['uid'] == 'user-2'