NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-08-17 04:58:16 -06:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

#8233 Restrict API key usage by source IP

Browse Source 
super authenicate()
This commit is contained in:
Pieter Lambrecht 2022-03-14 12:34:49 +01:00
parent 556bfad66a
commit a37d0897c2
1 changed files with 6 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
netbox/netbox/api/authentication.py
View File

@ -10,30 +10,13 @@ class TokenAuthentication(authentication.TokenAuthentication):
A custom authentication scheme which enforces Token expiration times. A custom authentication scheme which enforces Token expiration times.
""" """
model = Token model = Token
__request = False
def authenticate(self, request): def authenticate(self, request):
auth = authentication.get_authorization_header(request).split() self.request=request
return super().authenticate(request)
if not auth or auth[0].lower() != self.keyword.lower().encode(): def authenticate_credentials(self, key):
return None
if len(auth) == 1:
msg = 'Invalid token header. No credentials provided.'
raise exceptions.AuthenticationFailed(msg)
elif len(auth) > 2:
msg = 'Invalid token header. Token string should not contain spaces.'
raise exceptions.AuthenticationFailed(msg)
try:
token = auth[1].decode()
except UnicodeError:
msg = 'Invalid token header. Token string should not contain invalid characters.'
raise exceptions.AuthenticationFailed(msg)
return self.authenticate_credentials(request,token)
def authenticate_credentials(self, request, key):
model = self.get_model() model = self.get_model()
try: try:
token = model.objects.prefetch_related('user').get(key=key) token = model.objects.prefetch_related('user').get(key=key)
@ -41,7 +24,8 @@ class TokenAuthentication(authentication.TokenAuthentication):
raise exceptions.AuthenticationFailed("Invalid token") raise exceptions.AuthenticationFailed("Invalid token")
# Verify source IP is allowed # Verify source IP is allowed
if len(token.allowed_ipranges) > 0: request=self.request
if len(token.allowed_ipranges) > 0 and request:
if settings.PROXY_HEADER_REALIP in request.META: if settings.PROXY_HEADER_REALIP in request.META:
clientip = request.META[settings.PROXY_HEADER_REALIP].split(",")[0].strip() clientip = request.META[settings.PROXY_HEADER_REALIP].split(",")[0].strip()