# By: Riasat Ullah # This file contains views for handling Jira authorization pages. from constants import api_paths, label_names as lnm, pages, static_vars, url_paths, var_names from django.http import JsonResponse from django.shortcuts import redirect from django.views.decorators.http import require_http_methods from integrations import jira from system_tests.test_data import test_data_services from taskcallweb import settings from translators import label_translator as lt from utils import helpers, logging, s3 from validations import request_validator import json import requests @require_http_methods(['GET']) def redirect_to_jira_cloud_authorization_page(request): ''' Redirects to the Jira cloud authorization page. :param request: Http request :return: Http response ''' if request.method == 'GET': try: srv_ref = request.session[var_names.services][var_names.service_ref_id] jira_creds = s3.read_json(jira.jira_s3_bucket, jira.jira_s3_key) state = settings.REGION + '+' + srv_ref jira_oauth_url = jira.jira_cloud_oauth_path_format.format(jira_creds[jira.str_jira_client_id], state) return redirect(jira_oauth_url) except Exception as e: logging.exception(str(e)) return redirect(pages.login_url) @require_http_methods(['GET']) def authorize_jira_cloud_integration(request): ''' Handles the authorization of the Jira cloud integration process. :param request: Http request :return: Http response ''' if request.method == 'GET': if 'code' not in request.GET and 'state' not in request.GET: logging.error('Expected parameters "code" and "state" are missing. Denying access to Jira integration ' + 'authorization. Redirecting to login page.') return redirect(pages.login_url) else: code = request.GET['code'] state = request.GET['state'] state_region, state_srv_ref = state.split(' ') if state_region != settings.REGION and state_region == static_vars.aws_us_ohio: rdr_url = jira.get_web_path_us_tc_jira_cloud_authorization(code, state) return redirect(rdr_url) if not request_validator.user_in_session(request) or \ (var_names.services not in request.session and var_names.service_ref_id not in request.session[var_names.services] and not (var_names.integration_name in request.session[var_names.services] or var_names.integration_key in request.session[var_names.services])): logging.error('Expected service session variables were not found. Denying access to Jira integration ' + 'authorization. Redirecting to login page.') return redirect(pages.login_url) else: lang = request_validator.get_user_language(request) serv_session = request.session[var_names.services] srv_ref = serv_session[var_names.service_ref_id] failed = True if state_srv_ref == srv_ref: try: jira_creds = s3.read_json(jira.jira_s3_bucket, jira.jira_s3_key) post_body = { jira.str_jira_grant_type: 'authorization_code', jira.str_jira_client_id: jira_creds[jira.str_jira_client_id], jira.str_jira_client_secret: jira_creds[jira.str_jira_client_secret], jira.str_jira_code: code, jira.str_jira_redirect_uri: jira.jira_taskcall_callback_url } # Get Jira authorization and refresh token. Although an access token is provided in # this response, it cannot be used. The token has to be refreshed. oauth_resp = requests.post(jira.jira_token_retrieval_path, json=post_body) if oauth_resp.status_code == 200: oauth_data = oauth_resp.json() ref_req_body = { "grant_type": "refresh_token", "client_id": jira_creds[jira.str_jira_client_id], "client_secret": jira_creds[jira.str_jira_client_secret], "refresh_token": oauth_data[jira.str_jira_refresh_token] } ref_resp = requests.post(jira.jira_token_retrieval_path, json=ref_req_body) if ref_resp.status_code == 200: ref_data = ref_resp.json() access_token = ref_data[jira.str_jira_access_token] refresh_token = ref_data[jira.str_jira_refresh_token] # get the cloud ID from Jira with the access token cloud_id_resp = requests.get( jira.jira_cloud_id_retrieval_path, headers={"Authorization": "Bearer " + access_token, "Accept": "application/json"} ) if cloud_id_resp.status_code == 200: cloud_id_data = cloud_id_resp.json() jira_cloud_id = cloud_id_data[0][jira.str_jira_id] # Only if the Jira cloud 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: jira_cloud_id, var_names.access_token: access_token, var_names.refresh_token: refresh_token} dupe_check_status, dupe_check_output = helpers.post_api_request( api_paths.integrations_jira_cloud_tokens_update, dupe_post_body, request, lang=lang ) serv_session[var_names.vendor_endpoint_name] = jira_cloud_id if dupe_check_status == 200: if dupe_check_output: if var_names.integration_key in serv_session: return redirect(url_paths.configurations_services + '/' + srv_ref + '?tab=integrations&message=' + lt.get_label(lnm.msg_jira_cloud_reauthorized, lang)) else: serv_session[var_names.external_id] = jira_cloud_id serv_session[var_names.external_info] = { var_names.access_token: access_token, var_names.refresh_token: refresh_token } request.session[var_names.services] = serv_session failed = False except Exception as e: logging.exception(str(e)) if failed: logging.exception(lnm.err_verification_failed) return redirect(url_paths.configurations_services + '/' + srv_ref + '?error=' + lt.get_label(lnm.err_unauthorized_access, lang)) else: return redirect(url_paths.configurations_services + '/' + srv_ref + '?tab=integrations&modal=jira&integration=' + serv_session[var_names.integration_name]) @require_http_methods(['POST']) def get_jira_cloud_meta_data(request): ''' Get available Jira projects and issue types. :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): body = json.loads(request.body.decode()) if var_names.integration_key in body: post_body = body else: # The code is handled this way here in order to facilitate the process set up in the jira authorization # view to avoid duplicate entries of jira cloud ID in the organization_integration_type_details # table for the same organization. serv_session = request.session[var_names.services] post_body = {var_names.external_id: serv_session[var_names.vendor_endpoint_name]} if var_names.external_info in serv_session: post_body[var_names.access_token] = serv_session[var_names.external_info][var_names.access_token] post_body[var_names.refresh_token] = serv_session[var_names.external_info][var_names.refresh_token] if settings.TEST_MODE: return JsonResponse([test_data_services.jira_projects, test_data_services.jira_issue_types, test_data_services.jira_statuses], safe=False) else: status, output = helpers.post_api_request(api_paths.integrations_jira_cloud_meta_data, post_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_jira_cloud_integration(request): ''' Creates a new Jira integration. :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) and var_names.services in request.session and\ var_names.integration_type in request.session[var_names.services] and\ request.session[var_names.services][var_names.integration_type] == static_vars.integ_type_jira_cloud and\ var_names.service_ref_id in request.session[var_names.services] and\ var_names.integration_name in request.session[var_names.services]: req_body = json.loads(request.body.decode()) serv_session = request.session[var_names.services] post_body = { var_names.service_ref_id: serv_session[var_names.service_ref_id], var_names.integration_type: static_vars.integ_type_jira_cloud, var_names.integration_name: serv_session[var_names.integration_name], var_names.vendor_endpoint_name: serv_session[var_names.vendor_endpoint_name], var_names.additional_info: req_body[var_names.additional_info] } if var_names.external_id in serv_session: post_body[var_names.external_id] = serv_session[var_names.external_id] post_body[var_names.external_info] = serv_session[var_names.external_info] if settings.TEST_MODE: return JsonResponse('Action performed successfully', safe=False) else: status, output = helpers.post_api_request(api_paths.services_integrations_add, post_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_jira_cloud_integration(request): ''' Edits the configuration of an existing Jira integration. :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('Action performed successfully', safe=False) else: status, output = helpers.post_api_request(api_paths.services_integrations_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(['GET']) def initiate_jira_cloud_reauthorization(request): ''' Initiates the reauthorization of Jira Cloud integration of an organization. :param request: Http request :return: Http response ''' if request.method == 'GET': try: if 'service' not in request.GET or 'integration' not in request.GET: logging.error('Expected parameters "service" and "integration" are missing. ' + 'Denying access to Jira integration reauthorization. Redirecting to login page.') return redirect(pages.login_url) else: srv_ref = request.GET['service'] integ_key = request.GET['integration'] request.session[var_names.services] = { var_names.service_ref_id: srv_ref, var_names.integration_key: integ_key } jira_creds = s3.read_json(jira.jira_s3_bucket, jira.jira_s3_key) state = settings.REGION + '+' + srv_ref jira_oauth_url = jira.jira_cloud_oauth_path_format.format(jira_creds[jira.str_jira_client_id], state) return redirect(jira_oauth_url) except Exception as e: logging.exception(str(e)) return redirect(pages.login_url)