# By: Riasat Ullah # This file serves ITSM external SSO related pages and data. 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.core.cache import cache from django.http import JsonResponse from django.shortcuts import redirect, render from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from onelogin.saml2.auth import OneLogin_Saml2_Auth from taskcallweb import settings from system_tests.test_data import test_data_external from translators import label_translator as lt from utils import helpers, logging, sso_manager from validations import request_validator import configuration as configs import json @require_http_methods(['GET', 'POST']) def list_external_sso(request): ''' GET: Returns the external SSO list page. POST: Gets the external SSO list data. :param request: Http request :return: Http response ''' 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_itsm_sso, nav_bar_components ) if not has_view_perm: raise exceptions.PermissionDenied context = itsm_context.get_external_sso_list_context(lang, nav_bar_components) context[var_names.has_edit_permission] = has_edit_perm return render(request, pages.external_sso_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_external.external_sso_list, safe=False) else: status, output = helpers.post_api_request(api_paths.external_sso_list, dict(), 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 get_external_sso_details(request, sso_ref_id=None): ''' GET: Returns the external SSO details page. POST: Gets the details of an external SSO configuration. :param request: Http request :param sso_ref_id: reference ID of the SSO configuration :return: Http response ''' 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_itsm_sso, nav_bar_components ) if not has_edit_perm: raise exceptions.PermissionDenied context = itsm_context.get_external_sso_details_context(lang, nav_bar_components) context[var_names.context] = json.dumps({var_names.sso_ref_id: sso_ref_id}) context[var_names.has_edit_permission] = has_edit_perm return render(request, pages.external_sso_details_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): body = json.loads(request.body.decode()) if settings.TEST_MODE: for item in test_data_external.external_sso_list: if item[var_names.sso_ref_id] == body[var_names.sso_ref_id]: return JsonResponse(item, safe=False) return JsonResponse(lt.get_label(lnm.err_unknown_resource, lang), status=404) else: status, output = helpers.post_api_request(api_paths.external_sso_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_external_sso(request): ''' Create a new external SSO configuration. :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 settings.TEST_MODE: return JsonResponse(lt.get_label(lnm.msg_success, lang), safe=False) else: status, output = helpers.post_api_request(api_paths.external_sso_create, 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_external_sso(request): ''' Edit an external SSO configuration. :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 settings.TEST_MODE: return JsonResponse(lt.get_label(lnm.msg_success, lang), safe=False) else: status, output = helpers.post_api_request(api_paths.external_sso_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 delete_external_sso(request): ''' Delete an external SSO configuration. :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 settings.TEST_MODE: return JsonResponse(lt.get_label(lnm.msg_success, lang), safe=False) else: status, output = helpers.post_api_request(api_paths.external_sso_delete, 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 login_with_sso(request): ''' GET: Fetches the SSO segment of the login page. POST: Processes the logging in with SSO :param request: Http request :return: Http response ''' tc_domain = static_vars.regional_test_server_urls[settings.REGION][var_names.domain]\ if settings.TEST_MODE else static_vars.regional_urls[settings.REGION][var_names.domain] if request.method == 'POST': lang = request_validator.get_user_language(request) context = itsm_context.get_itsm_login_context(lang) context[var_names.method] = static_vars.sso context[var_names.taskcall_domain] = tc_domain try: page_type = request.POST[var_names.data_type] item_id = request.POST[var_names.item_id] if not request.session.session_key: request.session.save() request.session.set_expiry(2592000) if page_type == var_names.status_pages: body = {var_names.page_ref_id: item_id} status, output = helpers.post_api_request('', body) else: raise PermissionError request.method = 'GET' if status == 200: login_url = None # Handle Google SSO if output[var_names.integration_type] == static_vars.google_sso: google_sso_creds = sso_manager.get_sso_credentials(static_vars.google_sso) gd_response = cache.get(sso_manager.var_google_discovery_data) if gd_response is None: gd_status, gd_response = helpers.send_get_request(sso_manager.google_discovery_url) if gd_status == 200: cache.set(sso_manager.var_google_discovery_data, gd_response, timeout=configs.SSO_GOOGLE_DISCOVERY_CACHE_TIMEOUT) else: gd_response = None if gd_response is not None: login_url = sso_manager.google_oauth_url.format( gd_response[sso_manager.var_authorization_endpoint], google_sso_creds[var_names.client_id], sso_manager.taskcall_google_sso_redirect_uri, request.session.session_key, output[var_names.vendor_subdomain] ) # Handle Azure AD SSO elif output[var_names.integration_type] == static_vars.azure_ad_sso: azure_sso_creds = sso_manager.get_sso_credentials(static_vars.azure_ad_sso) azd_status, azd_response = helpers.send_get_request( sso_manager.azure_discovery_url.format(output[var_names.vendor_subdomain])) if azd_status == 200: minted_state = request.session.session_key minted_nonce = helpers.generate_random_string_key(59) login_url = sso_manager.azure_oauth_url.format( azd_response[sso_manager.var_authorization_endpoint], azure_sso_creds[var_names.client_id], sso_manager.taskcall_azure_sso_redirect_uri, minted_state, minted_nonce, output[var_names.vendor_subdomain] ) # set the cache data for az_cache_data = {var_names.ip_address: request.META.get(static_vars.ip_address_attribute), var_names.subdomain: output[var_names.subdomain], var_names.nonce: minted_nonce} new_key = helpers.make_cache_key(static_vars.cache_prefix_azure_sso, minted_state) cache.set(new_key, json.dumps(az_cache_data), timeout=configs.SSO_TEMP_VARIABLE_CACHE_TIMEOUT) else: context[var_names.error_message] = lt.get_label( lnm.err_sso_settings_vendor_domain_invalid, lang) return render(request, pages.external_sso_login_page, context=context) # Handle Okta SSO elif output[var_names.integration_type] == static_vars.okta: okta_status, okta_response = helpers.send_get_request( sso_manager.okta_discovery_url.format(output[var_names.vendor_subdomain])) if okta_status == 200: # set the token endpoint in session request.session[sso_manager.var_okta_token_endpoint] =\ okta_response[sso_manager.var_token_endpoint] request.session[sso_manager.var_okta_introspection_endpoint] =\ okta_response[sso_manager.var_introspection_endpoint] request.session[sso_manager.var_okta_client_id] =\ output[var_names.additional_info][var_names.client_id] request.session[sso_manager.var_okta_client_secret] =\ output[var_names.additional_info][var_names.client_secret] login_url = sso_manager.okta_oauth_url.format( okta_response[sso_manager.var_authorization_endpoint], output[var_names.additional_info][var_names.client_id], sso_manager.taskcall_okta_redirect_uri, request.session.session_key ) # Handle SAML 2.0 SSO elif output[var_names.integration_type] == static_vars.saml: acs_url = request.get_host() + '?subdomain=' + output[var_names.subdomain] saml_settings = sso_manager.get_saml_settings( acs_url, acs_url, output[var_names.saml_certificate], output[var_names.login_url], output[var_names.entity_id] ) req = { 'http_host': request.META['HTTP_HOST'], 'script_name': acs_url, 'server_port': request.META['SERVER_PORT'], 'get_data': request.GET.copy(), 'post_data': request.POST.copy() } login_url = OneLogin_Saml2_Auth(req, saml_settings).login() # set the cache data for SAML saml_cache_data = {var_names.ip_address: request.META.get(static_vars.ip_address_attribute), var_names.subdomain: output[var_names.subdomain], sso_manager.var_saml_settings: saml_settings} if var_names.mobile_app_redirect_url in request.session and \ var_names.mobile_app_push_token in request.session: saml_cache_data[var_names.mobile_app_redirect_url] = \ request.session[var_names.mobile_app_redirect_url] saml_cache_data[var_names.mobile_app_push_token] = \ request.session[var_names.mobile_app_push_token] cache.set(static_vars.cache_prefix_saml_sso, json.dumps(saml_cache_data), timeout=configs.SSO_TEMP_VARIABLE_CACHE_TIMEOUT) if login_url is not None: return redirect(login_url) else: context[var_names.error_message] = lt.get_label(lnm.err_system_error, lang) return render(request, pages.external_sso_login_page, context=context) else: context[var_names.error_message] = output.replace("'", "") return render(request, pages.external_sso_login_page, context=context) except Exception as e: logging.exception(str(e)) context[var_names.error_message] = lt.get_label(lnm.err_system_error, lang) return render(request, pages.external_sso_login_page, context=context) @require_http_methods(['GET']) def google_sso_authentication(request): ''' Handles the process of authenticating a user through Google SSO. The request for this view must originate from Google. :param request: Http request from Google :return: Http response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) context = itsm_context.get_itsm_login_context(lang) err_message = lnm.err_system_error try: # Ensure that the 'state' value is the same as that sent from originating request from TaskCall. # Also make sure that the ip address and subdomain values that should have been saved in the session # prior to making the originating request are also there. if request.GET.get('state') == request.session.session_key and var_names.ip_address in request.session\ and var_names.subdomain in request.session: code = request.GET.get('code') gd_response = cache.get(sso_manager.var_google_discovery_data) if gd_response is None: gd_status, gd_response = helpers.send_get_request(sso_manager.google_discovery_url) if gd_status == 200: cache.set(sso_manager.var_google_discovery_data, gd_response, timeout=configs.SSO_GOOGLE_DISCOVERY_CACHE_TIMEOUT) else: gd_response = None if gd_response is not None: token_endpoint = gd_response[sso_manager.var_token_endpoint] google_sso_creds = sso_manager.get_sso_credentials(static_vars.google_sso) data = { 'code': code, 'client_id': google_sso_creds[var_names.client_id], 'client_secret': google_sso_creds[var_names.client_secret], 'redirect_uri': sso_manager.taskcall_google_sso_redirect_uri, 'grant_type': 'authorization_code' } ag_status, ag_response = helpers.send_post_request(token_endpoint, body=data) if ag_status == 200: id_token = ag_response[sso_manager.var_id_token] redirect_url = helpers.get_redirect_page(request) token_name = helpers.get_itsm_token_name(redirect_url) token = helpers.create_public_itsm_request_token( redirect_url, life_days=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME + 1) resp = redirect(redirect_url) resp.set_cookie(token_name, token, max_age=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME, secure=True, httponly=True, samesite='None') return resp else: context[var_names.error_message] = lnm.err_unauthorized_access return render(request, pages.external_sso_login_page, context=context) else: err_message = lnm.err_system_error raise SystemError else: err_message = lnm.err_unauthorized_access raise PermissionError except Exception as e: logging.exception(str(e)) context[var_names.error_message] = lt.get_label(err_message, lang) return render(request, pages.external_sso_login_page, context=context) @csrf_exempt @require_http_methods(['POST']) def azure_sso_authentication(request): ''' Handles the process of authenticating a user through Azure AD SSO. The request for this view must originate from Microsoft Azure AD. :param request: Http request from Azure AD :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) context = itsm_context.get_itsm_login_context(lang) err_message = lnm.err_system_error try: # Ensure that the 'state' value is the same as that sent from originating request from TaskCall. # Also make sure that the ip address and subdomain values that should have been saved in the cache # prior to making the originating request are also there. cache_key = helpers.make_cache_key(static_vars.cache_prefix_azure_sso, request.POST.get('state')) az_cache_data = cache.get(cache_key) if az_cache_data is not None: az_cache_data = json.loads(az_cache_data) if (sso_manager.is_azure_csrf_exempt_origin(request.META[static_vars.http_referer_attribute]) or sso_manager.is_azure_csrf_exempt_origin(request.META[static_vars.http_origin_attribute])) and\ az_cache_data is not None and var_names.ip_address in az_cache_data and\ var_names.subdomain in az_cache_data: id_token = request.POST.get(sso_manager.var_id_token) redirect_url = helpers.get_redirect_page(request) token_name = helpers.get_itsm_token_name(redirect_url) token = helpers.create_public_itsm_request_token( redirect_url, life_days=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME + 1) resp = redirect(redirect_url) resp.set_cookie(token_name, token, max_age=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME, secure=True, httponly=True, samesite='None') cache.delete(cache_key) return resp else: err_message = lnm.err_unauthorized_access raise PermissionError except Exception as e: logging.exception(str(e)) context[var_names.error_message] = lt.get_label(err_message, lang) return render(request, pages.external_sso_login_page, context=context) @require_http_methods(['GET']) def okta_authentication(request): ''' Handles the process of authenticating a user through Okta. The request for this view must originate from Okta. :param request: Http request from Google :return: Http response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) context = itsm_context.get_itsm_login_context(lang) err_message = lt.get_label(lnm.err_system_error, lang) try: # Ensure that the 'state' value is the same as that sent from originating request from TaskCall. # Also make sure that the ip address and subdomain values that should have been saved in the session # prior to making the originating request are also there. if request.GET.get('state') == request.session.session_key and\ sso_manager.var_okta_token_endpoint in request.session and\ sso_manager.var_okta_introspection_endpoint in request.session and\ sso_manager.var_okta_client_id in request.session and\ sso_manager.var_okta_client_secret in request.session and\ var_names.ip_address in request.session and var_names.subdomain in request.session: code = request.GET.get('code') token_endpoint = request.session[sso_manager.var_okta_token_endpoint] auth_data = { 'code': code, 'client_id': request.session[sso_manager.var_okta_client_id], 'client_secret': request.session[sso_manager.var_okta_client_secret], 'redirect_uri': sso_manager.taskcall_okta_redirect_uri, 'grant_type': 'authorization_code' } auth_status, auth_response = helpers.send_post_request( token_endpoint, body=auth_data, content_type=static_vars.content_type_url_encoded) if auth_status == 200: id_token = auth_response[sso_manager.var_id_token] redirect_url = helpers.get_redirect_page(request) token_name = helpers.get_itsm_token_name(redirect_url) token = helpers.create_public_itsm_request_token( redirect_url, life_days=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME + 1) resp = redirect(redirect_url) resp.set_cookie(token_name, token, max_age=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME, secure=True, httponly=True, samesite='None') return resp else: if 'error' in request.GET and 'error_description' in request.GET: err_message = 'Okta Error: ' + request.GET.get('error_description') else: err_message = lt.get_label(lnm.err_system_error, lang) raise SystemError else: err_message = lt.get_label(lnm.err_unauthorized_access, lang) raise PermissionError except Exception as e: logging.exception(str(e)) context[var_names.error_message] = err_message return render(request, pages.external_sso_login_page, context=context) @csrf_exempt @require_http_methods(['POST']) def saml_sso_authentication(request): ''' Handles the process of authenticating a user through SAML 2.0. The request for this view must originate from the IdP. :param request: Http request from SAML IdP :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) context = itsm_context.get_itsm_login_context(lang) err_message = lt.get_label(lnm.err_system_error, lang) try: cache_key = static_vars.cache_prefix_saml_sso saml_cache_data = cache.get(cache_key) if saml_cache_data is not None: saml_cache_data = json.loads(saml_cache_data) # Make sure that the ip address, subdomain and saml settings that should have been saved in the cache # prior to making the originating request are there. if saml_cache_data is not None and var_names.ip_address in saml_cache_data and\ var_names.subdomain in saml_cache_data and sso_manager.var_saml_settings in saml_cache_data: if not request.session.session_key: request.session.save() request.session.set_expiry(2592000) saml_settings = saml_cache_data[sso_manager.var_saml_settings] req = { 'http_host': request.META['HTTP_HOST'], 'script_name': request.get_full_path(), 'server_port': request.META['SERVER_PORT'], 'get_data': request.GET.copy(), 'post_data': request.POST.copy() } auth = OneLogin_Saml2_Auth(req, saml_settings) auth.process_response() errors = auth.get_errors() if not errors: if auth.is_authenticated() and auth.get_nameid() is not None: email = auth.get_nameid() redirect_url = helpers.get_redirect_page(request) token_name = helpers.get_itsm_token_name(redirect_url) token = helpers.create_public_itsm_request_token( redirect_url, life_days=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME + 1) resp = redirect(redirect_url) resp.set_cookie(token_name, token, max_age=configs.DEFAULT_ITSM_PAGE_COOKIE_LIFETIME, secure=True, httponly=True, samesite='None') cache.delete(cache_key) return resp else: err_message = lnm.err_unauthorized_access raise PermissionError else: logging.error(auth.get_last_error_reason()) context[var_names.error_message] = str(errors) else: context[var_names.error_message] = lt.get_label(lnm.err_system_error, lang) return render(request, pages.external_sso_login_page, context=context) except Exception as e: logging.exception(str(e)) context[var_names.error_message] = err_message return render(request, pages.external_sso_login_page, context=context)