# By: Riasat Ullah # This file contains views for external status pages. from constants import api_paths, component_names as cnm, label_names as lnm, pages, static_vars, var_names from context_manager import itsm_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_status_dashboard 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 configuration as configs import filetype import json @require_http_methods(['GET', 'POST']) def status_pages_list(request): ''' GET: Gets Status Pages list page. POST: Get the list of status pages associated with the organization. :param request: Http request :return: GET: html page ''' 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_status_pages, nav_bar_components ) if not has_view_perm: raise exceptions.PermissionDenied context = itsm_context.get_status_pages_list_context(lang, nav_bar_components) context[var_names.has_edit_permission] = has_edit_perm return render(request, pages.status_pages_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): if settings.TEST_MODE: return JsonResponse(test_data_status_dashboard.status_pages_list, safe=False) else: body = dict() status, output = helpers.post_api_request(api_paths.status_pages_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(['GET', 'POST']) def status_page_details(request, page_ref_id=None): ''' GET: Returns the status page details page POST: Gets the status page details :param request: Http request :param page_ref_id: reference ID of the page :return: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): try: 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_status_pages, nav_bar_components ) if not has_view_perm or (page_ref_id is None and not has_edit_perm): raise exceptions.PermissionDenied context = itsm_context.get_status_pages_details_context(lang, nav_bar_components) context[var_names.context] = json.dumps({var_names.page_ref_id: page_ref_id}) context[var_names.has_edit_permission] = has_edit_perm context[var_names.has_team_permission] = request_validator.get_team_permission_status(request) context[var_names.has_private_status_pages_permission] = \ True if cnm.feat_status_pages_private in component_features else False if page_ref_id is None: context[var_names.is_new] = True if 'tab' in request.GET: context[var_names.tab_name] = request.GET.get('tab') return render(request, pages.status_pages_details_page, context=context) except ValueError as e: logging.exception(str(e)) return render(request, pages.status_pages_list_page) 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_status_dashboard.status_page_details[page_ref_id], safe=False) else: body = {var_names.page_ref_id: page_ref_id} status, output = helpers.post_api_request(api_paths.status_pages_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 create_status_page(request): ''' Creates a new status page. :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]) page_url = body[var_names.url] logo_name = body[var_names.logo_filename] icon_name = body[var_names.icon_filename] cover_name = body[var_names.cover_image_filename] del body[var_names.logo_filename] del body[var_names.icon_filename] del body[var_names.cover_image_filename] if settings.TEST_MODE: status, output = 200, 'new_ref_id' else: status, output = helpers.post_api_request(api_paths.status_pages_create, body, request, lang=lang) ################################### # Handle s3 image file uploads here ################################### if status == 200 and not settings.TEST_MODE and len(request.FILES) > 0: # Prepare the new payload that will be sent to the rest server to update the image location new_payload = {var_names.page_ref_id: output} uploaded_image_urls = dict() page_domain = helpers.get_status_page_domain_from_url(page_url) if page_domain is not None: for img_type in request.FILES: if img_type == var_names.logo: img_name, var_url = logo_name, var_names.logo_url elif img_type == var_names.icon: img_name, var_url = icon_name, var_names.icon_url elif cover_name is not None: img_name, var_url = cover_name, var_names.cover_image_url else: continue img_bucket, img_folder, img_key, img_url =\ client_data_path.get_status_page_image_path(page_domain, output, img_name, img_type) try: logging.info('Uploading status page image - ' + img_key) s3.upload_media_file(img_bucket, img_key, request.FILES[img_type]) uploaded_image_urls[var_url] = img_url except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception(lnm.msg_status_page_created_with_image_issues) return JsonResponse(lt.get_label(lnm.msg_status_page_created_with_image_issues, lang), status=207, safe=False) # Update the image file location information in the backend if len(uploaded_image_urls) > 0: new_payload[var_names.images] = uploaded_image_urls helpers.post_api_request(api_paths.status_pages_images_add_info, new_payload, 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_status_page(request): ''' Edits an existing status page. :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]) page_ref_id = body[var_names.page_ref_id] page_url = body[var_names.url] is_published = body[var_names.is_published] logo_name = body[var_names.logo_filename] icon_name = body[var_names.icon_filename] cover_name = body[var_names.cover_image_filename] has_cover_changed = body[var_names.has_cover_image_changed] del body[var_names.url] del body[var_names.url_extension] del body[var_names.logo_filename] del body[var_names.icon_filename] del body[var_names.cover_image_filename] del body[var_names.has_logo_changed] del body[var_names.has_icon_changed] del body[var_names.has_cover_image_changed] if settings.TEST_MODE: status, output = 200, 'Success' else: status, output = helpers.post_api_request(api_paths.status_pages_edit, body, request, lang=lang) ################################### # Handle s3 image file uploads here ################################### if status == 200 and not settings.TEST_MODE: pub_img_urls = None if not is_published and len(request.FILES) > 0: pub_sts, pub_output = helpers.post_api_request( api_paths.status_pages_images_published, {var_names.page_ref_id: page_ref_id}, request, lang=lang ) if pub_sts == 200: pub_img_urls = pub_output # Prepare the new payload that will be sent to the rest server to update the audio location new_payload = {var_names.page_ref_id: page_ref_id} uploaded_image_urls = dict() page_domain = helpers.get_status_page_domain_from_url(page_url) if has_cover_changed and var_names.cover_image not in request.FILES: uploaded_image_urls[var_names.cover_image_url] = None s3_bucket = client_data_path.get_client_data_s3_bucket() s3_key = client_data_path.get_status_page_image_type_folder_path(page_ref_id, var_names.cover_image) s3.delete_folder_files(s3_bucket, s3_key) if page_domain is not None: for img_type in request.FILES: if img_type == var_names.logo: img_name, var_url = logo_name, var_names.logo_url elif img_type == var_names.icon: img_name, var_url = icon_name, var_names.icon_url elif cover_name is not None: img_name, var_url = cover_name, var_names.cover_image_url else: continue img_bucket, img_folder, img_key, img_url =\ client_data_path.get_status_page_image_path(page_domain, page_ref_id, img_name, img_type) try: logging.info('Cleaning up old status page images - ' + img_folder) keep_files = [] if not is_published: pub_url = None if pub_img_urls is not None: if img_type == var_names.logo and var_names.logo_url in pub_img_urls: pub_url = pub_img_urls[var_names.logo_url] elif img_type == var_names.icon and var_names.icon_url in pub_img_urls: pub_url = pub_img_urls[var_names.icon_url] elif img_type == var_names.cover_image and \ var_names.cover_image_url in pub_img_urls: pub_url = pub_img_urls[var_names.cover_image_url] if pub_url is not None: keep_files.append(client_data_path.get_status_page_image_s3_key_from_url(pub_url)) s3.delete_folder_files(img_bucket, img_folder, files_to_keep=keep_files) logging.info('Uploading status page image - ' + img_key) s3.upload_media_file(img_bucket, img_key, request.FILES[img_type]) uploaded_image_urls[var_url] = img_url except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception(lnm.msg_status_page_created_with_image_issues) return JsonResponse(lt.get_label(lnm.msg_status_page_created_with_image_issues, lang), status=207, safe=False) # Update the image file location information in the backend if len(uploaded_image_urls) > 0: new_payload[var_names.images] = uploaded_image_urls helpers.post_api_request(api_paths.status_pages_images_add_info, new_payload, 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_status_page(request): ''' Deletes a status page. :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()) page_ref_id = body[var_names.page_ref_id] if settings.TEST_MODE: status, output = 200, 'Success' else: status, output = helpers.post_api_request(api_paths.status_pages_delete, body, request, lang=lang) if settings.TEST_MODE: return JsonResponse(output, status=status, safe=False) if status == 200: try: s3_bucket = client_data_path.get_client_data_s3_bucket() s3_key = client_data_path.status_pages_images.format(page_ref_id) s3.delete_folder_files(s3_bucket, s3_key) except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception('Failed to delete status page image files - ' + str(page_ref_id)) finally: output = lt.get_label(lnm.msg_status_page_deleted, 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 status_pages_endpoint(request): ''' Get the final endpoint that the live status page would be served from. It retrieves the organization domain and combines it with the standard status page extension. :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: itsm_url = static_vars.regional_test_mode_urls[settings.REGION]['itsm_base_url'] elif settings.TEST_SERVER: itsm_url = static_vars.regional_test_server_urls[settings.REGION]['itsm_base_url'] else: itsm_url = static_vars.regional_urls[settings.REGION]['itsm_base_url'] if settings.TEST_MODE: status, output = 200, 'apollo.taskcallapp.com' else: body = dict() status, output = helpers.post_api_request(api_paths.org_full_domain, body, request, lang=lang) if status == 200: output = itsm_url.format(output.split('.')[0]) + '/' + static_vars.status_pages_extension + '/' 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 status_pages_live_endpoint(request): ''' Get the live url of a status page. combines it with the standard status page extension. :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()) page_ref_id = body[var_names.page_ref_id] if settings.TEST_MODE: if page_ref_id in test_data_status_dashboard.status_page_details and \ test_data_status_dashboard.status_page_details[page_ref_id][var_names.is_published]: return JsonResponse(test_data_status_dashboard.status_page_details[page_ref_id][var_names.url], 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.status_pages_live_url, 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_status_page_image_files(request): ''' Validates that the status page 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): audio_files = request.FILES.getlist('files[]') for item in audio_files: file = filetype.guess(item) if file is not None and file.extension not in configs.STANDARD_IMAGE_EXTENSIONS and\ file.mime not in configs.STANDARD_IMAGE_MIME_TYPES: return JsonResponse(lt.get_label(lnm.err_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 status_page_components(request, page_ref_id): ''' Gets the published components (categories and business services) of a status page. Impacted business services on the post pages are selected from this list. :param request: Http request :param page_ref_id: reference ID of the status page whose incidents are being requested :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = {var_names.page_ref_id: page_ref_id} if settings.TEST_MODE: return JsonResponse( test_data_status_dashboard.status_page_details[page_ref_id][var_names.business_services], safe=False ) else: status, output = helpers.post_api_request(api_paths.status_pages_components, 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 publish_status_page(request): ''' Publish a status page. :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, lt.get_label(lnm.msg_status_page_published, lang) else: status, output = helpers.post_api_request(api_paths.status_pages_publish, 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 unpublish_status_page(request): ''' Unpublish a status page. :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, lt.get_label(lnm.msg_status_page_unpublished, lang) else: status, output = helpers.post_api_request(api_paths.status_pages_unpublish, 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)