Coverage for sm/api/authentication.py: 95%

1from datetime import timedelta 

2 

3from django.contrib.auth.hashers import check_password 

4from django.utils import timezone 

5from drf_spectacular.extensions import OpenApiAuthenticationExtension 

6from rest_framework.authentication import BaseAuthentication, get_authorization_header 

7from rest_framework.exceptions import AuthenticationFailed 

8 

9from sm.models import ApiKey 

10 

11MALFORMED_MESSAGE = ( 

12 "Invalid Authorization header. Expected format: " "ApiKey <client_id>:<secret>" 

13) 

14 

15 

16class ApiKeyAuthentication(BaseAuthentication): 

17 """ 

18 Authenticates requests using a client_id/secret API key pair. 

19 

20 The credentials are supplied in the Authorization header: 

21 

22 Authorization: ApiKey <client_id>:<secret> 

23 

24 A successful authentication returns the user the key belongs to, so all 

25 further permission and multi-tenancy checks behave exactly as they would 

26 for that user. 

27 """ 

28 

29 keyword = "ApiKey" 

30 

31 def authenticate(self, request): 

32 auth = get_authorization_header(request).split() 

33 if not auth or auth[0].lower() != self.keyword.lower().encode(): 

34 return None 

35 

36 if len(auth) != 2: 

37 raise AuthenticationFailed(MALFORMED_MESSAGE) 

38 

39 token = auth[1].decode("utf-8") 

40 if ":" not in token: 

41 raise AuthenticationFailed(MALFORMED_MESSAGE) 

42 

43 client_id, secret = token.split(":", 1) 

44 

45 key = ( 

46 ApiKey.objects.select_related("user") 

47 .filter(client_id=client_id, is_active=True) 

48 .first() 

49 ) 

50 if key is None or not check_password(secret, key.secret_hash): 

51 raise AuthenticationFailed("Invalid API key credentials.") 

52 

53 if key.is_expired(): 

54 raise AuthenticationFailed("API key has expired.") 

55 

56 if not key.user.is_active: 

57 raise AuthenticationFailed("User account is disabled.") 

58 

59 # Only write on a change to avoid a DB update on every single request. 

60 if not key.last_used_at or timezone.now() - key.last_used_at > timedelta( 

61 minutes=5 

62 ): 

63 ApiKey.objects.filter(pk=key.pk).update(last_used_at=timezone.now()) 

64 

65 return (key.user, key) 

66 

67 def authenticate_header(self, request): 

68 return self.keyword 

69 

70 

71class ApiKeyAuthenticationScheme(OpenApiAuthenticationExtension): 

72 target_class = ApiKeyAuthentication 

73 name = "ApiKeyAuth" 

74 

75 def get_security_requirement(self, auto_schema): 

76 return {self.name: []} 

77 

78 def get_security_definition(self, auto_schema): 

79 return { 

80 "type": "apiKey", 

81 "in": "header", 

82 "name": "Authorization", 

83 "description": ( 

84 "Client ID/secret pair. Send the credentials as " 

85 "`ApiKey <client_id>:<secret>` in the Authorization header." 

86 ), 

87 }