NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-26 18:38:38 -06:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

format to autopep8 compliance

Browse Source
This commit is contained in:
Max Rink 2021-02-09 11:59:22 +01:00
parent 676336364e
commit 1c67c60804
2 changed files with 43 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
netbox/netbox/authentication.py
View File

@ -106,16 +106,19 @@ class RemoteUserBackend(_RemoteUserBackend):
try:
group_list.append(Group.objects.get(name=name))
except Group.DoesNotExist:
logging.error(f"Could not assign group {name} to remotely-authenticated user {user}: Group not found")
logging.error(
f"Could not assign group {name} to remotely-authenticated user {user}: Group not found")
if group_list:
user.groups.set(group_list)
logger.debug(f"Assigned groups to remotely-authenticated user {user}: {group_list}")
else:
logger.debug(
f"Assigned groups to remotely-authenticated user {user}: {group_list}")
else:
user.groups.clear()
logger.debug(f"Stripping user {user} from Groups")
user.is_superuser = self._is_superuser(user)
logger.debug(f"User {user} is Superuser: {user.is_superuser}")
logger.debug(f"User {user} should be Superuser: {self._is_superuser(user)}")
logger.debug(
f"User {user} should be Superuser: {self._is_superuser(user)}")
user.is_staff = self._is_staff(user)
logger.debug(f"User {user} is Staff: {user.is_staff}")
@ -132,7 +135,8 @@ class RemoteUserBackend(_RemoteUserBackend):
object with the given username is not found in the database.
"""
logger = logging.getLogger('netbox.authentication.RemoteUserBackend')
logger.debug(f"trying to authenticate {remote_user} with groups {remote_groups}")
logger.debug(
f"trying to authenticate {remote_user} with groups {remote_groups}")
if not remote_user:
return
user = None
@ -155,7 +159,7 @@ class RemoteUserBackend(_RemoteUserBackend):
if self.user_can_authenticate(user):
if settings.REMOTE_AUTH_GROUP_SYNC_ENABLED:
if user is not None and not isinstance(user, AnonymousUser):
return self.configure_groups(user,remote_groups)
return self.configure_groups(user, remote_groups)
else:
return user
else:
@ -172,10 +176,11 @@ class RemoteUserBackend(_RemoteUserBackend):
user_groups.add(g.name)
logger.debug(f"User {user.username} is in Groups:{user_groups}")
result = user.username in superusers or (set(user_groups) & set(superuser_groups))
result = user.username in superusers or (
set(user_groups) & set(superuser_groups))
logger.debug(f"User {user.username} in Superuser Users :{result}")
return bool(result)
def _is_staff(self, user):
logger = logging.getLogger('netbox.authentication.RemoteUserBackend')
staff_groups = settings.REMOTE_AUTH_STAFF_GROUPS
@ -185,8 +190,9 @@ class RemoteUserBackend(_RemoteUserBackend):
user_groups = set()
for g in user.groups.all():
user_groups.add(g.name)
logger.debug(f"User {user.username} is in Groups:{user_groups}")
result = user.username in staff_users or (set(user_groups) & set(staff_groups))
logger.debug(f"User {user.username} is in Groups:{user_groups}")
result = user.username in staff_users or (
set(user_groups) & set(staff_groups))
logger.debug(f"User {user.username} in Staff Users :{result}")
return bool(result)
@ -199,18 +205,22 @@ class RemoteUserBackend(_RemoteUserBackend):
try:
group_list.append(Group.objects.get(name=name))
except Group.DoesNotExist:
logging.error(f"Could not assign group {name} to remotely-authenticated user {user}: Group not found")
logging.error(
f"Could not assign group {name} to remotely-authenticated user {user}: Group not found")
if group_list:
user.groups.add(*group_list)
logger.debug(f"Assigned groups to remotely-authenticated user {user}: {group_list}")
logger.debug(
f"Assigned groups to remotely-authenticated user {user}: {group_list}")
# Assign default object permissions to the user
permissions_list = []
for permission_name, constraints in settings.REMOTE_AUTH_DEFAULT_PERMISSIONS.items():
try:
object_type, action = resolve_permission_ct(permission_name)
object_type, action = resolve_permission_ct(
permission_name)
# TODO: Merge multiple actions into a single ObjectPermission per content type
obj_perm = ObjectPermission(actions=[action], constraints=constraints)
obj_perm = ObjectPermission(
actions=[action], constraints=constraints)
obj_perm.save()
obj_perm.users.add(user)
obj_perm.object_types.add(object_type)
@ -221,9 +231,11 @@ class RemoteUserBackend(_RemoteUserBackend):
"<app>.<action>_<model>. (Example: dcim.add_site)"
)
if permissions_list:
logger.debug(f"Assigned permissions to remotely-authenticated user {user}: {permissions_list}")
logger.debug(
f"Assigned permissions to remotely-authenticated user {user}: {permissions_list}")
else:
logger.debug(f"Skipped initial assignment of permissions and groups to remotely-authenticated user {user} as Group sync is enabled")
logger.debug(
f"Skipped initial assignment of permissions and groups to remotely-authenticated user {user} as Group sync is enabled")
return user

21
netbox/netbox/middleware.py
View File

@ -19,6 +19,7 @@ class LoginRequiredMiddleware(object):
"""
If LOGIN_REQUIRED is True, redirect all non-authenticated users to the login page.
"""
def __init__(self, get_response):
self.get_response = get_response
@ -48,7 +49,8 @@ class RemoteUserMiddleware(RemoteUserMiddleware_):
return settings.REMOTE_AUTH_HEADER
def process_request(self, request):
logger = logging.getLogger('netbox.authentication.RemoteUserMiddleware')
logger = logging.getLogger(
'netbox.authentication.RemoteUserMiddleware')
# Bypass middleware if remote authentication is not enabled
if not settings.REMOTE_AUTH_ENABLED:
return
@ -84,7 +86,8 @@ class RemoteUserMiddleware(RemoteUserMiddleware_):
# to authenticate the user.
if settings.REMOTE_AUTH_GROUP_SYNC_ENABLED:
logger.debug("Trying to sync Groups")
user = auth.authenticate(request, remote_user=username, remote_groups=self._get_groups(request))
user = auth.authenticate(
request, remote_user=username, remote_groups=self._get_groups(request))
else:
user = auth.authenticate(request, remote_user=username)
if user:
@ -92,11 +95,13 @@ class RemoteUserMiddleware(RemoteUserMiddleware_):
# by logging the user in.
request.user = user
auth.login(request, user)
def _get_groups(self, request):
logger = logging.getLogger('netbox.authentication.RemoteUserMiddleware')
groups_string = request.META.get(settings.REMOTE_AUTH_GROUP_HEADER, None)
def _get_groups(self, request):
logger = logging.getLogger(
'netbox.authentication.RemoteUserMiddleware')
groups_string = request.META.get(
settings.REMOTE_AUTH_GROUP_HEADER, None)
if groups_string:
groups = groups_string.split(settings.REMOTE_AUTH_GROUP_SEPERATOR)
else:
@ -104,6 +109,7 @@ class RemoteUserMiddleware(RemoteUserMiddleware_):
logger.debug(f"Groups are {groups}")
return groups
class ObjectChangeMiddleware(object):
"""
This middleware performs three functions in response to an object being created, updated, or deleted:
@ -118,6 +124,7 @@ class ObjectChangeMiddleware(object):
have been created. Conversely, deletions are acted upon immediately, so that the serialized representation of the
object is recorded before it (and any related objects) are actually deleted from the database.
"""
def __init__(self, get_response):
self.get_response = get_response
@ -137,6 +144,7 @@ class APIVersionMiddleware(object):
"""
If the request is for an API endpoint, include the API version as a response header.
"""
def __init__(self, get_response):
self.get_response = get_response
@ -152,6 +160,7 @@ class ExceptionHandlingMiddleware(object):
Intercept certain exceptions which are likely indicative of installation issues and provide helpful instructions
to the user.
"""
def __init__(self, get_response):
self.get_response = get_response