Coverage for app\deps.py: 91%

1import os 

2import jwt 

3from fastapi import Depends, HTTPException 

4from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials 

5from sqlalchemy.orm import Session 

6from .db import SessionLocal, User, UserStatus 

7 

8# Configuration 

9AUTH_SECRET = os.getenv("ADMIN_CORE_JWT_SECRET", "dev_admin_secret_change_me") 

10security = HTTPBearer(auto_error=False) 

11 

12def get_db(): 

13 db = SessionLocal() 

14 try: 

15 yield db 

16 finally: 

17 db.close() 

18 

19def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db)): 

20 if credentials is None: 

21 raise HTTPException(status_code=401, detail="Not authenticated") 

22 token = credentials.credentials 

23 try: 

24 payload = jwt.decode(token, AUTH_SECRET, algorithms=["HS256"], audience="admin-web") 

25 user_id = payload.get("sub") 

26 if not user_id: 

27 raise HTTPException(status_code=401, detail="Invalid token") 

28 if payload.get("iss") != "admin-core": 

29 raise HTTPException(status_code=401, detail="Invalid issuer") 

30 

31 # Support legacy tokens with 'sub' = 'admin'/'operator' 

32 user = db.query(User).filter(User.id == user_id).first() 

33 if not user: 

34 user = db.query(User).filter(User.username == str(user_id)).first() 

35 

36 if not user or user.status != UserStatus.ACTIVE: 

37 raise HTTPException(status_code=401, detail="User not found or inactive") 

38 

39 return { 

40 "id": user.id, 

41 "username": user.username, 

42 "email": user.email, 

43 "full_name": user.full_name, 

44 "role": user.role.value, 

45 "status": user.status.value, 

46 "last_login_at": user.last_login_at, 

47 } 

48 except jwt.PyJWTError: 

49 raise HTTPException(status_code=401, detail="Invalid token")