# By: Riasat Ullah # This file contains views for serving sso specific pages. from constants import api_paths, label_names as lnm, pages, static_vars, url_paths, var_names from context_manager import info_context from device_detector import DeviceDetector from django.core.cache import cache 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 system_tests.test_data import test_data_organization, test_data_users from taskcallweb import settings from translators import label_translator as lt from utils import helpers, logging, sso_manager from utils.custom_redirect import MobileAppRedirect from validations import request_validator import configuration as configs import json @require_http_methods(['GET', '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 == 'GET': lang = request_validator.get_user_language(request) context = info_context.get_login_context(lang) context[var_names.method] = static_vars.sso context[var_names.taskcall_domain] = tc_domain for_mobile = False device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if helpers.is_mobile_app_login(request) and device.is_mobile(): for_mobile = True try: if request_validator.user_in_session(request): return redirect(pages.incidents_url) else: return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) except Exception as e: logging.exception(str(e)) return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) elif request.method == 'POST': lang = request_validator.get_user_language(request) context = info_context.get_login_context(lang) context[var_names.method] = static_vars.sso context[var_names.taskcall_domain] = tc_domain for_mobile = False try: sub_domain = request.POST[var_names.subdomain] body = {var_names.subdomain: sub_domain} device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if helpers.is_mobile_app_login(request) and device.is_mobile(): for_mobile = True if not request.session.session_key: request.session.save() request.session.set_expiry(2592000) # Store the ip address now. This will be sent to the back end to retrieve the access token # after the sso authentication is complete. request.session[var_names.ip_address] = request.META.get(static_vars.ip_address_attribute) request.session[var_names.subdomain] = sub_domain if settings.TEST_MODE: status, output = 200, test_data_organization.sso_settings else: status, output = helpers.post_api_request(api_paths.org_sso_settings, body) 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: sub_domain, var_names.nonce: minted_nonce} if var_names.mobile_app_redirect_url in request.session and\ var_names.mobile_app_push_token in request.session: az_cache_data[var_names.mobile_app_redirect_url] =\ request.session[var_names.mobile_app_redirect_url] az_cache_data[var_names.mobile_app_push_token] =\ request.session[var_names.mobile_app_push_token] 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.login_mobile_app_page if for_mobile else pages.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 = url_paths.login_sso_saml + '?subdomain=' + sub_domain 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: sub_domain, 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.login_mobile_app_page if for_mobile else pages.login_page, context=context) else: context[var_names.error_message] = output.replace("'", "") return render(request, pages.login_mobile_app_page if for_mobile else pages.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.login_mobile_app_page if for_mobile else pages.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 = info_context.get_login_context(lang) err_message = lnm.err_system_error for_mobile = False try: device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if helpers.is_mobile_app_login(request) and device.is_mobile(): for_mobile = True # 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] body = { var_names.integration_type: static_vars.google_sso, var_names.id_token: id_token, var_names.subdomain: request.session[var_names.subdomain], var_names.ip_address: request.session[var_names.ip_address] } if for_mobile: body[var_names.access_method] = static_vars.app body[var_names.push_token] = helpers.get_mobile_app_push_token(request) else: body[var_names.access_method] = static_vars.web body[var_names.session_id] = request.session.session_key if settings.TEST_MODE: status, output = 200, test_data_users.user_authentication_details[ test_data_users.users_list[0][1]] else: status, output = helpers.post_api_request(api_paths.user_login, body) if status == 200: del request.session[var_names.ip_address] del request.session[var_names.subdomain] if for_mobile: redirect_url = helpers.get_mobile_app_full_redirect_url(request, output) helpers.clear_mobile_app_login_variables(request) return MobileAppRedirect(redirect_url) else: helpers.set_session_authorization_tokens(request, output) redirect_url = helpers.get_redirect_page(request) if helpers.has_only_stakeholder_rights(output, redirect_url): return redirect(pages.status_dashboard_url) return redirect(redirect_url) elif status == 307: if for_mobile: del request.session[var_names.ip_address] del request.session[var_names.subdomain] err_message = lnm.err_sso_mobile_app_auto_provision_not_allowed raise PermissionError else: # save the variables that will be needed to "register and launch" in the session request.session[var_names.sso_email] = output[var_names.email] request.session[var_names.registration_token] = output[var_names.token] request.session[var_names.sso_login_post_body] = body redirect_url = pages.auto_provision_member_url + '?state=' + request.session.session_key return redirect(redirect_url) else: context[var_names.error_message] = output.replace("'", "") return render(request, pages.login_mobile_app_page if for_mobile else pages.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.login_mobile_app_page if for_mobile else pages.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 = info_context.get_login_context(lang) err_message = lnm.err_system_error for_mobile = False 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) device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if device.is_mobile() and var_names.mobile_app_redirect_url in az_cache_data and\ var_names.mobile_app_push_token in az_cache_data: for_mobile = True 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: if not request.session.session_key: request.session.save() request.session.set_expiry(2592000) id_token = request.POST.get(sso_manager.var_id_token) body = { var_names.integration_type: static_vars.azure_ad_sso, var_names.id_token: id_token, var_names.subdomain: az_cache_data[var_names.subdomain], var_names.ip_address: az_cache_data[var_names.ip_address] } if for_mobile: body[var_names.access_method] = static_vars.app body[var_names.push_token] = az_cache_data[var_names.mobile_app_push_token] else: body[var_names.access_method] = static_vars.web body[var_names.session_id] = request.session.session_key if settings.TEST_MODE: status, output = 200, test_data_users.user_authentication_details[ test_data_users.users_list[0][1]] else: status, output = helpers.post_api_request(api_paths.user_login, body) if status == 200: if for_mobile: redirect_url = helpers.get_mobile_app_full_redirect_url( None, output, redirect_url=az_cache_data[var_names.mobile_app_redirect_url]) cache.delete(cache_key) helpers.clear_mobile_app_login_variables(request) return MobileAppRedirect(redirect_url) else: helpers.set_session_authorization_tokens(request, output) redirect_url = helpers.get_redirect_page(request) if helpers.has_only_stakeholder_rights(output, redirect_url): redirect_url = pages.status_dashboard_url resp = redirect(redirect_url) cache.delete(cache_key) return resp elif status == 307: if for_mobile: err_message = lnm.err_sso_mobile_app_auto_provision_not_allowed raise PermissionError else: # save the variables that will be needed to "register and launch" in the session request.session[var_names.ip_address] = request.META.get(static_vars.ip_address_attribute) request.session[var_names.subdomain] = az_cache_data[var_names.subdomain] request.session[var_names.sso_email] = output[var_names.email] request.session[var_names.registration_token] = output[var_names.token] request.session[var_names.sso_login_post_body] = body if var_names.state in request.session: del request.session[var_names.state] redirect_url = pages.auto_provision_member_url + '?state=' + request.session.session_key resp = redirect(redirect_url) cache.delete(cache_key) return resp else: context[var_names.error_message] = output.replace("'", "") return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) 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.login_mobile_app_page if for_mobile else pages.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 = info_context.get_login_context(lang) err_message = lt.get_label(lnm.err_system_error, lang) for_mobile = False try: device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if helpers.is_mobile_app_login(request) and device.is_mobile(): for_mobile = True # 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] body = { var_names.integration_type: static_vars.okta, var_names.id_token: id_token, var_names.vendor_endpoint: request.session[sso_manager.var_okta_introspection_endpoint], var_names.subdomain: request.session[var_names.subdomain], var_names.ip_address: request.session[var_names.ip_address] } if for_mobile: body[var_names.access_method] = static_vars.app body[var_names.push_token] = helpers.get_mobile_app_push_token(request) else: body[var_names.access_method] = static_vars.web body[var_names.session_id] = request.session.session_key if settings.TEST_MODE: status, output = 200, test_data_users.user_authentication_details[ test_data_users.users_list[0][1]] else: status, output = helpers.post_api_request(api_paths.user_login, body) if status == 200: del request.session[sso_manager.var_okta_token_endpoint] del request.session[sso_manager.var_okta_introspection_endpoint] del request.session[sso_manager.var_okta_client_id] del request.session[sso_manager.var_okta_client_secret] del request.session[var_names.ip_address] del request.session[var_names.subdomain] if for_mobile: redirect_url = helpers.get_mobile_app_full_redirect_url(request, output) helpers.clear_mobile_app_login_variables(request) return MobileAppRedirect(redirect_url) else: helpers.set_session_authorization_tokens(request, output) redirect_url = helpers.get_redirect_page(request) if helpers.has_only_stakeholder_rights(output, redirect_url): return redirect(pages.status_dashboard_url) return redirect(redirect_url) elif status == 307: # Delete the Okta specific variables del request.session[sso_manager.var_okta_token_endpoint] del request.session[sso_manager.var_okta_introspection_endpoint] if for_mobile: del request.session[var_names.ip_address] del request.session[var_names.subdomain] err_message = lt.get_label(lnm.err_sso_mobile_app_auto_provision_not_allowed, lang) raise PermissionError else: # save the variables that will be needed in "register and launch" function in the session request.session[var_names.sso_email] = output[var_names.email] request.session[var_names.registration_token] = output[var_names.token] request.session[var_names.sso_login_post_body] = body redirect_url = pages.auto_provision_member_url + '?state=' + request.session.session_key return redirect(redirect_url) else: context[var_names.error_message] = output.replace("'", "") return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) 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.login_mobile_app_page if for_mobile else pages.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 = info_context.get_login_context(lang) err_message = lt.get_label(lnm.err_system_error, lang) for_mobile = False 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) device = DeviceDetector(request.META[static_vars.http_user_agent_attribute]).parse() if device.is_mobile() and var_names.mobile_app_redirect_url in saml_cache_data and\ var_names.mobile_app_push_token in saml_cache_data: for_mobile = True # 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() body = { var_names.integration_type: static_vars.saml, var_names.email: email, var_names.subdomain: saml_cache_data[var_names.subdomain], var_names.ip_address: saml_cache_data[var_names.ip_address] } if for_mobile: body[var_names.access_method] = static_vars.app body[var_names.push_token] = saml_cache_data[var_names.mobile_app_push_token] else: body[var_names.access_method] = static_vars.web body[var_names.session_id] = request.session.session_key if settings.TEST_MODE: status, output = 200, test_data_users.user_authentication_details[ test_data_users.users_list[0][1]] else: status, output = helpers.post_api_request(api_paths.user_login, body) if status == 200: if for_mobile: redirect_url = helpers.get_mobile_app_full_redirect_url( None, output, redirect_url=saml_cache_data[var_names.mobile_app_redirect_url]) cache.delete(cache_key) helpers.clear_mobile_app_login_variables(request) return MobileAppRedirect(redirect_url) else: helpers.set_session_authorization_tokens(request, output) redirect_url = helpers.get_redirect_page(request) if helpers.has_only_stakeholder_rights(output, redirect_url): redirect_url = pages.status_dashboard_url resp = redirect(redirect_url) cache.delete(cache_key) return resp elif status == 307: if for_mobile: err_message = lnm.err_sso_mobile_app_auto_provision_not_allowed raise PermissionError else: # save the variables that will be needed to "register and launch" in the session request.session[var_names.ip_address] =\ request.META.get(static_vars.ip_address_attribute) request.session[var_names.subdomain] = saml_cache_data[var_names.subdomain] request.session[var_names.sso_email] = output[var_names.email] request.session[var_names.registration_token] = output[var_names.token] request.session[var_names.sso_login_post_body] = body if var_names.state in request.session: del request.session[var_names.state] redirect_url = pages.auto_provision_member_url + '?state=' + request.session.session_key resp = redirect(redirect_url) cache.delete(cache_key) return resp else: context[var_names.error_message] = output.replace("'", "") return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context) 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.login_mobile_app_page if for_mobile else pages.login_page, context=context) except Exception as e: logging.exception(str(e)) context[var_names.error_message] = err_message return render(request, pages.login_mobile_app_page if for_mobile else pages.login_page, context=context)