api.auth: HTTP Basic Auth working

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_atlgorithm: str = 'HS256'
+    jwt_algorithm: str = 'HS256'
 
 settings = Settings()
 

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

scopes/api/deps.py

@@ -8,15 +8,30 @@ 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 Auth = Annotated[str, Depends(auth_method())]
+type Users = Annotated[st_user.Users, Depends(users)]
+
+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 auth_method():
+def users(storage: Storage):
+    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=testing
+AUTH_METHOD=basic
 
 # openssl rand -hex 32:
 JWT_KEY=391a521ee5ba2b6ce67b8fdbb9f0ca88f40a87d3bc1e82974381b647bb0fc776

tests/test_api.py

@@ -3,14 +3,15 @@
 """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
+from httpx2 import ASGITransport, AsyncClient, BasicAuth
 
 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
@@ -23,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')
 
@@ -73,13 +70,14 @@ 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)
+        resp = clt.post('/users', json=inp, auth=auth)
         assert resp.status_code == 200
         u02 = resp.json()
         assert u02['uid'] == 'user-2'
@@ -87,7 +85,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)
+        resp = clt.post('/users', json=inp, auth=auth)
         assert resp.status_code == 200
         u02a = resp.json()
         assert u02a['uid'] == 'user-2'