Coverage for netbox/api/authentication.py: 64%

1import logging 

2 

3from django.conf import settings 

4from django.utils import timezone 

5from drf_spectacular.extensions import OpenApiAuthenticationExtension 

6from rest_framework import exceptions 

7from rest_framework.authentication import BaseAuthentication, get_authorization_header 

8from rest_framework.permissions import SAFE_METHODS, BasePermission, DjangoObjectPermissions 

9 

10from netbox.config import get_config 

11from users.constants import TOKEN_PREFIX 

12from users.models import Token 

13from utilities.request import get_client_ip 

14 

15V1_KEYWORD = 'Token' 

16V2_KEYWORD = 'Bearer' 

17 

18 

19class TokenAuthentication(BaseAuthentication): 

20 """ 

21 A custom authentication scheme which enforces Token expiration times and source IP restrictions. 

22 """ 

23 model = Token 

24 

25 def authenticate(self, request): 

26 # Authorization header is not present; ignore 

27 if not (auth := get_authorization_header(request).split()): 

28 return None 

29 # Unrecognized header; ignore 

30 if auth[0].lower() not in (V1_KEYWORD.lower().encode(), V2_KEYWORD.lower().encode()): 

31 return None 

32 # Check for extraneous token content 

33 if len(auth) != 2: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 raise exceptions.AuthenticationFailed( 

35 'Invalid authorization header: Must be in the form "Bearer <key>.<token>" or "Token <token>"' 

36 ) 

37 # Extract the key (if v2) & token plaintext from the auth header 

38 try: 

39 auth_value = auth[1].decode() 

40 except UnicodeError: 

41 raise exceptions.AuthenticationFailed('Invalid authorization header: Token contains invalid characters') 

42 

43 # Infer token version from presence or absence of prefix 

44 version = 2 if auth_value.startswith(TOKEN_PREFIX) else 1 

45 

46 if version == 1: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 key, plaintext = None, auth_value 

48 else: 

49 auth_value = auth_value.removeprefix(TOKEN_PREFIX) 

50 try: 

51 key, plaintext = auth_value.split('.', 1) 

52 except ValueError: 

53 raise exceptions.AuthenticationFailed( 

54 "Invalid authorization header: Could not parse key from v2 token. Did you mean to use 'Token' " 

55 "instead of 'Bearer'?" 

56 ) 

57 

58 # Look for a matching token in the database 

59 try: 

60 qs = Token.objects.prefetch_related('user') 

61 if version == 1: 61 ↛ 63line 61 didn't jump to line 63 because the condition on line 61 was never true

62 # Fetch v1 token by querying plaintext value directly 

63 token = qs.get(version=version, plaintext=plaintext) 

64 else: 

65 # Fetch v2 token by key, then validate the plaintext 

66 token = qs.get(version=version, key=key) 

67 if not token.validate(plaintext): 67 ↛ 69line 67 didn't jump to line 69 because the condition on line 67 was never true

68 # Key is valid but plaintext is not. Raise DoesNotExist to guard against key enumeration. 

69 raise Token.DoesNotExist() 

70 except Token.DoesNotExist: 

71 raise exceptions.AuthenticationFailed(f"Invalid v{version} token") 

72 

73 # Enforce source IP restrictions (if any) set on the token 

74 if token.allowed_ips: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 client_ip = get_client_ip(request) 

76 if client_ip is None: 

77 raise exceptions.AuthenticationFailed( 

78 'Client IP address could not be determined for validation. Check that the HTTP server is ' 

79 'correctly configured to pass the required header(s).' 

80 ) 

81 if not token.validate_client_ip(client_ip): 

82 raise exceptions.AuthenticationFailed( 

83 f"Source IP {client_ip} is not permitted to authenticate using this token." 

84 ) 

85 

86 # Enforce the Token is enabled 

87 if not token.enabled: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 raise exceptions.AuthenticationFailed('Token disabled') 

89 

90 # Enforce the Token's expiration time, if one has been set. 

91 if token.is_expired: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 raise exceptions.AuthenticationFailed('Token expired') 

93 

94 # Update last used, but only once per minute at most. This reduces write load on the database 

95 if not token.last_used or (timezone.now() - token.last_used).total_seconds() > 60: 

96 # If maintenance mode is enabled, assume the database is read-only, and disable updating the token's 

97 # last_used time upon authentication. 

98 if get_config().MAINTENANCE_MODE: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 logger = logging.getLogger('netbox.auth.login') 

100 logger.debug("Maintenance mode enabled: Disabling update of token's last used timestamp") 

101 else: 

102 Token.objects.filter(pk=token.pk).update(last_used=timezone.now()) 

103 

104 user = token.user 

105 

106 # When LDAP authentication is active try to load user data from LDAP directory 

107 if 'netbox.authentication.LDAPBackend' in settings.REMOTE_AUTH_BACKEND: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 from netbox.authentication import LDAPBackend 

109 ldap_backend = LDAPBackend() 

110 

111 # Load from LDAP if FIND_GROUP_PERMS is active 

112 # Always query LDAP when user is not active, otherwise it is never activated again 

113 if ldap_backend.settings.FIND_GROUP_PERMS or not token.user.is_active: 

114 ldap_user = ldap_backend.populate_user(token.user.username) 

115 # If the user is found in the LDAP directory use it, if not fallback to the local user 

116 if ldap_user: 

117 user = ldap_user 

118 

119 if not user.is_active: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

120 raise exceptions.AuthenticationFailed("User inactive") 

121 

122 return user, token 

123 

124 

125class TokenPermissions(DjangoObjectPermissions): 

126 """ 

127 Custom permissions handler which extends the built-in DjangoModelPermissions to validate a Token's write ability 

128 for unsafe requests (POST/PUT/PATCH/DELETE). 

129 """ 

130 # Override the stock perm_map to enforce view permissions 

131 perms_map = { 

132 'GET': ['%(app_label)s.view_%(model_name)s'], 

133 'OPTIONS': [], 

134 'HEAD': ['%(app_label)s.view_%(model_name)s'], 

135 'POST': ['%(app_label)s.add_%(model_name)s'], 

136 'PUT': ['%(app_label)s.change_%(model_name)s'], 

137 'PATCH': ['%(app_label)s.change_%(model_name)s'], 

138 'DELETE': ['%(app_label)s.delete_%(model_name)s'], 

139 } 

140 

141 def __init__(self): 

142 

143 # LOGIN_REQUIRED determines whether read-only access is provided to anonymous users. 

144 self.authenticated_users_only = settings.LOGIN_REQUIRED 

145 

146 super().__init__() 

147 

148 def _verify_write_permission(self, request): 

149 

150 # If token authentication is in use, verify that the token allows write operations (for unsafe methods). 

151 if request.method in SAFE_METHODS or request.auth.write_enabled: 151 ↛ 153line 151 didn't jump to line 153 because the condition on line 151 was always true

152 return True 

153 return False 

154 

155 def has_permission(self, request, view): 

156 

157 # Enforce Token write ability 

158 if isinstance(request.auth, Token) and not self._verify_write_permission(request): 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true

159 return False 

160 

161 return super().has_permission(request, view) 

162 

163 def has_object_permission(self, request, view, obj): 

164 

165 # Enforce Token write ability 

166 if isinstance(request.auth, Token) and not self._verify_write_permission(request): 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true

167 return False 

168 

169 return super().has_object_permission(request, view, obj) 

170 

171 

172class TokenWritePermission(BasePermission): 

173 """ 

174 Verify the token has write_enabled for unsafe methods, without requiring specific model permissions. 

175 Used for custom actions that accept user data but don't map to standard CRUD operations. 

176 """ 

177 

178 def has_permission(self, request, view): 

179 if not isinstance(request.auth, Token): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true

180 raise exceptions.PermissionDenied( 

181 "TokenWritePermission requires token authentication." 

182 ) 

183 return bool(request.method in SAFE_METHODS or request.auth.write_enabled) 

184 

185 

186class TokenSyncPermission(TokenPermissions): 

187 """ 

188 Require a model's sync permission for the sync action, in place of the add permission a POST otherwise 

189 maps to. Token write ability is still enforced by the parent class. 

190 """ 

191 perms_map = { 

192 **TokenPermissions.perms_map, 

193 'POST': ['%(app_label)s.sync_%(model_name)s'], 

194 } 

195 

196 

197class IsAuthenticatedOrLoginNotRequired(BasePermission): 

198 """ 

199 Returns True if the user is authenticated or LOGIN_REQUIRED is False. 

200 """ 

201 def has_permission(self, request, view): 

202 if not settings.LOGIN_REQUIRED: 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true

203 return True 

204 return request.user.is_authenticated 

205 

206 

207class TokenScheme(OpenApiAuthenticationExtension): 

208 target_class = 'netbox.api.authentication.TokenAuthentication' 

209 name = 'tokenAuth' 

210 match_subclasses = True 

211 

212 def get_security_definition(self, auto_schema): 

213 return { 

214 'type': 'apiKey', 

215 'in': 'header', 

216 'name': 'Authorization', 

217 'description': '`Token <token>` (v1) or `Bearer <key>.<token>` (v2)', 

218 }