API Users wit Cookies

Sorry for the wrong branch

auth/dependencies.py

@@ -0,0 +1,33 @@
+from fastapi import Depends, status, Request
+from fastapi.security import OAuth2PasswordBearer
+import jwt
+# auth/dependencies.py
+
+from sqlalchemy.orm import Session
+from database import get_db
+from models.user import User
+from config import settings
+from exceptions import UnauthorizedError
+
+
+def get_me(request: Request, db: Session = Depends(get_db)) -> User:
+
+    token = request.cookies.get("access_token")
+    print(token)
+    if not token:
+        raise UnauthorizedError("Token missing")
+    try:
+        # 1. Token entschlüsseln
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+        user_id: str = payload.get("sub")
+        if user_id is None:
+            raise UnauthorizedError("Token invalid or expired")
+    except jwt.PyJWTError:
+        raise UnauthorizedError("Token invalid or expired")
+
+    # 2. User aus der Datenbank laden
+    user = db.get(User, int(user_id))
+    if user is None:
+        raise UnauthorizedError("Token invalid or expired")
+
+    return user

auth/security.py

@@ -0,0 +1,23 @@
+from datetime import datetime, timedelta, timezone
+import jwt
+from pwdlib import PasswordHash
+from pwdlib.hashers.argon2 import Argon2Hasher
+from config import settings #Settings from .env
+
+
+
+password_hash = PasswordHash((Argon2Hasher(),))
+
+def hash_password(password: str) -> str:
+    return password_hash.hash(password)
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return password_hash.verify(plain_password, hashed_password)
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
+    to_encode = data.copy()
+    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+
+

config.py

@@ -0,0 +1,17 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+class Settings(BaseSettings):
+    # Tandoor Server
+    API_KEY: str
+    TANDOOR_URL: str
+
+    # Security
+    SECRET_KEY: str
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+
+    # Sagt Pydantic, dass es die Werte aus der .env-Datei laden soll
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
+
+# Instanz erzeugen, die du im gesamten Projekt importierst
+settings = Settings()

crud/user.py

@@ -0,0 +1,79 @@
+# crud/user.py
+from sqlalchemy import select, exists, func
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import IntegrityError
+from models.user import User
+from exceptions import UserNotFoundError, UserAlreadyExistsError
+from auth.security import hash_password
+
+
+def get_users(db: Session) -> list[User]:
+    stmt = select(User)
+    users = db.scalars(stmt).all()
+
+    if not users:
+        raise UserNotFoundError(f"No users in Database")
+
+    return users
+
+def get_user_by_id(db: Session, user_id: int) -> User:
+    stmt = select(User).where(User.id == user_id)
+    user = db.scalars(stmt).first()
+
+    if user is None:
+        raise UserNotFoundError(f"User with id: {user_id} not found")
+
+    return user
+
+def get_user_by_email(db: Session, email: str) -> User:
+    stmt = select(User).where(func.lower(User.email) == func.lower(email))
+    user = db.scalars(stmt).first()
+
+    if user is None:
+        raise UserNotFoundError(f"User with email: {email} not found")
+
+    return user
+
+
+def create_user(db: Session, email: str, password: str, role: bool = "user") -> User:
+    hashed = hash_password(password)
+    user = User(email=email, password=hashed, role=role)
+
+    try:
+        db.add(user)
+        db.commit()
+        db.refresh(user)
+        return user
+    except IntegrityError as err:
+        db.rollback()
+        raise UserAlreadyExistsError(f"User with email: '{email}' already exists") from err
+
+
+def update_user(db: Session, user_id: int, user_data: dict) -> User:
+    user = get_user_by_id(db, user_id)
+    if not user:
+        raise UserNotFoundError(f"User with ID {user_id} not found")
+
+    # Passwort hashen, falls es im Update enthalten ist
+    if "password" in user_data and user_data["password"] is not None:
+        user_data["password"] = hash_password(user_data.pop("password"))
+
+    # Nur die mitgeschickten Felder auf dem Objekt setzen
+    for field, value in user_data.items():
+        setattr(user, field, value)
+
+    try:
+        db.commit()
+        db.refresh(user)
+        return user
+    except IntegrityError as err:
+        db.rollback()
+        raise UserAlreadyExistsError("Email is already in use") from err
+
+def delete_user(db: Session, user_id: int) -> User:
+    user = get_user_by_id(db, user_id)
+    if user is None:
+        raise UserNotFoundError(f"User with id: {user_id} not found")
+    db.delete(user)
+    db.commit()
+    return user

database.py

@@ -0,0 +1,21 @@
+# database.py
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, declarative_base
+
+DATABASE_URL = "sqlite:///./tandur.db"   # Datei landet im Projektordner
+
+engine = create_engine(
+    DATABASE_URL,
+    connect_args={"check_same_thread": False}   # nötig für SQLite + FastAPI
+)
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

exceptions.py

@@ -0,0 +1,79 @@
+from typing import Any
+
+
+class AppError(Exception):
+    """Basisklasse für alle eigenen Fehler dieser App"""
+    def __init__(self, message: str = "Ein unerwarteter Fehler ist aufgetreten."):
+        self.message = message
+        super().__init__(self.message)
+
+
+# ==========================================
+# 1. Datenbank- & Ressourcen-Fehler
+# ==========================================
+
+class DatabaseError(AppError):
+    """Basisklasse für Datenbank-bezogene Fehler"""
+    pass
+
+
+class NotFoundError(DatabaseError):
+    """Ein angefragter Datensatz existiert nicht"""
+    def __init__(self, resource: str = "Ressource", identifier: Any = None):
+        if identifier:
+            msg = f"{resource} mit ID/Kennung '{identifier}' wurde nicht gefunden."
+        else:
+            msg = f"{resource} wurde nicht gefunden."
+        super().__init__(msg)
+
+
+class AlreadyExistsError(DatabaseError):
+    """Ein Unique-Constraint wurde verletzt (z. B. doppelte E-Mail)"""
+    def __init__(self, resource: str = "Ressource", field: str = "Eintrag", value: Any = None):
+        if value:
+            msg = f"{resource} mit {field} '{value}' existiert bereits."
+        else:
+            msg = f"{resource} existiert bereits."
+        super().__init__(msg)
+
+
+# Domänen-spezifische Ressourcen-Fehler
+class UserNotFoundError(NotFoundError):
+    def __init__(self, identifier: Any = None):
+        super().__init__(resource="User", identifier=identifier)
+
+
+class UserAlreadyExistsError(AlreadyExistsError):
+    def __init__(self, field: str = "E-Mail", value: Any = None):
+        super().__init__(resource="User", field=field, value=value)
+
+
+# ==========================================
+# 2. Authentifizierung & Autorisierung
+# ==========================================
+
+class AuthError(AppError):
+    """Basisklasse für Authentifizierungs- und Rechte-Fehler"""
+    pass
+
+
+class UnauthorizedError(AuthError):
+    """Ungültiges Token oder falsche Login-Daten"""
+    def __init__(self, message: str = "Ungültige Anmeldedaten oder Token abgelaufen."):
+        super().__init__(message)
+
+
+class ForbiddenError(AuthError):
+    """Fehlende Berechtigungen (z.B. Kein Admin)"""
+    def __init__(self, message: str = "Unzureichende Berechtigungen für diese Aktion."):
+        super().__init__(message)
+
+
+# ==========================================
+# 3. Validierungsfehler (Business Logic)
+# ==========================================
+
+class ValidationError(AppError):
+    """Fehler in der Business-Logik (unabhängig von Pydantic)"""
+    def __init__(self, message: str = "Ungültige Eingabedaten."):
+        super().__init__(message)

main.py

@@ -1,9 +1,57 @@
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, Request, status
+from fastapi.responses import JSONResponse
 from fastapi.templating import Jinja2Templates
+from exceptions import UserNotFoundError, UserAlreadyExistsError, UnauthorizedError, ForbiddenError, ValidationError
+import models #Database
+from routers import api_router #API
 
-app = FastAPI()
+app = FastAPI(title="Tandur API")
+app.include_router(api_router)
 templates = Jinja2Templates(directory="templates")
 
+######### ERROR handling #########
+
+# 1. Globale Regel für 404
+@app.exception_handler(UserNotFoundError)
+def user_not_found_handler(request: Request, exc: UserNotFoundError):
+    return JSONResponse(
+        status_code=status.HTTP_404_NOT_FOUND,
+        content={"detail": str(exc)},
+    )
+
+# 2. Globale Regel für 409
+@app.exception_handler(UserAlreadyExistsError)
+def user_exists_handler(request: Request, exc: UserAlreadyExistsError):
+    return JSONResponse(
+        status_code=status.HTTP_409_CONFLICT,
+        content={"detail": str(exc)},
+    )
+
+@app.exception_handler(UnauthorizedError)
+def unauthorized_handler(request: Request, exc: UnauthorizedError):
+    return JSONResponse(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        content={"detail": str(exc)},
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+@app.exception_handler(ForbiddenError)
+def permissions_handler(request: Request, exc: ForbiddenError):
+    return JSONResponse(
+        status_code=status.HTTP_403_FORBIDDEN,
+        content={"detail": str(exc)},
+    )
+
+@app.exception_handler(ValidationError)
+def permissions_handler(request: Request, exc: ValidationError):
+    return JSONResponse(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        content={"detail": str(exc)},
+    )
+
+
+
+
 @app.get("/")
 async def home(request: Request):
     return templates.TemplateResponse(request=request, name="index.html")

models/__init__.py

@@ -0,0 +1 @@
+from .user import User

models/user.py

@@ -0,0 +1,12 @@
+from sqlalchemy import Column, Integer, String, Boolean, DateTime
+from sqlalchemy.sql import func
+from database import Base
+
+class User(Base):
+    __tablename__ = "users"
+
+    id          = Column(Integer, primary_key=True, index=True, autoincrement=True)
+    email       = Column(String,  unique=True, index=True, nullable=False)
+    password    = Column(String,  nullable=False)
+    role        = Column(String,  default="user")
+    created_at  = Column(DateTime(timezone=True), server_default=func.now())

routers/__init__.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from .users import router as users_router
+from .auth import router as auth_router
+
+
+
+api_router = APIRouter()
+api_router.include_router(users_router, prefix="/api")
+api_router.include_router(auth_router, prefix="/api")

routers/auth.py

@@ -0,0 +1,40 @@
+# routers/auth.py
+from fastapi import APIRouter, Depends, status, Response
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
+
+from database import get_db
+from schemas.auth import LoginRequest
+from schemas.user import UserResponse
+#from exceptions import UserAlreadyExistsError, UserNotFoundError
+from crud.user import get_user_by_email
+from auth.security import verify_password, create_access_token
+
+
+router = APIRouter(prefix="/auth", tags=["Authentication"])
+
+
+
+
+@router.post("/login", response_model=UserResponse, status_code=status.HTTP_200_OK)
+def login(
+    data: LoginRequest,
+    response: Response,
+    db: Session = Depends(get_db)
+):
+    user = get_user_by_email(db, email=data.username)
+
+    if not user or not verify_password(data.password, user.password):
+        raise UnauthorizedError("Wrong username or password")
+
+    access_token = create_access_token(data={"sub": str(user.id)})
+
+    response.set_cookie(
+        key="access_token",
+        value=access_token,
+        httponly=True,  # Schutz gegen XSS
+        secure=False,  # Im local dev (HTTP) auf False, auf Prod (HTTPS) auf True
+        samesite="lax",
+    )
+
+    return user

routers/users.py

@@ -0,0 +1,59 @@
+# routers/users.py
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from database import get_db
+from schemas.user import UserCreate, UserResponse, UserUpdate
+from exceptions import UserAlreadyExistsError, UserNotFoundError, ForbiddenError
+from crud.user import get_users, get_user_by_id, get_user_by_email, create_user, update_user, delete_user
+from auth.dependencies import get_me
+
+router = APIRouter(prefix="/users", tags=["Users"])
+
+@router.get("/", response_model=list[UserResponse])
+def read_users(db: Session = Depends(get_db), me: User = Depends(get_me)):
+    if(me.role != "admin"):
+        raise ForbiddenError("Not authorized")
+    return get_users(db=db)
+
+@router.get("/me", response_model=UserResponse)
+def read_me(db: Session = Depends(get_db), me: User = Depends(get_me)):
+    return me
+
+@router.get("/by-email", response_model=UserResponse)
+def read_user_by_email(email: str, db: Session = Depends(get_db), me: User = Depends(get_me)):
+    return get_user_by_email(db=db, email=email)
+
+@router.get("/{user_id}", response_model=UserResponse)
+def read_user(user_id: int, db: Session = Depends(get_db), me: User = Depends(get_me)):
+    if(user_id != me.id and me.role != "admin"):
+        raise ForbiddenError("Not authorized")
+    return get_user_by_id(db=db, user_id=user_id)
+
+
+
+@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
+def register_user(user_data: UserCreate, db: Session = Depends(get_db), me: User = Depends(get_me)):
+    if(me.role != "admin"):
+        raise ForbiddenError("Not authorized")
+    return create_user(
+        db=db,
+        email=user_data.email,
+        password=user_data.password,
+        role=user_data.role
+    )
+
+@router.patch("/{user_id}", response_model=UserResponse, status_code=status.HTTP_200_OK)
+def patch_user(user_id: int, user_data: UserUpdate, db: Session = Depends(get_db), me: User = Depends(get_me)):
+    if(user_id != me.id and me.role != "admin"):
+        raise ForbiddenError("Not authorized")
+    update_dict = user_data.model_dump(exclude_unset=True)
+    return update_user(db=db, user_id=user_id, user_data=update_dict)
+
+@router.delete("/{user_id}", response_model=UserResponse)
+def del_user(user_id: int, db: Session = Depends(get_db), me: User = Depends(get_me)):
+    if(user_id != me.id and me.role != "admin"):
+        raise ForbiddenError("Not authorized")
+    return delete_user(db=db, user_id=user_id)
+
+

schemas/auth.py

@@ -0,0 +1,15 @@
+# /schemas/auth.py
+from pydantic import BaseModel
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
+
+class TokenData(BaseModel):
+    user_id: int | None = None
+
+
+class AuthResponse(BaseModel):
+    message: str

schemas/user.py

@@ -0,0 +1,19 @@
+from pydantic import BaseModel
+
+class UserCreate(BaseModel):
+    email: str
+    password: str
+    role: str | None = "user"
+
+class UserUpdate(BaseModel):
+    email: str | None = None
+    password: str | None = None
+    role: str | None = None
+
+class UserResponse(BaseModel):
+    id: int
+    email: str
+    role: str
+
+    class Config:
+        from_attributes = True  # Für SQLAlchemy 2.0 Kompatibilität

test.py

@@ -0,0 +1,36 @@
+# test_db.py (im Projekt-Root, NICHT in app/ falls main.py woanders liegt)
+from database import SessionLocal, engine, Base
+import models
+
+Base.metadata.create_all(bind=engine)  # falls Tabelle noch nicht existiert
+
+from crud.user import create_user, get_user_by_email, update_user, delete_user, verify_password
+
+db = SessionLocal()
+
+# Erstellen
+u = create_user(db, email="2test@example.com", password="test1234", role="admin")
+print("Erstellt:", u.id, u.email, u.role)
+
+# Lesen
+found = get_user_by_email(db, "1test@example.com")
+print(found)
+print("Gefunden:", found.email if found else "nicht gefunden")
+
+# Passwort prüfen
+print("Passwort korrekt:", verify_password("test1234", found.password))
+print("Falsches Passwort:", verify_password("falsch", found.password))
+
+# Updaten
+updated = update_user(db, u.id, email="neu@example.com")
+print("Aktualisiert:", updated.email)
+
+# Löschen
+success = delete_user(db, u.id)
+print("Gelöscht:", success)
+
+# Prüfen, ob wirklich weg
+gone = get_user_by_email(db, "neu@example.com")
+print("Nach Löschen noch da?:", gone)
+
+db.close()