mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-07-09 02:44:10 -04:00
92f78b14fe
the js `fetch` function will follow redirects blindly and return you the final 200 response. when there's an error, we don't want to go anywhere, and we want to use the flask `flash` functionality to pop up an error page (the likely reason for seeing this is using a yubikey that isn't associated with your user). using `flash` and then `window.location.reload()` handles this fine. However, when the user does log in succesfully we need to properly log them in - this includes: * checking their account isn't over the max login count * resetting failed login count to 0 if not * setting a new session id in the database (so other browser windows are logged out) * checking if they need to revalidate their email access (every 90 days) * clearing old user out of the cache This code all happens in the ajax function rather than being in a separate redirect, so that you can't just navigate to the login flow. I wasn't able to unit test that function due how it uses the session and other flask globals, so moved the auth into its own function so it's easy to stub out all that CBOR nonsense. TODO: We still need to pass any `next` URLs through the chain from login page all the way through the javascript AJAX calls and redirects to the log_in_user function
228 lines
8.1 KiB
Python
228 lines
8.1 KiB
Python
from notifications_python_client.errors import HTTPError
|
|
|
|
from app.models.roles_and_permissions import (
|
|
translate_permissions_from_admin_roles_to_db,
|
|
)
|
|
from app.notify_client import NotifyAdminAPIClient, cache
|
|
|
|
ALLOWED_ATTRIBUTES = {
|
|
'name',
|
|
'email_address',
|
|
'mobile_number',
|
|
'auth_type',
|
|
'updated_by',
|
|
'current_session_id'
|
|
}
|
|
|
|
|
|
class UserApiClient(NotifyAdminAPIClient):
|
|
|
|
def init_app(self, app):
|
|
super().init_app(app)
|
|
self.admin_url = app.config['ADMIN_BASE_URL']
|
|
|
|
def register_user(self, name, email_address, mobile_number, password, auth_type):
|
|
data = {
|
|
"name": name,
|
|
"email_address": email_address,
|
|
"mobile_number": mobile_number,
|
|
"password": password,
|
|
"auth_type": auth_type
|
|
}
|
|
user_data = self.post("/user", data)
|
|
return user_data['data']
|
|
|
|
def get_user(self, user_id):
|
|
return self._get_user(user_id)['data']
|
|
|
|
@cache.set('user-{user_id}')
|
|
def _get_user(self, user_id):
|
|
return self.get("/user/{}".format(user_id))
|
|
|
|
def get_user_by_email(self, email_address):
|
|
user_data = self.post('/user/email', data={'email': email_address})
|
|
return user_data['data']
|
|
|
|
def get_user_by_email_or_none(self, email_address):
|
|
try:
|
|
return self.get_user_by_email(email_address)
|
|
except HTTPError as e:
|
|
if e.status_code == 404:
|
|
return None
|
|
raise e
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def update_user_attribute(self, user_id, **kwargs):
|
|
data = dict(kwargs)
|
|
disallowed_attributes = set(data.keys()) - ALLOWED_ATTRIBUTES
|
|
if disallowed_attributes:
|
|
raise TypeError('Not allowed to update user attributes: {}'.format(
|
|
", ".join(disallowed_attributes)
|
|
))
|
|
|
|
url = "/user/{}".format(user_id)
|
|
user_data = self.post(url, data=data)
|
|
return user_data['data']
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def archive_user(self, user_id):
|
|
return self.post('/user/{}/archive'.format(user_id), data=None)
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def reset_failed_login_count(self, user_id):
|
|
url = "/user/{}/reset-failed-login-count".format(user_id)
|
|
user_data = self.post(url, data={})
|
|
return user_data['data']
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def update_password(self, user_id, password, validated_email_access=False):
|
|
data = {"_password": password}
|
|
if validated_email_access:
|
|
data["validated_email_access"] = validated_email_access
|
|
url = "/user/{}/update-password".format(user_id)
|
|
user_data = self.post(url, data=data)
|
|
return user_data['data']
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def verify_password(self, user_id, password):
|
|
try:
|
|
url = "/user/{}/verify/password".format(user_id)
|
|
data = {"password": password}
|
|
self.post(url, data=data)
|
|
return True
|
|
except HTTPError as e:
|
|
if e.status_code == 400 or e.status_code == 404:
|
|
return False
|
|
|
|
def send_verify_code(self, user_id, code_type, to, next_string=None):
|
|
data = {'to': to}
|
|
if next_string:
|
|
data['next'] = next_string
|
|
if code_type == 'email':
|
|
data['email_auth_link_host'] = self.admin_url
|
|
endpoint = '/user/{0}/{1}-code'.format(user_id, code_type)
|
|
self.post(endpoint, data=data)
|
|
|
|
def send_verify_email(self, user_id, to):
|
|
data = {'to': to}
|
|
endpoint = '/user/{0}/email-verification'.format(user_id)
|
|
self.post(endpoint, data=data)
|
|
|
|
def send_already_registered_email(self, user_id, to):
|
|
data = {'email': to}
|
|
endpoint = '/user/{0}/email-already-registered'.format(user_id)
|
|
self.post(endpoint, data=data)
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def check_verify_code(self, user_id, code, code_type):
|
|
data = {'code_type': code_type, 'code': code}
|
|
endpoint = '/user/{}/verify/code'.format(user_id)
|
|
try:
|
|
self.post(endpoint, data=data)
|
|
return True, ''
|
|
except HTTPError as e:
|
|
if e.status_code == 400 or e.status_code == 404:
|
|
return False, e.message
|
|
raise e
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def verify_webauthn_login(self, user_id, is_successful):
|
|
data = {'successful': is_successful}
|
|
endpoint = f'/user/{user_id}/verify/webauthn-login'
|
|
try:
|
|
self.post(endpoint, data=data)
|
|
return True, ''
|
|
except HTTPError as e:
|
|
if e.status_code == 403:
|
|
return False, e.message
|
|
raise e
|
|
|
|
def get_users_for_service(self, service_id):
|
|
endpoint = '/service/{}/users'.format(service_id)
|
|
return self.get(endpoint)['data']
|
|
|
|
def get_users_for_organisation(self, org_id):
|
|
endpoint = '/organisations/{}/users'.format(org_id)
|
|
return self.get(endpoint)['data']
|
|
|
|
@cache.delete('service-{service_id}')
|
|
@cache.delete('service-{service_id}-template-folders')
|
|
@cache.delete('user-{user_id}')
|
|
def add_user_to_service(self, service_id, user_id, permissions, folder_permissions):
|
|
# permissions passed in are the combined admin roles, not db permissions
|
|
endpoint = '/service/{}/users/{}'.format(service_id, user_id)
|
|
data = {
|
|
'permissions': [{'permission': x} for x in translate_permissions_from_admin_roles_to_db(permissions)],
|
|
'folder_permissions': folder_permissions,
|
|
}
|
|
|
|
self.post(endpoint, data=data)
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def add_user_to_organisation(self, org_id, user_id):
|
|
resp = self.post('/organisations/{}/users/{}'.format(org_id, user_id), data={})
|
|
return resp['data']
|
|
|
|
@cache.delete('service-{service_id}-template-folders')
|
|
@cache.delete('user-{user_id}')
|
|
def set_user_permissions(self, user_id, service_id, permissions, folder_permissions=None):
|
|
# permissions passed in are the combined admin roles, not db permissions
|
|
data = {
|
|
'permissions': [{'permission': x} for x in translate_permissions_from_admin_roles_to_db(permissions)],
|
|
}
|
|
|
|
if folder_permissions is not None:
|
|
data['folder_permissions'] = folder_permissions
|
|
|
|
endpoint = '/user/{}/service/{}/permission'.format(user_id, service_id)
|
|
self.post(endpoint, data=data)
|
|
|
|
def send_reset_password_url(self, email_address, next_string=None):
|
|
endpoint = '/user/reset-password'
|
|
data = {'email': email_address}
|
|
if next_string:
|
|
data['next'] = next_string
|
|
self.post(endpoint, data=data)
|
|
|
|
def find_users_by_full_or_partial_email(self, email_address):
|
|
endpoint = '/user/find-users-by-email'
|
|
data = {'email': email_address}
|
|
users = self.post(endpoint, data=data)
|
|
return users
|
|
|
|
@cache.delete('user-{user_id}')
|
|
def activate_user(self, user_id):
|
|
return self.post("/user/{}/activate".format(user_id), data=None)
|
|
|
|
def send_change_email_verification(self, user_id, new_email):
|
|
endpoint = '/user/{}/change-email-verification'.format(user_id)
|
|
data = {'email': new_email}
|
|
self.post(endpoint, data)
|
|
|
|
def get_organisations_and_services_for_user(self, user_id):
|
|
endpoint = '/user/{}/organisations-and-services'.format(user_id)
|
|
return self.get(endpoint)
|
|
|
|
def get_webauthn_credentials_for_user(self, user_id):
|
|
endpoint = f'/user/{user_id}/webauthn'
|
|
|
|
return self.get(endpoint)['data']
|
|
|
|
def create_webauthn_credential_for_user(self, user_id, credential):
|
|
endpoint = f'/user/{user_id}/webauthn'
|
|
|
|
return self.post(endpoint, data=credential.serialize())
|
|
|
|
def update_webauthn_credential_name_for_user(self, *, user_id, credential_id, new_name_for_credential):
|
|
endpoint = f'/user/{user_id}/webauthn/{credential_id}'
|
|
|
|
return self.post(endpoint, data={"name": new_name_for_credential})
|
|
|
|
def delete_webauthn_credential_for_user(self, *, user_id, credential_id):
|
|
endpoint = f'/user/{user_id}/webauthn/{credential_id}'
|
|
|
|
return self.delete(endpoint)
|
|
|
|
|
|
user_api_client = UserApiClient()
|