authentification version0

src/auth_project/__init__.py

@@ -0,0 +1,2 @@
+def main() -> None:
+    print("Hello from auth-project!")

src/auth_project/api/deps.py

@@ -0,0 +1,72 @@
+from typing import Annotated
+
+import jwt
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy.orm import Session
+
+from src.auth_project.core.config import settings
+from src.auth_project.db.session import get_db
+from src.auth_project.models.user import User
+from src.auth_project.repositories.user_repository import get_by_id
+
+
+bearer_scheme = HTTPBearer(auto_error=False)
+
+
+def get_current_user(
+    credentials: Annotated[
+        HTTPAuthorizationCredentials | None,
+        Depends(bearer_scheme),
+    ],
+    db: Annotated[Session, Depends(get_db)],
+) -> User:
+    if credentials is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Authentification requise.",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    try:
+        payload = jwt.decode(
+            credentials.credentials,
+            settings.jwt_secret_key,
+            algorithms=[settings.jwt_algorithm],
+            options={"require": ["sub", "iat", "exp"]},
+        )
+
+        user_id = int(payload["sub"])
+
+    except (jwt.InvalidTokenError, ValueError, TypeError, KeyError):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token invalide ou expiré.",
+            headers={"WWW-Authenticate": "Bearer"},
+        ) from None
+
+    user = get_by_id(db, user_id)
+
+    if user is None or not user.is_active:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Compte introuvable ou désactivé.",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    return user
+
+
+def require_roles(*allowed_roles: str):
+    def role_checker(
+        current_user: Annotated[User, Depends(get_current_user)],
+    ) -> User:
+        if current_user.role not in allowed_roles:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Permissions insuffisantes.",
+            )
+
+        return current_user
+
+    return role_checker

src/auth_project/api/routes/admin_users.py

@@ -0,0 +1,110 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from src.auth_project.api.deps import require_roles
+from src.auth_project.core.security import hash_password
+from src.auth_project.db.session import get_db
+from src.auth_project.models.user import User
+from src.auth_project.repositories.user_repository import create_user, get_by_id, get_by_email, get_users
+from src.auth_project.schemas.user import UserCreate, UserRead, UserRoleUpdate, UserStatusUpdate
+
+
+router = APIRouter(prefix="/admin/users", tags=["User management"])
+
+SuperAdmin = Annotated[User, Depends(require_roles("super_admin"))]
+
+
+@router.post(
+    "/create_account",
+    response_model=UserRead,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_account(
+    data: UserCreate,
+    db: Annotated[Session, Depends(get_db)],
+    _: SuperAdmin,
+) -> User:
+    email = str(data.email).strip().lower()
+
+    if get_by_email(db, email) is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Cette adresse email est déjà utilisée.",
+        )
+
+    try:
+        return create_user(
+            db,
+            email=email,
+            hashed_password=hash_password(data.password),
+            role=data.role,
+        )
+    except IntegrityError:
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Cette adresse email est déjà utilisée.",
+        ) from None
+
+
+@router.get("/list_accounts", response_model=list[UserRead])
+def list_accounts(
+    db: Annotated[Session, Depends(get_db)],
+    _: SuperAdmin,
+    skip: int = Query(default=0, ge=0),
+    limit: int = Query(default=50, ge=1, le=100),
+) -> list[User]:
+    return get_users(db, skip=skip, limit=limit)
+
+
+@router.patch("/{user_id}/role", response_model=UserRead)
+def update_role(
+    user_id: int,
+    data: UserRoleUpdate,
+    db: Annotated[Session, Depends(get_db)],
+    current_admin: SuperAdmin,
+) -> User:
+    target = get_by_id(db, user_id)
+
+    if target is None:
+        raise HTTPException(status_code=404, detail="Utilisateur introuvable.")
+
+    if target.role == "super_admin" or target.id == current_admin.id:
+        raise HTTPException(
+            status_code=403,
+            detail="Modification de ce rôle interdite.",
+        )
+
+    target.role = data.role
+    db.commit()
+    db.refresh(target)
+
+    return target
+
+
+@router.patch("/{user_id}/status", response_model=UserRead)
+def update_status(
+    user_id: int,
+    data: UserStatusUpdate,
+    db: Annotated[Session, Depends(get_db)],
+    current_admin: SuperAdmin,
+) -> User:
+    target = get_by_id(db, user_id)
+
+    if target is None:
+        raise HTTPException(status_code=404, detail="Utilisateur introuvable.")
+
+    if target.id == current_admin.id or target.role == "super_admin":
+        raise HTTPException(
+            status_code=403,
+            detail="Modification du statut de ce compte interdite.",
+        )
+
+    target.is_active = data.is_active
+    db.commit()
+    db.refresh(target)
+
+    return target

src/auth_project/api/routes/auth.py

@@ -0,0 +1,38 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from src.auth_project.api.deps import get_current_user
+from src.auth_project.db.session import get_db
+from src.auth_project.models.user import User
+from src.auth_project.schemas.auth import LoginRequest, TokenResponse
+from src.auth_project.schemas.user import UserRead
+from src.auth_project.services.auth_service import authenticate_user
+
+
+router = APIRouter(prefix="/auth", tags=["Authentication"])
+
+
+@router.post("/login", response_model=TokenResponse)
+def login(
+    data: LoginRequest,
+    db: Annotated[Session, Depends(get_db)],
+) -> TokenResponse:
+    result = authenticate_user(db, data.email, data.password)
+
+    if result is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Email ou mot de passe incorrect.",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    return result
+
+
+@router.get("/me", response_model=UserRead)
+def get_me(
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> User:
+    return current_user

src/auth_project/cli/create_super_admin.py

@@ -0,0 +1,58 @@
+from getpass import getpass
+
+from sqlalchemy import select
+
+from src.auth_project.core.security import hash_password
+from src.auth_project.db.base import Base
+from src.auth_project.db.session import SessionLocal, engine
+from src.auth_project.models.user import User
+
+
+def main() -> None:
+    # Utile pour ce premier démarrage local.
+    Base.metadata.create_all(bind=engine)
+
+    with SessionLocal() as db:
+        existing = db.scalar(
+            select(User.id).where(User.role == "super_admin").limit(1)
+        )
+
+        if existing is not None:
+            print("Un super administrateur existe déjà.")
+            return
+
+        email = input("Email du super administrateur : ").strip().lower()
+        password = getpass("Mot de passe (12 caractères minimum) : ")
+        confirmation = getpass("Confirme le mot de passe : ")
+
+        if len(password) < 12:
+            print("Le mot de passe doit contenir au moins 12 caractères.")
+            return
+
+        if password != confirmation:
+            print("Les mots de passe ne correspondent pas.")
+            return
+
+        duplicate = db.scalar(
+            select(User.id).where(User.email == email)
+        )
+
+        if duplicate is not None:
+            print("Cette adresse email existe déjà.")
+            return
+
+        admin = User(
+            email=email,
+            hashed_password=hash_password(password),
+            role="super_admin",
+            is_active=True,
+        )
+
+        db.add(admin)
+        db.commit()
+
+        print("Super administrateur créé avec succès.")
+
+
+if __name__ == "__main__":
+    main()

src/auth_project/core/config.py

@@ -0,0 +1,17 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    database_url: str
+    jwt_secret_key: str
+    jwt_algorithm: str = "HS256"
+    access_token_expire_minutes: int = 30
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+
+settings = Settings()

src/auth_project/core/security.py

@@ -0,0 +1,40 @@
+from datetime import datetime, timedelta, timezone
+
+import jwt
+from pwdlib import PasswordHash
+
+from src.auth_project.core.config import settings
+
+
+password_hash = PasswordHash.recommended()
+
+
+def hash_password(password: str) -> str:
+    """Transforme le mot de passe en hash sécurisé."""
+    return password_hash.hash(password)
+
+
+def verify_password(password: str, hashed_password: str) -> bool:
+    """Vérifie un mot de passe sans stocker sa valeur en clair."""
+    return password_hash.verify(password, hashed_password)
+
+
+def create_access_token(user_id: int, role: str) -> str:
+    """Crée un token JWT à durée de vie limitée."""
+    now = datetime.now(timezone.utc)
+    expires_at = now + timedelta(
+        minutes=settings.access_token_expire_minutes
+    )
+
+    payload = {
+        "sub": str(user_id),
+        "role": role,
+        "iat": now,
+        "exp": expires_at,
+    }
+
+    return jwt.encode(
+        payload,
+        settings.jwt_secret_key,
+        algorithm=settings.jwt_algorithm,
+    )

src/auth_project/db/base.py

@@ -0,0 +1,5 @@
+from sqlalchemy.orm import DeclarativeBase
+
+
+class Base(DeclarativeBase):
+    pass

src/auth_project/db/session.py

@@ -0,0 +1,27 @@
+from collections.abc import Generator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from src.auth_project.core.config import settings
+
+
+engine = create_engine(
+    settings.database_url,
+    pool_pre_ping=True,
+)
+
+SessionLocal = sessionmaker(
+    bind=engine,
+    class_=Session,
+    autoflush=False,
+    expire_on_commit=False,
+)
+
+
+def get_db() -> Generator[Session, None, None]:
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

src/auth_project/main.py

@@ -0,0 +1,35 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from src.auth_project.db.base import Base
+from src.auth_project.db.session import engine
+from src.auth_project.models.user import User  # Enregistre le modèle auprès de SQLAlchemy.
+from src.auth_project.api.routes.auth import router as auth_router
+from src.auth_project.api.routes.admin_users import router as admin_users_router
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Création simple des tables pour le développement local.
+    Base.metadata.create_all(bind=engine)
+    yield
+
+
+app = FastAPI(
+    title="Authentication API",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+app.include_router(auth_router)
+app.include_router(admin_users_router)
+
+@app.get("/")
+def accueil():
+    return {"message": "Bienvenue sur mon API"}
+
+    
+@app.get("/health", tags=["Health"])
+def health_check() -> dict[str, str]:
+    return {"status": "ok"}

src/auth_project/models/user.py

@@ -0,0 +1,50 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import Boolean, DateTime, Integer, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.auth_project.db.base import Base
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(
+        Integer,
+        primary_key=True,
+        index=True,
+    )
+
+    email: Mapped[str] = mapped_column(
+        String(320),
+        unique=True,
+        index=True,
+        nullable=False,
+    )
+
+    hashed_password: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+    )
+
+    role: Mapped[str] = mapped_column(
+        String(30),
+        nullable=False,
+        default="user",
+    )
+
+    is_active: Mapped[bool] = mapped_column(
+        Boolean,
+        nullable=False,
+        default=True,
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=utc_now,
+    )

src/auth_project/repositories/user_repository.py

@@ -0,0 +1,52 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from src.auth_project.models.user import User
+
+
+def get_by_email(db: Session, email: str) -> User | None:
+    normalized_email = email.strip().lower()
+
+    statement = select(User).where(User.email == normalized_email)
+    return db.scalar(statement)
+
+
+def get_by_id(db: Session, user_id: int) -> User | None:
+    return db.get(User, user_id)
+
+
+def create_user(
+    db: Session,
+    *,
+    email: str,
+    hashed_password: str,
+    role: str,
+) -> User:
+    user = User(
+        email=email.strip().lower(),
+        hashed_password=hashed_password,
+        role=role,
+        is_active=True,
+    )
+
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+
+    return user
+
+
+def get_users(
+    db: Session,
+    *,
+    skip: int = 0,
+    limit: int = 50,
+) -> list[User]:
+    statement = (
+        select(User)
+        .order_by(User.id)
+        .offset(skip)
+        .limit(limit)
+    )
+
+    return list(db.scalars(statement).all())

src/auth_project/schemas/auth.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, EmailStr, Field
+
+
+class LoginRequest(BaseModel):
+    email: EmailStr
+    password: str = Field(min_length=1)
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"

src/auth_project/schemas/user.py

@@ -0,0 +1,33 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, EmailStr, Field
+
+
+UserRole = Literal["super_admin", "candidat", "student", "user", "prof", "alumni","entreprise"]
+CreatableRole = Literal["candidat", "student", "user", "prof", "alumni","entreprise"]
+
+
+class UserCreate(BaseModel):
+    email: EmailStr
+    password: str = Field(min_length=12, max_length=128)
+    role: CreatableRole
+
+#we cannot create a super_admin ! => differnce between UserRole and CreatableRole
+
+class UserRead(BaseModel):
+    id: int
+    email: EmailStr
+    role: UserRole
+    is_active: bool
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class UserRoleUpdate(BaseModel):
+    role: CreatableRole
+
+
+class UserStatusUpdate(BaseModel):
+    is_active: bool

src/auth_project/services/auth_service.py

@@ -0,0 +1,30 @@
+from sqlalchemy.orm import Session
+
+from src.auth_project.core.security import (
+    create_access_token,
+    verify_password,
+)
+from src.auth_project.repositories.user_repository import get_by_email
+from src.auth_project.schemas.auth import TokenResponse
+
+
+def authenticate_user(
+    db: Session,
+    email: str,
+    password: str,
+) -> TokenResponse | None:
+    user = get_by_email(db, email)
+
+    # Même réponse si l'email ou le mot de passe est incorrect.
+    if user is None:
+        return None
+
+    if not verify_password(password, user.hashed_password):
+        return None
+
+    if not user.is_active:
+        return None
+
+    token = create_access_token(user.id, user.role)
+
+    return TokenResponse(access_token=token)