bede/tfm_ainventory
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: implementare JWT Bearer authentication pe toți routers [C-01]

Browse Source 
Implementare completă a autentificării Bearer token:
- Creiez backend/auth.py: funcții JWT (create_access_token, get_current_user, get_current_admin)
- Modific /users/login: returnează TokenResponse cu JWT token și expirare 8h
- Adaug Depends(get_current_user) pe toate endpoint-urile API
- [M-02] user_id extras din JWT token, nu din request body
- [L-01] Token cu exp claim pentru sesiuni frontend

Acces endpoints-uri:
- GET/POST /users/: authenticated users
- POST /users/: admin only
- PUT /users/{id}, DELETE /users/{id}, /ldap-config, /test-ldap: admin only
- Toți routers (items, operations, categories): authenticated users minimum

Modificări dependențe:
- Adaug: python-jose[cryptography]>=3.3.0, slowapi>=0.1.9 (pentru H-02 rate limiting)

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Bedeleanu
2026-04-11 13:38:24 +03:00
parent 247ea45408
commit 9b6adad618
7 changed files with 338 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
backend/routers/users.py
View File

@@ -7,7 +7,7 @@ import ldap3
from ldap3.utils.conv import escape_filter_chars
import json
import os
from .. import models, schemas, database
from .. import models, schemas, database, auth
from ..logger import log
router = APIRouter(prefix="/users", tags=["users"])
@@ -110,7 +110,11 @@ def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
@router.get("/", response_model=List[schemas.User])
def get_users(db: Session = Depends(get_db)):
def get_users(
db: Session = Depends(get_db),
current_user: auth.TokenData = Depends(auth.get_current_user)
):
"""[C-01] Lista utilizatori — doar pentru useri autentificati."""
users = db.query(models.User).all()
# Auto-seed if empty
if not users:
@@ -130,11 +134,16 @@ def get_users(db: Session = Depends(get_db)):
return users
@router.post("/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
def create_user(
user: schemas.UserCreate,
db: Session = Depends(get_db),
current_user: auth.TokenData = Depends(auth.get_current_admin)
):
"""[C-01] Creare utilizator — doar admin."""
existing = db.query(models.User).filter(models.User.username == user.username).first()
if existing:
raise HTTPException(status_code=400, detail="Username already exists")
hashed = get_password_hash(user.password) if user.password else None
new_user = models.User(username=user.username, role=user.role, origin="local", hashed_password=hashed)
db.add(new_user)
@@ -142,10 +151,13 @@ def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db.refresh(new_user)
return new_user
@router.post("/login")
@router.post("/login", response_model=schemas.TokenResponse)
def login(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
"""
[C-01] Login endpoint: validează credențiale și returnează JWT token Bearer.
"""
user = db.query(models.User).filter(models.User.username == form_data.username).first()
# Try local authentication
authenticated = False
if user and user.hashed_password:
@@ -155,7 +167,7 @@ def login(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
# [SECURITY FIX C-02] Bypass-ul pentru utilizatori fără parolă a fost eliminat.
# Utilizatorii LDAP trebuie să se autentifice prin fluxul LDAP de mai jos.
pass
# If local failed, try LDAP
if not authenticated:
ldap_role = authenticate_ldap(form_data.username, form_data.password)
@@ -163,12 +175,12 @@ def login(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
authenticated = True
# Cache hash for offline support
new_hash = get_password_hash(form_data.password)
# If user doesn't exist locally, create a stub for role management
if not user:
user = models.User(
username=form_data.username,
role=ldap_role,
username=form_data.username,
role=ldap_role,
origin="ldap",
hashed_password=new_hash
)
@@ -182,49 +194,79 @@ def login(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
db.commit()
db.refresh(user)
else:
raise HTTPException(status_code=400, detail="Invalid username or password, or insufficient permissions")
return user
raise HTTPException(status_code=401, detail="Invalid username or password, or insufficient permissions")
if not authenticated or not user:
raise HTTPException(status_code=401, detail="Invalid username or password")
# [C-01] Generare JWT token
token = auth.create_access_token(
user_id=user.id,
username=user.username,
role=user.role
)
return schemas.TokenResponse(
access_token=token,
token_type="bearer",
user_id=user.id,
username=user.username,
role=user.role
)
@router.put("/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user_update: schemas.UserUpdate, db: Session = Depends(get_db)):
def update_user(
user_id: int,
user_update: schemas.UserUpdate,
db: Session = Depends(get_db),
current_user: auth.TokenData = Depends(auth.get_current_admin)
):
"""[C-01] Actualizare utilizator — doar admin."""
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if not db_user:
raise HTTPException(status_code=404, detail="User not found")
if user_update.username and db_user.username == "Admin" and user_update.username != "Admin":
raise HTTPException(status_code=400, detail="Cannot change Admin username")
raise HTTPException(status_code=400, detail="Cannot change Admin username")
if user_update.username:
# Check if username already taken by another user
existing = db.query(models.User).filter(models.User.username == user_update.username, models.User.id != user_id).first()
if existing:
raise HTTPException(status_code=400, detail="Username already exists")
db_user.username = user_update.username
if user_update.password:
db_user.hashed_password = get_password_hash(user_update.password)
if user_update.role:
db_user.role = user_update.role
db.commit()
db.refresh(db_user)
return db_user
@router.get("/ldap-config")
def get_ldap_settings():
def get_ldap_settings(current_user: auth.TokenData = Depends(auth.get_current_admin)):
"""[C-01] Obține config LDAP — doar admin."""
return get_ldap_config()
@router.post("/ldap-config")
def update_ldap_settings(config: dict):
def update_ldap_settings(
config: dict,
current_user: auth.TokenData = Depends(auth.get_current_admin)
):
"""[C-01] Actualizează config LDAP — doar admin."""
config_path = os.path.join(database.DATA_DIR, "ldap_config.json")
with open(config_path, "w") as f:
json.dump(config, f)
return {"message": "Config saved"}
@router.post("/test-ldap")
def test_ldap_connection(config: dict):
def test_ldap_connection(
config: dict,
current_user: auth.TokenData = Depends(auth.get_current_admin)
):
import socket
try:
# Extract host and port
@@ -273,14 +315,19 @@ def test_ldap_connection(config: dict):
return {"status": "error", "message": f"Network Error: {str(e)}"}
@router.delete("/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
def delete_user(
user_id: int,
db: Session = Depends(get_db),
current_user: auth.TokenData = Depends(auth.get_current_admin)
):
"""[C-01] Ștergere utilizator — doar admin."""
user = db.query(models.User).filter(models.User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
if user.username == "Admin":
raise HTTPException(status_code=400, detail="Cannot delete default Admin")
raise HTTPException(status_code=400, detail="Cannot delete default Admin")
db.delete(user)
db.commit()
return {"message": "User deleted"}