# By: Riasat Ullah # This file contains code for serving organization specific pages. from constants import api_paths, component_names as cnm, label_names as lnm, pages, static_vars, var_names from context_manager import components_context, configurations_context, users_context from django.core import exceptions from django.http import 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 from translators import label_translator as lt from utils import client_data_path, helpers, logging, s3 from validations import request_validator import botocore.exceptions import json @require_http_methods(['GET', 'POST']) def get_details(request): ''' GET: Gets the organization details page. POST: Gets the organization details as json. :param request: Http request :return: Http response for GET | JSON response for POST ''' 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) has_view_perm, has_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_organization, nav_bar_components ) if not has_view_perm: raise exceptions.PermissionDenied context = configurations_context.get_organization_info_context(lang, nav_bar_components) context[var_names.has_edit_permission] = has_edit_perm context[var_names.user_lang_items] = components_context.get_languages_content(context) # special features context context[var_names.has_sso_permission] =\ True if cnm.feat_sso_authentication in component_features else False if settings.TEST_MODE: context[var_names.taskcall_domain] = static_vars.regional_test_server_urls[settings.REGION][ var_names.domain] status, output = 200, test_data_organization.allowed_user_roles else: context[var_names.taskcall_domain] = static_vars.regional_urls[settings.REGION][var_names.domain] 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.organization_info_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: return JsonResponse(test_data_organization.organization_details, safe=False) else: body = dict() status, output = helpers.post_api_request(api_paths.org_details, 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_details(request): ''' Edit the organization details. :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('Details have been edited', safe=False) else: body = json.loads(request.body.decode()) status, output = helpers.post_api_request(api_paths.org_details_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 update_subdomain(request): ''' Add organization admins. :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('Subdomain has been added successfully', safe=False) else: body = json.loads(request.body.decode()) status, output = helpers.post_api_request(api_paths.org_subdomain_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_full_domain(request): ''' Get the full domain of the organization including the regional URL. :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( 'apollo.' + static_vars.regional_test_server_urls[settings.REGION][var_names.domain], safe=False ) else: body = dict() status, output = helpers.post_api_request(api_paths.org_full_domain, 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_sso_settings(request): ''' Get the SSO settings of an 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): if settings.TEST_MODE: status, output = 200, test_data_organization.sso_settings else: body = dict() status, output = helpers.post_api_request(api_paths.org_sso_settings, body, request, lang=lang) if status == 200: output[var_names.redirect_url] = settings.REDIRECT_BASE if status == 404: status, output = 200, None 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 save_sso_settings(request): ''' Save the new SSO setting of an 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()) if settings.TEST_MODE: return JsonResponse('Success', safe=False) else: status, output = helpers.post_api_request(api_paths.org_sso_settings_save, 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 save_account_closure_verification(request): ''' Saves verification code for an account closure request. :param request: HttpRequest :return: JsonResponse ''' 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 sent', safe=False) else: body = dict() status, output = helpers.post_api_request( api_paths.org_closure_verification_save, 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 verify_account_closure(request): ''' Checks if the verification code provided for account closure verifies or not. :param request: HttpRequest :return: JsonResponse ''' 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, 'closure_token' else: ip_address = request.META.get(static_vars.ip_address_attribute) body[var_names.ip_address] = ip_address status, output = helpers.post_api_request(api_paths.org_closure_verification_verify, body, request, lang=lang) if status == 200: request.session[var_names.closure_token] = output 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 close_account(request): ''' Closes an organization's account permanently. :param request: HttpRequest :return: JsonResponse ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): cls_tk = request.session[var_names.closure_token] if settings.TEST_MODE: return JsonResponse('Account has been closed', safe=False) else: # Retrieve information needed to delete files stored on s3. account_id, live_call_routing_list = None, None # Retrieve the account ID of the organization to delete user profile pictures. acc_status, acc_output = helpers.post_api_request(api_paths.org_details, dict(), request, lang=lang) if acc_status == 200: account_id = acc_output[var_names.account_id] # Retrieve the list of live call routing specifications of the organization for deleting their audio. body = {var_names.data_type: [var_names.live_call_routing]} lcr_status, lcr_output = helpers.post_api_request(api_paths.org_list_maker, body, request, lang=lang) if lcr_status == 200 and lcr_output is not None and len(lcr_output) > 0: live_call_routing_list = lcr_output if acc_status == 200 and lcr_status == 200: # Close the organization here. This is the main part of the process. ip_address = request.META.get(static_vars.ip_address_attribute) body = {var_names.ip_address: ip_address, var_names.token: cls_tk} status, output = helpers.post_api_request(api_paths.org_closure_close, body, request, lang=lang) # Delete all the s3 files of the organization once it is successfully closed in the backend. if status == 200: try: # Delete user profile pictures img_bucket, img_fldr = client_data_path.get_organization_profile_pictures_path(account_id) logging.info('Deleting organization user profile pictures - ' + img_fldr) s3.delete_folder_files(img_bucket, img_fldr) # Delete live call routing audio if live_call_routing_list is not None: for item in live_call_routing_list: lcr_ref_key = item[1] grt_bucket, grt_fldr = client_data_path.get_greeting_audio_path(lcr_ref_key, '')[:2] end_bucket, end_fldr = client_data_path.get_ending_audio_path(lcr_ref_key, '')[:2] s3.delete_single_file(grt_bucket, grt_fldr) s3.delete_single_file(end_bucket, end_fldr) except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception('Failed to delete organization s3 files after closing its account') return JsonResponse(output, status=status, safe=False) return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=500, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False)