# By: Riasat Ullah # This file contains code for serving user specific pages. from constants import api_paths, component_names as cnm, label_names as lnm, pages, static_vars,\ tutorial_names, var_names from context_manager import components_context, info_context, users_context from device_detector import DeviceDetector from django.core import exceptions from django.http import Http404, HttpResponse, JsonResponse from django.shortcuts import redirect, render from django.views.decorators.http import require_http_methods from taskcallweb import settings from system_tests.test_data import test_data_organization, test_data_policies, test_data_services, test_data_users from translators import label_translator as lt from urllib.parse import urlencode from utils import client_data_path, helpers, logging, s3 from utils.custom_redirect import MobileAppRedirect from validations import request_validator import botocore.exceptions import configuration as configs import filetype import json import pytz @require_http_methods(['GET', 'POST']) def login(request): ''' Fetches the login page for a GET request and processes login for a POST request. :param request: Http request :return: Http response ''' tc_domain = static_vars.regional_test_server_urls[settings.REGION][var_names.domain]\ if settings.TEST_MODE else static_vars.regional_urls[settings.REGION][var_names.domain] if request.method == 'GET': lang = request_validator.get_user_language(request) region = request_validator.get_host_region(request) context = info_context.get_login_context(lang) context[var_names.taskcall_domain] = tc_domain for_mobile = False device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if helpers.is_mobile_app_login(request) and device.is_mobile(): for_mobile = True # Redirect to the correct region if the region saved in cookies is different from the server region if region != settings.REGION: switch_page_name = 'login/mobile-app' if for_mobile else 'login' switch_redirect_url = helpers.get_redirect_url_for_switching_region(request, region, switch_page_name) if switch_redirect_url is not None: return redirect(switch_redirect_url) # This will determine how the page should be rendered. If the method is 'sso', then the SSO section # of the page will be rendered. Otherwise, the standard section would be shown. if 'method' in request.GET: context[var_names.method] = request.GET.get('method') try: if request_validator.user_in_session(request): return redirect(pages.incidents_url) else: return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) except Exception as e: logging.exception(str(e)) return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) elif request.method == 'POST': lang = request_validator.get_user_language(request) context = info_context.get_login_context(lang) context[var_names.taskcall_domain] = tc_domain for_mobile = False try: email = request.POST[var_names.email] password = request.POST[var_names.password] device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if helpers.is_mobile_app_login(request) and device.is_mobile(): for_mobile = True if not request.session.session_key: request.session.save() request.session.set_expiry(2592000) body = { var_names.email: email, var_names.password: password, var_names.ip_address: request.META.get(static_vars.ip_address_attribute) } if for_mobile: body[var_names.access_method] = static_vars.app body[var_names.push_token] = helpers.get_mobile_app_push_token(request) else: body[var_names.access_method] = static_vars.web body[var_names.session_id] = request.session.session_key if settings.TEST_MODE: status, output = 401, lt.get_label(lnm.err_invalid_email_or_password, settings.LANGUAGE_CODE) for key in test_data_users.user_details: if test_data_users.user_details[key][var_names.email] == email: status, output = 200, test_data_users.user_authentication_details[key] else: status, output = helpers.post_api_request(api_paths.user_login, body) request.method = 'GET' if status == 200: if for_mobile: redirect_url = helpers.get_mobile_app_full_redirect_url(request, output) helpers.clear_mobile_app_login_variables(request) return MobileAppRedirect(redirect_url) else: helpers.set_session_authorization_tokens(request, output) redirect_url = helpers.get_redirect_page(request) if helpers.has_only_stakeholder_rights(output, redirect_url): redirect_url = pages.status_dashboard_url resp = redirect(redirect_url) resp.set_cookie(var_names.language, output[var_names.language], max_age=configs.DEFAULT_COOKIE_LIFETIME, secure=True, domain=static_vars.base_domain, httponly=True, samesite='None') resp.set_cookie(var_names.host_region, output[var_names.host_region], max_age=configs.DEFAULT_COOKIE_LIFETIME, secure=True, domain=static_vars.base_domain, httponly=True, samesite='None') return resp else: context[var_names.error_message] = output.replace("'", "") return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) except Exception as e: logging.exception(str(e)) context[var_names.error_message] = lt.get_label(lnm.err_system_error, lang) return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) @require_http_methods(['GET']) def login_from_mobile_app(request): ''' Handles the first step of logging in a user from the mobile app. It stores the mobile app redirect url and the push token in cache and then redirects to the login page. These stored variables should be used later on to log the user into the mobile app. They should be cleared before redirecting back to the mobile app. :param request: Http request :return: Http response ''' if request.method == 'GET': device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if device.is_mobile(): helpers.set_mobile_app_login_variables(request) return redirect(pages.login_url) @require_http_methods(['POST']) def logout(request): ''' Handles a logout request. There is no need to make a logout call to the API since no session data is stored in the API. All session data is stored and handled on the web server. So, just deleting the session variable would be enough. :param request: Http request :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if not settings.TEST_MODE: helpers.post_api_request(api_paths.user_logout, dict(), request, lang=lang) helpers.clear_session_authorization_tokens(request) request.session.set_expiry(5) return redirect(pages.login_url) else: return redirect(pages.login_url) @require_http_methods(['GET']) def get_first_name(request): ''' Get the first name of the user. This should be used to show the name on the dashboard. :param request: Http request :return: JSON response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): first_name = '' if var_names.display_name in request.session: first_name = request.session[var_names.display_name].split(' ')[0] return JsonResponse(first_name, status=200, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET']) def get_session_user_display_details(request): ''' Get the display details of the current session user - first name and profile picture url. This should be used to show the name and the avatar on the dashboard. :param request: Http request :return: JSON response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): first_name, image_url = '', '' if var_names.display_name in request.session: first_name = request.session[var_names.display_name].split(' ')[0] if var_names.profile_picture in request.session: image_url = request.session[var_names.profile_picture] details = {var_names.first_name: first_name, var_names.profile_picture: image_url} return JsonResponse(details, status=200, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def get_requested_user_accounts(request): ''' Get the requested accounts. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: return JsonResponse(test_data_organization.requested_accounts, safe=False) else: body = dict() status, output = helpers.post_api_request(api_paths.org_requested_accounts_list, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def request_new_user_account(request): ''' Request a new user account. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: return JsonResponse('Account verification code has been sent', safe=False) else: body = json.loads(request.body.decode()) status, output = helpers.post_api_request(api_paths.org_requested_accounts_add, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def resend_new_user_account_verification_code(request): ''' Re-sends the verification code of a new user account. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: return JsonResponse('Verification code has been resent', safe=False) else: body = json.loads(request.body.decode()) status, output = helpers.post_api_request(api_paths.org_requested_accounts_resend_code, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def cancel_requested_user_account(request): ''' Cancel a requested user account. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: return JsonResponse('Requested account has been cancelled', safe=False) else: body = json.loads(request.body.decode()) status, output = helpers.post_api_request(api_paths.org_requested_accounts_cancel, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def get_user_list(request): ''' GET: Gets the user list page. POST: Gets the list of users. :param request: Http request :return: GET: Http response | POST: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): lang = request_validator.get_user_language(request) nav_bar_components = request_validator.get_nav_bar_components(request) has_view_perm, has_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_users, nav_bar_components ) if not has_view_perm: raise exceptions.PermissionDenied context = users_context.get_user_list_context(lang, nav_bar_components) context[var_names.has_edit_permission] = has_edit_perm if settings.TEST_MODE: status, output = 200, test_data_organization.allowed_user_roles else: body = {var_names.data_type: [var_names.user_roles]} status, output = helpers.post_api_request(api_paths.org_list_maker, body, request, lang=lang) if status == 200: context[var_names.user_role_items] = users_context.get_user_roles_content(output, context) return render(request, pages.user_list_page, context=context) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: output = test_data_users.org_members if var_names.keywords in body and body[var_names.keywords] is not None: keywords = body[var_names.keywords].lower() output = [x for x in output if keywords in x[1].lower()] if var_names.row_limit in body and var_names.row_offset in body: row_limit = body[var_names.row_limit] row_offset = body[var_names.row_offset] if row_limit <= len(output): start_ = row_offset end_ = row_offset + row_limit output = output[start_:end_] return JsonResponse(output, status=200, safe=False) else: status, output = helpers.post_api_request(api_paths.org_members_list, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def get_user_display_name(request): ''' Gets the display name of a user. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: preferred_username = body[var_names.preferred_username] if preferred_username is None: item = list(test_data_users.user_details.values())[0] display_name = item[var_names.first_name] + ' ' + item[var_names.last_name] return JsonResponse(display_name, safe=False) for item in list(test_data_users.user_details.values()): if item[var_names.preferred_username] == preferred_username: display_name = item[var_names.first_name] + ' ' + item[var_names.last_name] return JsonResponse(display_name, safe=False) return JsonResponse(lt.get_label(lnm.err_unknown_resource, lang), status=404, safe=False) else: status, output = helpers.post_api_request(api_paths.user_profile_display_name, body, request) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def get_user_profile(request, preferred_username=None): ''' GET: Gets the user profile page. POST: Gets the details of the user. :param request: Http request :param preferred_username: the preferred username of the user :return: GET: Http response | POST: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): lang = request_validator.get_user_language(request) nav_bar_components = request_validator.get_nav_bar_components(request) component_features = request_validator.get_component_features(request) context = users_context.get_user_profile_context(lang, nav_bar_components) # These rules are specifically for the user who is accessing the profile. This will be # needed to determine if this user can add/remove component roles of the user whose profile # is being viewed. This will be needed for accounts with advanced permissions. # Only Admins and the Owner can change the component level permissions of users. # Stakeholders cannot be given component roles (this is checked on the client side script). has_mem_view_perm, has_mem_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_users, nav_bar_components ) context[var_names.has_edit_member_permission] = has_mem_edit_perm if preferred_username is None: has_view_perm, has_edit_perm = True, True else: has_view_perm, has_edit_perm = has_mem_view_perm, has_mem_edit_perm if not has_view_perm: raise exceptions.PermissionDenied context[var_names.has_edit_permission] = has_edit_perm context[var_names.has_edit_password_permission] = True if preferred_username is None else False context[var_names.has_business_service_subscription_permission] =\ True if cnm.feat_users_business_subscriptions in component_features else False context[var_names.has_component_flexible_role] = \ True if cnm.feat_users_component_flexible_role in request_validator.get_component_features(request) \ else False context[var_names.has_contextual_search_permission] = request_validator.has_edit_permission( request, cnm.dis_com_tags) context[var_names.has_groups_permission] = request_validator.has_edit_permission( request, cnm.dis_com_itsm_groups) context[var_names.context] = json.dumps({var_names.preferred_username: preferred_username}) context[var_names.user_lang_items] = components_context.get_languages_content(context) if settings.TEST_MODE: status, output = 200, test_data_organization.allowed_user_roles else: body = {var_names.data_type: [var_names.user_roles]} status, output = helpers.post_api_request(api_paths.org_list_maker, body, request, lang=lang) if status == 200: context[var_names.user_role_items] = users_context.get_user_roles_content(output, context) context[var_names.user_component_role_items] = users_context.get_advanced_component_roles_content(context) return render(request, pages.user_profile_page, context=context) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: if preferred_username is None: user_disp_name = request.session[var_names.display_name] for item in test_data_users.users_list: if user_disp_name == item[0]: return JsonResponse(test_data_users.user_details[item[1]], safe=False) return JsonResponse(lt.get_label(lnm.err_unknown_resource, lang), status=404, safe=False) else: for item in list(test_data_users.user_details.values()): if item[var_names.preferred_username] == preferred_username: return JsonResponse(item, safe=False) return JsonResponse(lt.get_label(lnm.err_unknown_resource, lang), status=404, safe=False) else: body = {var_names.preferred_username: preferred_username} status, output = helpers.post_api_request(api_paths.user_profile, body, request) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def edit_user_profile(request): ''' Edit a user's profile. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) user_self = body[var_names.for_user] del body[var_names.for_user] if settings.TEST_MODE: status, output = 200, 'User profile has been updated' else: status, output = helpers.post_api_request(api_paths.user_profile_update, body, request, lang=lang) if status == 200: if user_self: curr_lang = request_validator.get_user_language(request) new_lang = body[var_names.language] if curr_lang != new_lang: request.session[var_names.language] = new_lang curr_display_name = request.session[var_names.display_name] new_display_name = body[var_names.first_name] + ' ' + body[var_names.last_name] if curr_display_name != new_display_name: request.session[var_names.display_name] = new_display_name return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def edit_user_role(request): ''' Edit a user's account role. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('Role has been updated', safe=False) else: status, output = helpers.post_api_request(api_paths.org_members_edit_role, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def unlock_user(request): ''' Unlock a user. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('User account has been unlocked', safe=False) else: status, output = helpers.post_api_request(api_paths.org_members_unlock, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def delete_user(request): ''' Deletes a user from the organization. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) pref_name = body[var_names.member_name] if settings.TEST_MODE: status, output = 200, 'Success' else: status, output = helpers.post_api_request(api_paths.org_members_delete, body, request, lang=lang) # Delete the user profile picture if the user was deleted successfully if status == 200 and not settings.TEST_MODE: try: if settings.TEST_MODE: new_status, new_output = 200, test_data_organization.organization_details else: new_status, new_output = helpers.post_api_request(api_paths.org_details, dict(), request, lang=lang) if new_status == 200: img_bucket, img_fldr = client_data_path.get_user_profile_picture_path( new_output[var_names.account_id], pref_name, '')[:2] logging.info('Deleting user profile picture - ' + img_fldr) s3.delete_folder_files(img_bucket, img_fldr) except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception('Failed to delete user profile picture on s3 - ' + pref_name) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def forgot_password(request): ''' Allows a user to reset their password through verification if they have forgotten it. This function should be used only from the login page. If the user wants to update their password then they should use the edit_password function. :param request: Http request :return: JSON response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) context = info_context.get_forgot_password_context(lang) return render(request, pages.forgot_password_page, context=context) elif request.method == 'POST': body = json.loads(request.body.decode()) if settings.TEST_MODE: status, output = 200, 'temp_token' else: status, output = helpers.post_api_request(api_paths.user_password_forgot, body) if status == 200: request.session[var_names.temp_token] = output output = 'success' return JsonResponse(output, status=status, safe=False) @require_http_methods(['POST']) def verify_forgotten_password_reset_attempt(request): ''' Checks if the verification code provided by a user for a forgotten password reset attempt verifies or not. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) body = json.loads(request.body.decode()) if var_names.temp_token in request.session: if settings.TEST_MODE: status, output = 200, 'verified' else: status, output = helpers.send_post_request(api_paths.user_password_forgot_verify, body, request.session[var_names.temp_token], lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def reset_forgotten_password(request): ''' Allow a user to edit it's password by creating a verification code for them to verify with. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if var_names.temp_token in request.session: body = json.loads(request.body.decode()) if settings.TEST_MODE: status, output = 200, 'success' else: status, output = helpers.send_post_request(api_paths.user_password_forgot_reset, body, request.session[var_names.temp_token], lang=lang) if status == 200: del request.session[var_names.temp_token] return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def reset_password(request): ''' Resets a user's password. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: status, output = 200, 'success' else: status, output = helpers.post_api_request(api_paths.user_password_reset, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def forgot_subdomain(request): ''' If a user forgets the subdomain of their organization account, then they can request to have that sent to their email. This is only for SSO enabled accounts. This is a pre-login step. :param request: Http request :return: JSON response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) context = info_context.get_forgot_subdomain_context(lang) return render(request, pages.forgot_subdomain_page, context=context) elif request.method == 'POST': body = json.loads(request.body.decode()) if settings.TEST_MODE: status, output = 200, 'Success' else: status, output = helpers.post_api_request(api_paths.user_subdomain_forgot, body) return JsonResponse(output, status=status, safe=False) @require_http_methods(['POST']) def get_all_timezones(request): ''' Get all the timezones that TaskCall allows (all pytz timezones) :param request: Http request :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: return JsonResponse(pytz.all_timezones, safe=False) except Exception as e: logging.exception(str(e)) return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=500, safe=False) @require_http_methods(['POST']) def get_user_timezone(request): ''' Gets the timezone that the user is making the request from :param request: Http request :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: ip_address = request.META.get(static_vars.ip_address_attribute) if settings.TEST_MODE: return JsonResponse('Africa/Nairobi', safe=False) else: timezone = helpers.get_info_from_ip_address(ip_address, static_vars.ip_timezone_attribute) return JsonResponse(timezone, safe=False) except Exception as e: logging.exception(str(e)) return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=500, safe=False) @require_http_methods(['POST']) def get_alert_notification_rules(request): ''' Gets the alert notification rules of a user. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse(list(test_data_users.user_alert_notification_rules.values())[0], safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_alerts_list, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def create_alert_notification_rule(request): ''' Create a new alert notification rule. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('Created', safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_alerts_create, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def edit_alert_notification_rule(request): ''' Edits an alert notification rule. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('Edited', safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_alerts_edit, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def delete_alert_notification_rule(request): ''' Delete an alert notification. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('Deleted', safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_alerts_delete, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def get_hand_off_notification_rules(request): ''' Gets the hand-off notification rules of a user. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse(list(test_data_users.user_hand_off_notification_rules.values())[0], safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_hand_off_list, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def create_hand_off_notification_rule(request): ''' Create a new hand-off notification rule. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('Created', safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_hand_off_create, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def edit_hand_off_notification_rule(request): ''' Edits a hand-off notification rule. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('Edited', safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_hand_off_edit, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def delete_hand_off_notification_rule(request): ''' Delete a hand-off notification. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse('Deleted', safe=False) else: status, output = helpers.post_api_request(api_paths.user_notification_rules_hand_off_delete, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def get_user_business_service_subscriptions(request, preferred_username=None): ''' GET: Gets the page for showing a user's business service subscriptions page. POST: Gets the list of all the business services the user subscribes to. :param request: Http request :param preferred_username: preferred username of the user whose subscriptions are being requested; if None then gets the subscriptions of the user making the request :return: GET: html page | POST: list of subscriptions ''' if request.method == 'GET': if request_validator.user_in_session(request): lang = request_validator.get_user_language(request) nav_bar_components = request_validator.get_nav_bar_components(request) if cnm.feat_users_business_subscriptions not in request_validator.get_component_features(request): raise exceptions.PermissionDenied if preferred_username is not None and not request_validator.has_edit_permission(request, cnm.dis_com_users): raise exceptions.PermissionDenied context = users_context.get_user_business_service_subscriptions_context(lang, nav_bar_components) context[var_names.context] = json.dumps({var_names.preferred_username: preferred_username}) return render(request, pages.user_business_service_subscriptions_page, context=context) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = {var_names.preferred_username: preferred_username} if settings.TEST_MODE: return JsonResponse(test_data_services.business_services_basic_list[0:2], safe=False) else: status, output = helpers.post_api_request(api_paths.business_services_subscriptions, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def update_business_service_subscriptions(request): ''' Updates a user's business service subscriptions. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) preferred_username = body[var_names.preferred_username] has_edit_perm = request_validator.has_view_permission(request, cnm.dis_com_business_services) if not has_edit_perm: raise exceptions.PermissionDenied if preferred_username is not None and not request_validator.has_edit_permission(request, cnm.dis_com_users): raise exceptions.PermissionDenied if settings.TEST_MODE: return JsonResponse('Subscriptions have been saved', safe=False) else: status, output = helpers.post_api_request(api_paths.business_services_subscriptions_update, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def get_user_advanced_permissions(request): ''' Updates a user's business service subscriptions. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if cnm.feat_users_component_flexible_role not in request_validator.get_component_features(request): raise exceptions.PermissionDenied if settings.TEST_MODE: return JsonResponse(test_data_services.user_advanced_component_permissions, safe=False) else: status, output = helpers.post_api_request(api_paths.user_advanced_permissions_list, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def add_user_advanced_permissions(request): ''' Add an advanced permission for a user on a component. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if cnm.feat_users_component_flexible_role not in request_validator.get_component_features(request): raise exceptions.PermissionDenied if settings.TEST_MODE: return JsonResponse('Success', safe=False) else: status, output = helpers.post_api_request(api_paths.user_advanced_permissions_create, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def delete_user_advanced_permissions(request): ''' Deletes a user's advanced permissions. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if cnm.feat_users_component_flexible_role not in request_validator.get_component_features(request): raise exceptions.PermissionDenied if settings.TEST_MODE: return JsonResponse('Success', safe=False) else: status, output = helpers.post_api_request(api_paths.user_advanced_permissions_delete, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def validate_profile_image_files(request): ''' Validates that the user profile image files are in the correct format. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): image_files = request.FILES.getlist('files[]') for item in image_files: file = filetype.guess(item) if file is not None and file.extension not in configs.USER_PROFILE_IMAGE_EXTENSIONS and\ file.mime not in configs.USER_PROFILE_IMAGE_MIME_TYPES: return JsonResponse(lt.get_label(lnm.err_user_profile_image_format, lang), status=422, safe=False) return JsonResponse(lt.get_label(lnm.msg_success, lang), safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def update_user_profile_picture(request): ''' Uploads a user's profile picture to s3 (if a new picture is provided) and updates the database entry. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.POST[var_names.data]) for_user = body[var_names.for_user] pref_name = body[var_names.preferred_username] if var_names.file_name in body and var_names.profile_picture in request.FILES: image_name = body[var_names.file_name] if settings.TEST_MODE: status, output = 200, test_data_organization.organization_details else: status, output = helpers.post_api_request(api_paths.org_details, dict(), request, lang=lang) if status == 200 and not settings.TEST_MODE: img_bucket, img_fldr, img_key, img_url = client_data_path.get_user_profile_picture_path( output[var_names.account_id], pref_name, image_name) # Upload the user profile image to s3 try: logging.info('Uploading user profile picture - ' + img_key) s3.delete_folder_files(img_bucket, img_fldr) s3.upload_media_file(img_bucket, img_key, request.FILES[var_names.profile_picture]) # add the image path/url to the payload to save in the database picture_url = img_url except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception('Failed to upload image file on s3') return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=500, safe=False) else: return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=500, safe=False) else: if var_names.avatar in body: picture_url = client_data_path.get_taskcall_avatar_url(body[var_names.avatar]) else: return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=500, safe=False) # Update the user image file url in the backend if settings.TEST_MODE: status, output = 200, 'success' else: new_payload = {var_names.preferred_username: pref_name, var_names.profile_picture: picture_url} status, output = helpers.post_api_request(api_paths.user_profile_picture_update, new_payload, request, lang=lang) if status == 200: output = picture_url if for_user: request.session[var_names.profile_picture] = picture_url return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def exists_onboarding_flow(request): ''' Check if the onboarding flow exists or not. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: flow_id = body[var_names.onboarding_flow] ttr_name = body[var_names.tutorial_name] if (flow_id == '1245' and ttr_name == tutorial_names.welcome) or\ (flow_id == '5421' and ttr_name == tutorial_names.intro): status, output = 200, True else: status, output = 200, False else: status, output = helpers.post_api_request(api_paths.user_onboarding_flow_exists, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET']) def unsubscribe_from_emails(request): ''' Unsubscribe a user from a tutorial email. :param request: Http request :return: JSON response ''' if request.method == 'GET': if 'type' in request.GET and 'id' in request.GET: subscription_type = request.GET.get('type') recipient_id = request.GET.get('id') if subscription_type == static_vars.onboarding: if not settings.TEST_MODE: body = {var_names.token: recipient_id} helpers.post_api_request(api_paths.user_tutorials_onboarding_email_unsubscribe, body) context = info_context.get_email_subscription_cancel_context(static_vars.lang_en) return render(request, pages.email_unsubscribe_page, context=context) else: raise Http404 else: raise Http404 @require_http_methods(['GET', 'POST']) def get_user_on_call_shifts(request, preferred_username=None): ''' GET: Gets the user on-call shifts page. POST: Gets the on-call shift details of the user. :param request: Http request :param preferred_username: the preferred username of the user :return: GET: Http response | POST: JSON response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): nav_bar_components = request_validator.get_nav_bar_components(request) context = users_context.get_user_on_call_shifts_context(lang, nav_bar_components) # These rules are specifically for the user who is accessing the profile. This will be # needed to determine if this user can add/remove component roles of the user whose profile # is being viewed. This will be needed for accounts with advanced permissions. # Only Admins and the Owner can change the component level permissions of users. # Stakeholders cannot be given component roles (this is checked on the client side script). has_mem_view_perm, has_mem_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_users, nav_bar_components ) context[var_names.has_edit_member_permission] = has_mem_edit_perm if preferred_username is None: has_view_perm, has_edit_perm = True, True else: has_view_perm, has_edit_perm = has_mem_view_perm, has_mem_edit_perm if not has_view_perm: raise exceptions.PermissionDenied context[var_names.has_edit_permission] = has_edit_perm context[var_names.context] = json.dumps({var_names.preferred_username: preferred_username}) return render(request, pages.user_on_call_shifts_page, context=context) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) body[var_names.preferred_username] = preferred_username if settings.TEST_MODE: status, output = 200, test_data_policies.on_call_schedule else: status, output = helpers.post_api_request(api_paths.user_on_call_shifts, body, request) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def override_on_call_shifts(request, preferred_username=None): ''' Override the on-call shifts of a user. :param request: Http request :param preferred_username: the preferred username of the user (in this view not only is it used to make sure the user has the correct rights, but we include it in the URL to make sure that any other on-call shifts url does not get picked up) :return: GET: Http response | POST: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): nav_bar_components = request_validator.get_nav_bar_components(request) if preferred_username is not None: has_mem_view_perm, has_mem_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_users, nav_bar_components ) if not has_mem_edit_perm: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) body = json.loads(request.body.decode()) if settings.TEST_MODE: status, output = 200, [] else: # Avoid duplicating overrides. Multiple shifts can be associated to the same routine. routines_list = list(set(body[var_names.routines])) failed_overrides = [] for item in routines_list: data_to_send = { var_names.routine_ref_id: item, var_names.routine_layers: { var_names.layer: 1, var_names.layer_name: 'Override', var_names.valid_start: body[var_names.valid_start], var_names.valid_end: body[var_names.valid_end], var_names.is_exception: True, var_names.rotation_start: body[var_names.rotation_start], var_names.shift_length: body[var_names.shift_length], var_names.rotation_frequency: 1, var_names.skip_days: [], var_names.rotations: body[var_names.rotations] } } item_status, item_output = helpers.post_api_request( api_paths.routines_override, data_to_send, request) if item_status != 200: failed_overrides.append(item) if len(failed_overrides) == 0: status, output = 200, [] else: status, output = 400, failed_overrides return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET']) def get_calendar_ics(request, preferred_username=None): ''' Gets the ics (calendar) file of on-call shifts. :param request: Http request :param preferred_username: the preferred username of the user :return: Http response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): nav_bar_components = request_validator.get_nav_bar_components(request) has_mem_view_perm, has_mem_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_users, nav_bar_components ) if preferred_username is None: has_view_perm, has_edit_perm = True, True else: has_view_perm, has_edit_perm = has_mem_view_perm, has_mem_edit_perm if not has_view_perm: raise exceptions.PermissionDenied start_date = request.GET[var_names.start_date] end_date = request.GET[var_names.end_date] body = { var_names.preferred_username: preferred_username, var_names.start_date: start_date, var_names.end_date: end_date, var_names.timezone: request.GET[var_names.timezone] } if settings.TEST_MODE: status, output = 200, test_data_policies.on_call_schedule else: status, output = helpers.post_api_request(api_paths.user_on_call_shifts, body, request) if status == 200: try: ics = helpers.prepare_calendar(output[var_names.data], output[var_names.timezone], output[var_names.display_name]) if ics: response = HttpResponse(ics, content_type='text/calendar') filename = output[var_names.display_name] + ' ' + start_date + ' - ' + end_date + '.ics' content = "attachment; filename={0}".format(filename) response['Content-Disposition'] = content return response return JsonResponse(lt.get_label(lnm.err_calendar_download, lang), status=500, safe=False) except Exception as e: logging.exception(str(e)) return JsonResponse(lt.get_label(lnm.err_calendar_download, lang), status=500, safe=False) else: return JsonResponse(lt.get_label(lnm.err_calendar_download, lang), status=500, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def get_webcal_url(request, preferred_username=None): ''' Gets the URL of the WebCal feed of a user. :param request: Http request :param preferred_username: the preferred username of the user :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): nav_bar_components = request_validator.get_nav_bar_components(request) has_mem_view_perm, has_mem_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_users, nav_bar_components ) if preferred_username is None: has_view_perm, has_edit_perm = True, True else: has_view_perm, has_edit_perm = has_mem_view_perm, has_mem_edit_perm if not has_view_perm: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) body = {var_names.preferred_username: preferred_username} if settings.TEST_MODE: status, output = 200, { var_names.preferred_username: test_data_users.users_list[0][1] if preferred_username is None else preferred_username, var_names.subdomain: test_data_organization.organization_details[var_names.subdomain], var_names.account_id: test_data_organization.organization_details[var_names.account_id] } else: status, output = helpers.post_api_request(api_paths.user_on_call_webcal_info, body, request) if status == 200: ref_dict = static_vars.regional_urls if settings.TEST_SERVER: ref_dict = static_vars.regional_test_server_urls cal_root = '{0}/public/{1}/{2}/on-call-shifts?'.format( ref_dict[settings.REGION][var_names.redirect_base], output[var_names.subdomain], output[var_names.account_id] ) params = {var_names.users: output[var_names.preferred_username]} cal_url = cal_root + urlencode(params) return JsonResponse(cal_url, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False)