# By: Riasat Ullah # This file contains code for organization related views. from constants import api_paths, label_names as lnm, pages, static_vars, tutorial_names, url_paths, var_names from django.http import JsonResponse from django.shortcuts import redirect from django.views.decorators.http import require_http_methods from taskcallweb import settings from translators import label_translator as lt from system_tests.test_data import test_data_users from utils import client_data_path, helpers, logging from utils.exceptions import InvalidRequest from validations import request_validator import datetime import json @require_http_methods(['POST']) def save_org_owner_email(request): ''' Initiate the organization registration process. Store the email with which the registration is being initiated. Send the url for the next step of the process as response. :param request: Http request :return: JSON response -> redirect url path ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: body = json.loads(request.body.decode()) if var_names.email in body: owner_email = body[var_names.email] # check if the email address already exists as a lead if not settings.TEST_MODE: leads = helpers.get_email_leads() if owner_email not in leads: leads[owner_email] = datetime.datetime.now().strftime(static_vars.json_timestamp_format) helpers.update_email_leads(leads) request.session[var_names.registration_info] = {var_names.email: owner_email} reg_site_identifier = helpers.generate_random_string_key() reg_site_expiry = datetime.datetime.now() + datetime.timedelta(minutes=10) next_step_path = url_paths.register + '?step=1' resp = JsonResponse(next_step_path, safe=False) resp.set_cookie(var_names.registration_site_identifier, reg_site_identifier, expires=reg_site_expiry, secure=True, httponly=True, samesite='Strict') return resp else: return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401) 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 verify_org_region_subdomain_email(request): ''' Step 1 of organization registration process. Verify the subdomain and email are unique in the region. If the region itself needs to be changed, do so here. Send the url for the next step of the process as response. :param request: Http request :return: JSON response -> redirect url path ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: if var_names.registration_site_identifier not in request.COOKIES and\ var_names.registration_info not in request.session: return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401, safe=False) reg_info = request.session[var_names.registration_info] body = json.loads(request.body.decode()) region = body[var_names.host_region] subdomain = body[var_names.subdomain] owner_email = reg_info[var_names.email] if region == settings.REGION: req_body = {var_names.email: owner_email, var_names.subdomain: subdomain} if settings.TEST_MODE: req_stat, req_out = 200, lnm.msg_success else: req_stat, req_out = helpers.post_api_request( api_paths.org_subdomain_email_unique, req_body, lang=lang) if req_stat == 200: reg_info[var_names.subdomain] = subdomain request.session[var_names.registration_info] = reg_info return JsonResponse(url_paths.register + '?step=2', safe=False) else: return JsonResponse(req_out, status=400, safe=False) else: if settings.TEST_MODE: re_dir_base = static_vars.regional_test_mode_urls[region]['redirect_base'] elif settings.TEST_SERVER: re_dir_base = static_vars.regional_test_server_urls[region]['redirect_base'] else: re_dir_base = static_vars.regional_urls[region]['redirect_base'] re_dir_url = re_dir_base + url_paths.relative_url_registration_organization_switch_region + '?token=' +\ helpers.create_org_registration_switch_region_token(owner_email, subdomain) resp = JsonResponse(re_dir_url, status=200, safe=False) resp.delete_cookie(var_names.registration_site_identifier) del request.session[var_names.registration_info] return resp 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(['GET']) def initiate_cross_region_switch(request): ''' Part of step 2 when the region selected is not the region of the app. The request is redirected here to switch the region. :param request: Http request :return: JSON response -> redirect url path ''' if request.method == 'GET': lang = request_validator.get_user_language(request) try: if var_names.token in request.GET: owner_email, subdomain =\ helpers.read_org_registration_switch_region_token(request.GET.get(var_names.token)) reg_info = {var_names.email: owner_email} reg_site_identifier = helpers.generate_random_string_key() reg_site_expiry = datetime.datetime.now() + datetime.timedelta(minutes=10) if subdomain is None: # check if the email address already exists as a lead if not settings.TEST_MODE: leads = helpers.get_email_leads() if owner_email not in leads: leads[owner_email] = datetime.datetime.now().strftime(static_vars.json_timestamp_format) helpers.update_email_leads(leads) resp = redirect(url_paths.register + '?step=1', status=200, safe=False) else: req_body = {var_names.email: owner_email, var_names.subdomain: subdomain} if settings.TEST_MODE: req_stat, req_out = 200, lnm.msg_success else: req_stat, req_out = helpers.post_api_request( api_paths.org_subdomain_email_unique, req_body, lang=lang) if req_stat == 200: reg_info[var_names.subdomain] = subdomain resp = redirect(url_paths.register + '?step=2') else: resp = redirect(url_paths.register + '?step=1', status=200, safe=False) resp.set_cookie(var_names.registration_site_identifier, reg_site_identifier, expires=reg_site_expiry, secure=True, httponly=True, samesite='Strict') request.session[var_names.registration_info] = reg_info return resp 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 save_org_owner_personal_info(request): ''' Step 2 of organization registration process. Save the owner's personal information. Send a verification code to the owner's email. Send the url for the next step of the process as response. :param request: Http request :return: JSON response -> redirect url path ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: if var_names.registration_site_identifier not in request.COOKIES and\ var_names.registration_info not in request.session: return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401, safe=False) reg_info = request.session[var_names.registration_info] body = json.loads(request.body.decode()) owner_email = reg_info[var_names.email] vrf_body = {var_names.email: owner_email} if settings.TEST_MODE: vrf_stat, vrf_out = 200, 'code has been sent' else: vrf_stat, vrf_out = helpers.post_api_request( api_paths.user_verification_code_create, vrf_body, lang=lang) if vrf_stat == 200: reg_info = {**reg_info, **body} request.session[var_names.registration_info] = reg_info next_step_path = url_paths.register + '?step=3' return JsonResponse(next_step_path, safe=False) else: return JsonResponse(vrf_out, status=vrf_stat, 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 verify_org_owner(request): ''' Step 3 of the organization registration process. Verify if the code entered is correct or not. Send the url for the next step of the process as response. :param request: Http request :return: JSON response -> str ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: if var_names.registration_site_identifier not in request.COOKIES and\ var_names.registration_info not in request.session: return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401, safe=False) reg_info = request.session[var_names.registration_info] ip_address = request.META.get(static_vars.ip_address_attribute) body = json.loads(request.body.decode()) body[var_names.email] = reg_info[var_names.email] body[var_names.ip_address] = ip_address if settings.TEST_MODE: status, output = 200, 'registration_token' else: status, output = helpers.post_api_request(api_paths.user_verification_email_verify, body, lang=lang) if status == 200: request.session[var_names.registration_token] = output next_step_path = url_paths.register + '?step=4' return JsonResponse(next_step_path, safe=False) else: return JsonResponse(output, status=status, safe=False) except InvalidRequest as e: logging.exception(str(e)) return JsonResponse(str(e), status=400) 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 register_organization(request): ''' Step 4. Register the organization. :param request: Http request :return: JSON response -> str ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: if var_names.registration_site_identifier not in request.COOKIES and\ var_names.registration_info not in request.session: logging.error('Registration site identifier and/or session info missing') return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401, safe=False) elif var_names.registration_token not in request.session: logging.error('Registration token missing') return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401, safe=False) else: reg_tk = request.session[var_names.registration_token] reg_info = request.session[var_names.registration_info] ip_address = request.META.get(static_vars.ip_address_attribute) timezone = helpers.get_info_from_ip_address(ip_address, static_vars.ip_timezone_attribute) body = json.loads(request.body.decode()) body = { **body, **reg_info, var_names.ip_address: ip_address, var_names.timezone: timezone, var_names.token: reg_tk, var_names.profile_picture: client_data_path.get_taskcall_avatar_url() } if settings.TEST_MODE: status_reg, output_reg = 200, '1245' else: status_reg, output_reg = helpers.post_api_request(api_paths.org_create, body, lang=lang) if status_reg == 200: login_body = { var_names.email: body[var_names.email], var_names.password: body[var_names.password], var_names.ip_address: ip_address, var_names.access_method: static_vars.web, var_names.session_id: request.session.session_key } if settings.TEST_MODE: status_login, output_login = 200, test_data_users.user_authentication_details[ test_data_users.users_list[0][1]] else: status_login, output_login = helpers.post_api_request( api_paths.user_login, login_body, lang=lang) if status_login == 200: helpers.set_session_authorization_tokens(request, output_login) redirect_url = pages.incidents_url + '?flow=' + output_reg +\ '&tutorial=' + tutorial_names.welcome if helpers.has_only_stakeholder_rights(output_login, redirect_url): redirect_url = pages.status_dashboard_url resp = JsonResponse(redirect_url, safe=False) else: resp = JsonResponse(output_login, status=status_login, safe=False) resp.delete_cookie(var_names.registration_site_identifier) del request.session[var_names.registration_info] return resp else: return JsonResponse(output_reg, status=status_reg, safe=False) except InvalidRequest as e: logging.exception(str(e)) return JsonResponse(str(e), status=400) 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 verify_member_code(request): ''' Verifies if the code entered for a member's registration is correct or not. :param request: Http request :return: JSON response -> org id ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: body = json.loads(request.body.decode()) member_email = body[var_names.email] ip_address = request.META.get(static_vars.ip_address_attribute) body[var_names.ip_address] = ip_address if settings.TEST_MODE: status, output = 200, 'registration_token' else: status, output = helpers.post_api_request(api_paths.user_verification_email_verify, body, lang=lang) if status == 200: reg_site_identifier = helpers.generate_random_string_key() reg_site_expiry = datetime.datetime.now() + datetime.timedelta(minutes=10) request.session[var_names.registration_info] = {var_names.email: member_email} request.session[var_names.registration_token] = output next_step_path = url_paths.register_member + '?step=1' resp = JsonResponse(next_step_path, safe=False) resp.set_cookie(var_names.registration_site_identifier, reg_site_identifier, expires=reg_site_expiry, secure=True, httponly=True, samesite='Strict') return resp return JsonResponse(output, status=status, safe=False) except InvalidRequest as e: logging.exception(str(e)) return JsonResponse(str(e), status=400) 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 register_member(request): ''' Registers a member to an organization. :param request: Http request :return: JSON response -> str ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: if var_names.registration_site_identifier not in request.COOKIES and\ var_names.registration_info not in request.session: logging.error('Registration site identifier and/or session info missing') return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401, safe=False) elif var_names.registration_token not in request.session: logging.error('Registration token missing') return JsonResponse(lt.get_label(lnm.err_invalid_request, lang), status=401, safe=False) else: reg_tk = request.session[var_names.registration_token] reg_info = request.session[var_names.registration_info] ip_address = request.META.get(static_vars.ip_address_attribute) timezone = helpers.get_info_from_ip_address(ip_address, static_vars.ip_timezone_attribute) body = json.loads(request.body.decode()) body[var_names.email] = reg_info[var_names.email] body[var_names.ip_address] = ip_address body[var_names.timezone] = timezone body[var_names.token] = reg_tk body[var_names.profile_picture] = client_data_path.get_taskcall_avatar_url() if settings.TEST_MODE: status_reg, output_reg = 200, '5421' else: status_reg, output_reg = helpers.post_api_request(api_paths.org_members_register, body, lang=lang) if status_reg == 200: login_body = { var_names.email: body[var_names.email], var_names.password: body[var_names.password], var_names.ip_address: request.META.get(static_vars.ip_address_attribute), var_names.access_method: static_vars.web, var_names.session_id: request.session.session_key } if settings.TEST_MODE: status_login, output_login = 200, test_data_users.user_authentication_details[ test_data_users.users_list[0][1]] else: status_login, output_login = helpers.post_api_request( api_paths.user_login, login_body, lang=lang) if status_login == 200: helpers.set_session_authorization_tokens(request, output_login) redirect_url = pages.incidents_url # Only regular users have an onboarding flow. Stakeholders do not. if output_reg is not None: redirect_url = redirect_url + '?flow=' + output_reg + '&tutorial=' + tutorial_names.intro if helpers.has_only_stakeholder_rights(output_login, redirect_url): redirect_url = pages.status_dashboard_url resp = JsonResponse(redirect_url, safe=False) else: resp = JsonResponse(output_login, status=status_login, safe=False) resp.delete_cookie(var_names.registration_site_identifier) del request.session[var_names.registration_info] return resp else: return JsonResponse(output_reg, status=status_reg, safe=False) except InvalidRequest as e: logging.exception(str(e)) return JsonResponse(str(e), status=400) 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 register_member_and_login(request): ''' Registers a member to an organization and logs them in directly. This will only be allowed for SSO auto provisioning. :param request: Http request :return: JSON response -> str ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: # Ip address and subdomain should have been saved in the session in the originating request from TaskCall # to the sso vendor. Registration token, sso email and sso login post body should have been saved in the # session after the first GET response received from the vendor was processed. if var_names.ip_address in request.session and var_names.subdomain in request.session and\ var_names.registration_token in request.session and var_names.sso_email in request.session and\ var_names.sso_login_post_body in request.session: ip_address = request.session[var_names.ip_address] subdomain = request.session[var_names.subdomain] reg_tk = request.session[var_names.registration_token] email = request.session[var_names.sso_email] reg_body = json.loads(request.body.decode()) timezone = helpers.get_info_from_ip_address(ip_address, static_vars.ip_timezone_attribute) reg_body[var_names.email] = email reg_body[var_names.password] = helpers.generate_random_string_key(12) reg_body[var_names.ip_address] = ip_address reg_body[var_names.timezone] = timezone reg_body[var_names.token] = reg_tk reg_body[var_names.profile_picture] = client_data_path.get_taskcall_avatar_url() reg_body[var_names.subdomain] = subdomain if settings.TEST_MODE: status_reg, output_reg = 200, '5421' else: status_reg, output_reg = helpers.post_api_request( api_paths.org_members_register, reg_body, lang=lang) if status_reg == 200: login_body = request.session[var_names.sso_login_post_body] if settings.TEST_MODE: status_login, output_login = 200, test_data_users.user_authentication_details[ test_data_users.users_list[0][1]] else: status_login, output_login = helpers.post_api_request(api_paths.user_login, login_body) if status_login == 200: helpers.set_session_authorization_tokens(request, output_login) redirect_url = helpers.get_redirect_page(request) # Only regular users have an onboarding flow. Stakeholders do not. if output_reg is not None: redirect_url = redirect_url + '?flow=' + output_reg + '&tutorial=' + tutorial_names.intro if helpers.has_only_stakeholder_rights(output_login, redirect_url): redirect_url = pages.status_dashboard_url # delete all the session variables del request.session[var_names.ip_address] del request.session[var_names.subdomain] del request.session[var_names.registration_token] del request.session[var_names.sso_email] del request.session[var_names.sso_login_post_body] return JsonResponse(redirect_url, safe=False) else: error = output_login.replace("'", "") else: error = output_reg.replace("'", "") else: error = lt.get_label(lnm.err_unauthorized_access, lang) return JsonResponse(error, status=401, safe=False) except InvalidRequest as e: logging.exception(str(e)) return JsonResponse(str(e), status=400) except Exception as e: logging.exception(str(e)) return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=500, safe=False)