starred/MediaManager-maxdorninger
1
0
Fork 0
mirror of https://github.com/maxdorninger/MediaManager.git synced 2026-04-17 15:13:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

add working auth and sessions with passwords, refactor variable names and make psycopg cursor return a dictionary instead of an unnamed tuple

Browse Source
This commit is contained in:
maxid
2025-02-22 18:15:08 +01:00
parent e38d2e872c
commit 8b6d31c6a0
7 changed files with 155 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
MediaManager/src/auth/__init__.py
View File

@@ -1 +1,69 @@
import logging
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status, APIRouter
from fastapi.security import OAuth2, OAuth2AuthorizationCodeBearer
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel
import database
from database import UserInternal
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
uid: str | None = None
# to get a string like this run:
# openssl rand -hex 32
# TODO: remove secrets from files
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
log = logging.getLogger(__name__)
log.level = logging.DEBUG
log.addHandler(logging.StreamHandler())
router = APIRouter()
async def get_current_user(token: str) -> UserInternal:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
log.debug("token: "+ token)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
log.debug("jwt payload: "+payload.__str__())
user_uid: str = payload.get("sub")
log.debug("jwt payload sub (user uid): "+user_uid)
if user_uid is None:
raise credentials_exception
token_data = TokenData(uid=user_uid)
except InvalidTokenError:
log.warning("received invalid token: "+token)
raise credentials_exception
user = database.get_user(uid=token_data.uid)
if user is None:
log.debug("user not found")
raise credentials_exception
log.debug("received user: "+user.__str__())
return user
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt

40
MediaManager/src/auth/oidc.py
View File

@@ -1,9 +1,37 @@
from os import environ
from fastapi import Depends, APIRouter
from fastapi.security import OpenIdConnect
from fastapi.openapi.models import OAuthFlows, OAuthFlowAuthorizationCode
from fastapi.security import OpenIdConnect, OAuth2AuthorizationCodeBearer
from pydantic import BaseModel
from auth import router
#TODO: Implement OAuth2/Open ID Connect
class Settings(BaseModel):
OAUTH2_AUTHORIZATION_URL: str
OAUTH2_TOKEN_URL: str
OAUTH2_SCOPE: str
@property
def oauth2_flows(self) -> OAuthFlows:
return OAuthFlows(
authorizationCode=OAuthFlowAuthorizationCode(
authorizationUrl=self.OAUTH2_AUTHORIZATION_URL,
tokenUrl=self.OAUTH2_TOKEN_URL,
scopes={self.OAUTH2_SCOPE: "Access to this API"},
),
)
oauth2 = OAuth2AuthorizationCodeBearer(
authorizationUrl="/authorize",
tokenUrl="/token",
)
@router.get("/foo")
async def bar(token = Depends()):
return token
oidc = OpenIdConnect(openIdConnectUrl="http://localhost:8080/realms/tools/.well-known/openid-configuration")
app = APIRouter()
@app.get("/foo")
async def bar(token = Depends(oidc)):
return token

72
MediaManager/src/auth/password.py
View File

@@ -1,45 +1,16 @@
from datetime import datetime, timedelta, timezone
from datetime import timedelta
from typing import Annotated
import bcrypt
import jwt
from fastapi import Depends, HTTPException, status, APIRouter
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel
import database
from auth import ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, Token, router
from database import UserInternal
# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
"disabled": False,
}
}
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: str | None = None
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = APIRouter()
def verify_password(plain_password, hashed_password):
@@ -63,46 +34,15 @@ def authenticate_user(email: str, password: str) -> bool | UserInternal:
:param password: password of the user
:return: if authentication succeeds, returns the user object with added name and lastname, otherwise or if the user doesn't exist returns False
"""
user = database.get_user(email)
user = database.get_user(email=email)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return True
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except InvalidTokenError:
raise credentials_exception
user = database.get_user(token_data.username)
if user is None:
raise credentials_exception
return user
@app.post("/token")
@router.post("/token")
async def login_for_access_token(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
@@ -111,11 +51,11 @@ async def login_for_access_token(
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.email}, expires_delta=access_token_expires
data={"sub": user.id}, expires_delta=access_token_expires
)
return Token(access_token=access_token, token_type="bearer")

55
MediaManager/src/database.py
View File

@@ -5,10 +5,13 @@ from logging import getLogger
from uuid import uuid4
import psycopg
from annotated_types.test_cases import cases
from psycopg.rows import dict_row
from pydantic import BaseModel
log = getLogger(__name__)
log.level = logging.DEBUG
log.addHandler(logging.StreamHandler())
class User(BaseModel):
@@ -62,9 +65,11 @@ class PgDatabase(Database):
port=os.getenv("DB_PORT"),
user=os.getenv("DB_USERNAME"),
password=os.getenv("DB_PASSWORD"),
dbname=os.getenv("DB_NAME")
dbname=os.getenv("DB_NAME"),
row_factory=dict_row
)
def init_db():
with PgDatabase() as db:
db.connection.execute("""
@@ -92,38 +97,46 @@ def create_user(user: UserInternal) -> bool:
:return: True if user was created, False otherwise
"""
with PgDatabase() as db:
try:
db.connection.execute(
"""
INSERT INTO users (id, name, lastname, email, hashed_password)
VALUES (%s, %s, %s, %s, %s)
""",
(user.id, user.name, user.lastname, user.email, user.hashed_password)
)
except psycopg.errors.UniqueViolation as e:
log.error(e)
return False
try:
db.connection.execute(
"""
INSERT INTO users (id, name, lastname, email, hashed_password)
VALUES (%s, %s, %s, %s, %s)
""",
(user.id, user.name, user.lastname, user.email, user.hashed_password)
)
except psycopg.errors.UniqueViolation as e:
log.error(e)
return False
log.info("User inserted successfully")
log.debug(f"Inserted following User:", user.model_dump())
log.debug(f"Inserted following User: "+ user.model_dump())
return True
def get_user(email: str) -> UserInternal | None:
def get_user(email: str = None, uid: str = None) -> UserInternal | None:
"""
either specify the email or uid to search for the user, if both parameters are provided the uid will be used
:param email: the users email address
:return: if user was found its is returned, otherwise None
:param uid: the users id
:return: if user was found it a UserInternal object is returned, otherwise None
"""
with PgDatabase() as db:
result = db.connection.execute(
"SELECT id, name, lastname, email, hashed_password FROM users WHERE email=%s",
(email,)
).fetchone()
if email is not None and uid is None:
result = db.connection.execute(
"SELECT id, name, lastname, email, hashed_password FROM users WHERE email=%s",
(email,)
).fetchone()
if uid is not None:
result = db.connection.execute(
"SELECT id, name, lastname, email, hashed_password FROM users WHERE id=%s",
(uid,)
).fetchone()
if result is None:
return None
user = UserInternal.model_construct(**dict(zip(["id", "name", "lastname", "email", "hashed_password"], result)))
user = UserInternal(id = result["id"], name = result["name"], lastname = result["lastname"], email = result["email"], hashed_password = result["hashed_password"])
log.debug(f"Retrieved User succesfully: {user.model_dump()} ")
return user

3
MediaManager/src/main.py
View File

@@ -8,8 +8,7 @@ from auth import password
app = FastAPI()
app.include_router(users.router, tags=["users"])
app.include_router(password.app, tags=["authentication"])
app.include_router(password.router, tags=["authentication"])
if __name__ == "__main__":

4
MediaManager/src/routers/__init__.py
View File

@@ -0,0 +1,4 @@
import logging
log = logging.getLogger(__name__)
log.level = logging.DEBUG

17
MediaManager/src/routers/users.py
View File

@@ -6,8 +6,10 @@ from pydantic import BaseModel
from starlette.responses import JSONResponse
import database
from auth.password import authenticate_user, get_password_hash
from auth import get_current_user
from auth.password import get_password_hash
from database import User, UserInternal
from routers import log
router = APIRouter(
prefix="/users",
@@ -23,9 +25,6 @@ class CreateUser(User):
"""
password: str
log = logging.getLogger(__name__)
log.level = logging.DEBUG
@router.post("/",status_code=201, responses = {
409: {"model": Message, "description": "User with provided email already exists"},
201:{"model": UserInternal, "description": "User created successfully"}
@@ -44,15 +43,15 @@ async def create_user(
@router.get("/me", response_model=User)
@router.get("/me")
async def read_users_me(
current_user: User = Depends(authenticate_user),
current_user: UserInternal = Depends(get_current_user),
):
return current_user
return JSONResponse(status_code=200, content=current_user.model_dump())
@router.get("/me/items")
async def read_own_items(
current_user: User = Depends(authenticate_user),
current_user: User = Depends(get_current_user),
):
return [{"item_id": "Foo", "owner": current_user.username}]
return [{"item_id": "Foo", "owner": current_user.name}]