# By: Riasat Ullah # This file contains code for handling Datadog integration related views. from constants import api_paths, component_names as cnm, label_names as lnm, pages, static_vars, url_paths, var_names from context_manager import integrations_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 integrations import datadog from taskcallweb import settings from translators import label_translator as lt from utils import helpers, logging from validations import request_validator import json import requests @require_http_methods(['GET', 'POST']) def initiate_datadog_integration(request): ''' Initiates the Datadog integration process. :param request: Http request :return: Http response ''' if request.method == 'GET': region = request_validator.get_host_region(request) if settings.REGION != region and region == static_vars.aws_us_ohio: if 'site' in request.GET and var_names.domain in request.GET: site = request.GET['site'] domain = request.GET[var_names.domain] rdr_url = datadog.get_web_path_us_tc_datadog_authorization(site, domain) return redirect(rdr_url) if 'site' not in request.GET or var_names.domain not in request.GET: logging.error('Expected parameters "site" or "domain" is missing. ' + 'Denying access to Datadog integration authorization. Redirecting to login page.') return redirect(pages.login_url) else: if 'site' in request.GET: site = request.GET['site'] domain = request.GET[var_names.domain] request.session[datadog.str_datadog_site] = site request.session[datadog.str_datadog_domain] = domain 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_services, nav_bar_components ) if not has_edit_perm: raise exceptions.PermissionDenied context = integrations_context.get_datadog_authorization_context(lang) return render(request, pages.integrations_datadog_authorization, 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(lt.get_label(lnm.msg_success, lang), status=400, safe=False) else: body = json.loads(request.body.decode()) if datadog.str_datadog_site in request.session and\ datadog.str_datadog_domain in request.session and\ var_names.integration_name in body and var_names.service_ref_id in body: # variables sent from Datadog that have been stored in the session site = request.session[datadog.str_datadog_site] # integration details that have been received with this request serv_ref_id = body[var_names.service_ref_id] request.session[var_names.services] = body rdr_path = datadog.get_datadog_oauth_path(site, serv_ref_id) return JsonResponse(rdr_path, safe=False) else: return JsonResponse(lt.get_label(lnm.err_verification_failed, lang), status=400, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET']) def authorize_datadog_integration(request): ''' Handles the authorization of the Datadog integration process. :param request: Http request :return: Http response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) if datadog.str_datadog_site in request.session and var_names.services in request.session and\ var_names.service_ref_id in request.session[var_names.services] and\ var_names.integration_name in request.session[var_names.services]: dd_site = request.session[datadog.str_datadog_site] integ_details = request.session[var_names.services] serv_ref_id = integ_details[var_names.service_ref_id] integ_name = integ_details[var_names.integration_name] failed = True if 'error' in request.GET: logging.error = request.GET['error'] + ' - Datadog request authorization failed' else: try: if 'code' not in request.GET: logging.error('Code parameter was not received from Datadog') dd_oid = request.GET['dd_oid'] dd_creds = datadog.get_datadog_credentials() post_body = { 'client_id': dd_creds[var_names.client_id], 'client_secret': dd_creds[var_names.client_secret], 'redirect_uri': datadog.get_datadog_to_tc_redirect_uri(), 'grant_type': 'authorization_code', 'code_verifier': serv_ref_id, 'code': request.GET['code'] } resp = requests.post(datadog.datadog_access_token_path.format(dd_site), data=post_body) status = resp.status_code data = resp.json() dd_acc_tok = data['access_token'] dd_ref_tok = data['refresh_token'] if status == 200 and 'error' not in data: # Only if the Datadog account ID is not already associated to this organization, # then create an entry in the organization_integration_type_details. Do not duplicate. dupe_post_body = {var_names.external_id: dd_oid, var_names.vendor_endpoint: dd_site, var_names.access_token: dd_acc_tok, var_names.refresh_token: dd_ref_tok} dupe_check_status, dupe_check_output = helpers.post_api_request( api_paths.integrations_datadog_tokens_update, dupe_post_body, request, lang=lang ) itg_post_body = { var_names.service_ref_id: serv_ref_id, var_names.integration_type: static_vars.integ_type_datadog, var_names.integration_name: integ_name, var_names.vendor_endpoint_name: dd_oid, var_names.external_id: dd_oid } if dupe_check_status == 200 and not dupe_check_output: itg_post_body[var_names.external_info] = { var_names.vendor_endpoint: dd_site, var_names.access_token: dd_acc_tok, var_names.refresh_token: dd_ref_tok } itg_add_status, itg_add_output = helpers.post_api_request( api_paths.services_integrations_add, itg_post_body, request, lang=lang) if itg_add_status != 200: logging.error(itg_add_output) failed = False except Exception as e: logging.exception(str(e)) if var_names.services in request.session: del request.session[var_names.services] if datadog.str_datadog_site in request.session: del request.session[datadog.str_datadog_site] if datadog.str_datadog_domain in request.session: del request.session[datadog.str_datadog_domain] if failed: logging.exception(lnm.err_verification_failed) return redirect(url_paths.configurations_services + '/' + serv_ref_id + '?error=' + lt.get_label(lnm.err_unauthorized_access, lang)) else: return redirect(url_paths.configurations_services + '/' + serv_ref_id + '?tab=integrations&message=' + lt.get_label(lnm.msg_service_integration_added, lang)) else: return redirect(url_paths.configurations_services)