nouveau Back

main.py

@@ -1,58 +1,92 @@
-# backend.py
-from fastapi import FastAPI, Request, Depends, HTTPException
-from fastapi.responses import HTMLResponse, RedirectResponse
-from fastapi.staticfiles import StaticFiles
-from fastapi.templating import Jinja2Templates
+from fastapi import FastAPI, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from pymongo import MongoClient
-from starlette.middleware.sessions import SessionMiddleware
+from pymongo.errors import DuplicateKeyError
+from bson import ObjectId
+from datetime import datetime, timedelta
+import jwt
+from passlib.context import CryptContext
 
 app = FastAPI()
-app.mount("/static", StaticFiles(directory="static"), name="static")
-app.add_middleware(SessionMiddleware, secret_key="votre_secret_key_secure")
-
-# Configuration MongoDB
-client = MongoClient("mongodb+srv://lebaykserver:mVef8e5F1GRtbA14@test.6bzxp.mongodb.net/")
-db = client["auth_demo"]
-users = db["users"]
-establishments = db["establishments"]
-
-# Créer quelques données de test
-if establishments.count_documents({}) == 0:
-    establishments.insert_many([{"name": "École 1"}, {"name": "Université A"}, {"name": "Lycée B"}])
-
-@app.get("/", response_class=HTMLResponse)
-async def root(request: Request):
-    if not request.session.get("user"):
-        return RedirectResponse(url="/login")
-    return templates.TemplateResponse("dashboard.html", {"request": request})
-
-@app.get("/login", response_class=HTMLResponse)
-async def login_page(request: Request):
-    if request.session.get("user"):
-        return RedirectResponse(url="/")
-    return templates.TemplateResponse("login.html", {"request": request})
-
-@app.get("/api/establishments")
-async def get_establishments():
-    return [e["name"] for e in establishments.find({}, {"_id": 0, "name": 1})]
-
-@app.get("/api/login")
-async def login(request: Request, username: str, password: str, establishment: str):
-    user = users.find_one({
-        "username": username,
-        "password": password,
-        "establishment": establishment
-    })
+
+# Configuration de la base de données MongoDB
+client = MongoClient("mongodb://localhost:27017/")
+db = client["user_db"]
+users_collection = db["users"]
+
+# Configuration de la sécurité
+SECRET_KEY = "your_secret_key"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+# Fonction pour vérifier le mot de passe
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+# Fonction pour hasher le mot de passe
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+# Fonction pour créer un token JWT
+def create_access_token(data: dict, expires_delta: timedelta = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+# Fonction pour authentifier un utilisateur
+def authenticate_user(email: str, password: str):
+    user = users_collection.find_one({"email": email})
     if not user:
-        raise HTTPException(status_code=401, detail="Identifiants incorrects")
-    request.session["user"] = username
-    return {"status": "success"}
-
-@app.get("/api/logout")
-async def logout(request: Request):
-    request.session.pop("user", None)
-    return RedirectResponse(url="/login")
-
-@app.get("/api/check_session")
-async def check_session(request: Request):
-    return {"authenticated": bool(request.session.get("user"))}
+        return False
+    if not verify_password(password, user["hashed_password"]):
+        return False
+    return user
+
+# Route pour s'inscrire
+@app.post("/register")
+async def register(email: str, password: str):
+    hashed_password = get_password_hash(password)
+    try:
+        users_collection.insert_one({"email": email, "hashed_password": hashed_password})
+        return {"message": "User created successfully"}
+    except DuplicateKeyError:
+        raise HTTPException(status_code=400, detail="Email already registered")
+
+# Route pour se connecter
+@app.post("/token")
+async def login(form_data: OAuth2PasswordRequestForm = Depends()):
+    user = authenticate_user(form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            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
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+# Route protégée
+@app.get("/protected")
+async def protected_route(token: str = Depends(oauth2_scheme)):
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        email = payload.get("sub")
+        if email is None:
+            raise HTTPException(status_code=400, detail="Invalid token")
+        user = users_collection.find_one({"email": email})
+        if user is None:
+            raise HTTPException(status_code=404, detail="User not found")
+        return {"message": "You are authenticated", "user": email}
+    except jwt.PyJWTError:
+        raise HTTPException(status_code=401, detail="Invalid token")

requirements.txt

@@ -1,4 +1,3 @@
 fastapi
-uvicorn[standard]
-pymongo
-python-multipart
+uvicorn
+pymongo