Coverage for core\auth.py: 31%

1import jwt 

2from datetime import datetime, timedelta, timezone 

3from django.conf import settings 

4from ninja.security import HttpBearer 

5from django.contrib.auth import get_user_model 

6 

7User = get_user_model() 

8 

9def create_access_token(user_id: str, organization_id: str = None) -> str: 

10 """Create a short-lived JWT access token.""" 

11 expire = datetime.now(timezone.utc) + timedelta(minutes=settings.JWT_ACCESS_TOKEN_LIFETIME_MINUTES) 

12 payload = { 

13 'sub': str(user_id), 

14 'org_id': str(organization_id) if organization_id else None, 

15 'exp': expire, 

16 'iat': datetime.now(timezone.utc), 

17 'type': 'access' 

18 } 

19 return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) 

20 

21def create_refresh_token(user_id: str) -> str: 

22 """Create a long-lived JWT refresh token.""" 

23 expire = datetime.now(timezone.utc) + timedelta(days=settings.JWT_REFRESH_TOKEN_LIFETIME_DAYS) 

24 payload = { 

25 'sub': str(user_id), 

26 'exp': expire, 

27 'iat': datetime.now(timezone.utc), 

28 'type': 'refresh' 

29 } 

30 return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) 

31 

32def decode_token(token: str) -> dict: 

33 """Decode and validate a JWT token.""" 

34 try: 

35 payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) 

36 return payload 

37 except jwt.ExpiredSignatureError: 

38 return None 

39 except jwt.InvalidTokenError: 

40 return None 

41 

42class JWTAuth(HttpBearer): 

43 """Django Ninja Bearer token authentication.""" 

44 def authenticate(self, request, token): 

45 payload = decode_token(token) 

46 if not payload or payload.get('type') != 'access': 

47 return None 

48 

49 user_id = payload.get('sub') 

50 try: 

51 user = User.objects.get(id=user_id, is_active=True) 

52 request.user = user 

53 request.tenant_id = payload.get('org_id') 

54 return user 

55 except User.DoesNotExist: 

56 return None