39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
|
from multiprocessing.sharedctypes import Value, Array
|
||
|
|
from util.validation import MAX_LENGTH
|
||
|
|
|
||
|
|
class SuperUserManager(object):
|
||
|
|
""" In-memory helper class for quickly accessing (and updating) the valid
|
||
|
|
set of super users. This class communicates across processes to ensure
|
||
|
|
that the shared set is always the same.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, app):
|
||
|
|
usernames = app.config.get('SUPER_USERS', [])
|
||
|
|
usernames_str = ','.join(usernames)
|
||
|
|
|
||
|
|
self._max_length = len(usernames_str) + MAX_LENGTH + 1
|
||
|
|
self._array = Array('c', self._max_length, lock=True)
|
||
|
|
self._array.value = usernames_str
|
||
|
|
|
||
|
|
def is_superuser(self, username):
|
||
|
|
""" Returns if the given username represents a super user. """
|
||
|
|
usernames = self._array.value.split(',')
|
||
|
|
return username in usernames
|
||
|
|
|
||
|
|
def register_superuser(self, username):
|
||
|
|
""" Registers a new username as a super user for the duration of the container.
|
||
|
|
Note that this does *not* change any underlying config files.
|
||
|
|
"""
|
||
|
|
usernames = self._array.value.split(',')
|
||
|
|
usernames.append(username)
|
||
|
|
new_string = ','.join(usernames)
|
||
|
|
|
||
|
|
if len(new_string) <= self._max_length:
|
||
|
|
self._array.value = new_string
|
||
|
|
else:
|
||
|
|
raise Exception('Maximum superuser count reached. Please report this to support.')
|
||
|
|
|
||
|
|
def has_superusers(self):
|
||
|
|
""" Returns whether there are any superusers defined. """
|
||
|
|
return bool(self._array.value)
|