# By: Riasat Ullah # This file contains functions related to external SSO views. from dbqueries.itsm import db_external_sso from exceptions.user_exceptions import InvalidRequest, UnauthorizedRequest from rest_framework.decorators import api_view from rest_framework.response import Response from translators import label_translator as _lt from utils import errors, info, logging, permissions, times, tokenizer, var_names from utils.db_connection import CONN_POOL from validations import request_validator import jwt @api_view(['POST']) def create_external_sso(request, conn=None): ''' Creates a new external SSO setting for an organization. :param request: Http request :param conn: db connection :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.sso_name, var_names.integration_type] optional_fields = [var_names.description, var_names.saml_certificate, var_names.saml_key, var_names.login_url, var_names.logout_url, var_names.metadata_url, var_names.entity_id, var_names.vendor_id, var_names.vendor_subdomain, var_names.additional_info] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, expected_fields, optional_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_EXTERNAL_SSO_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.is_user_admin(user_perm): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) current_time = times.get_current_timestamp() db_external_sso.create_external_sso( conn, current_time, org_id, request.data[var_names.sso_name], request.data[var_names.description], request.data[var_names.integration_type], saml_certificate=request.data[var_names.saml_certificate] if var_names.saml_certificate in request.data else None, saml_key=request.data[var_names.saml_key] if var_names.saml_key in request.data else None, login_url=request.data[var_names.login_url] if var_names.login_url in request.data else None, logout_url=request.data[var_names.logout_url] if var_names.logout_url in request.data else None, metadata_url=request.data[var_names.metadata_url] if var_names.metadata_url in request.data else None, entity_id=request.data[var_names.entity_id] if var_names.entity_id in request.data else None, vendor_id=request.data[var_names.vendor_id] if var_names.vendor_id in request.data else None, vendor_subdomain=request.data[var_names.vendor_subdomain] if var_names.vendor_subdomain in request.data else None, additional_info=request.data[var_names.additional_info] if var_names.additional_info in request.data else None ) return Response(_lt.get_label(info.msg_external_sso_created, lang)) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_invalid_request, lang), status=401) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn) @api_view(['POST']) def edit_external_sso(request, conn=None): ''' Edit an existing external SSO setting. :param request: Http request :param conn: db connection :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.sso_ref_id, var_names.sso_name, var_names.integration_type] optional_fields = [var_names.description, var_names.saml_certificate, var_names.saml_key, var_names.login_url, var_names.logout_url, var_names.metadata_url, var_names.entity_id, var_names.vendor_id, var_names.vendor_subdomain, var_names.additional_info] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, expected_fields, optional_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_EXTERNAL_SSO_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.is_user_admin(user_perm): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) current_time = times.get_current_timestamp() db_external_sso.edit_external_sso( conn, current_time, org_id, request.data[var_names.sso_ref_id], request.data[var_names.sso_name], request.data[var_names.description], request.data[var_names.integration_type], saml_certificate=request.data[var_names.saml_certificate] if var_names.saml_certificate in request.data else None, saml_key=request.data[var_names.saml_key] if var_names.saml_key in request.data else None, login_url=request.data[var_names.login_url] if var_names.login_url in request.data else None, logout_url=request.data[var_names.logout_url] if var_names.logout_url in request.data else None, metadata_url=request.data[var_names.metadata_url] if var_names.metadata_url in request.data else None, entity_id=request.data[var_names.entity_id] if var_names.entity_id in request.data else None, vendor_id=request.data[var_names.vendor_id] if var_names.vendor_id in request.data else None, vendor_subdomain=request.data[var_names.vendor_subdomain] if var_names.vendor_subdomain in request.data else None, additional_info=request.data[var_names.additional_info] if var_names.additional_info in request.data else None ) return Response(_lt.get_label(info.msg_external_sso_edited, lang)) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_invalid_request, lang), status=401) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn) @api_view(['POST']) def delete_external_sso(request, conn=None): ''' Delete an external SSO setting. :param request: Http request :param conn: db connection :return: Http response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.sso_ref_id] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, expected_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_EXTERNAL_SSO_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.is_user_admin(user_perm): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) db_external_sso.delete_external_sso(conn, times.get_current_timestamp(), org_id, request.data[var_names.sso_ref_id]) return Response(_lt.get_label(info.msg_external_sso_deleted, lang)) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_invalid_request, lang), status=401) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn) @api_view(['POST']) def list_external_sso(request, conn=None): ''' Gets the list of external SSO configurations of an organization. :param request: Http request :param conn: db connection :return: JSON response -> list ''' if request.method == 'POST': lang = request_validator.get_user_language(request) try: conn = CONN_POOL.get_db_conn() if conn is None else conn user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_EXTERNAL_SSO_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.is_user_admin(user_perm): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) data_list = db_external_sso.list_external_sso(conn, times.get_current_timestamp(), org_id) return Response(data_list) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn) @api_view(['POST']) def get_external_sso_details(request, conn=None): ''' Gets the details of an external SSO configuration. :param request: Http request :param conn: db connection :return: JSON response -> list ''' if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.sso_ref_id] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, expected_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_EXTERNAL_SSO_PERMISSION): return Response(_lt.get_label(errors.err_subscription_rights, lang), status=403) if not permissions.is_user_admin(user_perm): return Response(_lt.get_label(errors.err_user_rights, lang), status=403) details = db_external_sso.get_external_sso_details(conn, times.get_current_timestamp(), org_id, request.data[var_names.sso_ref_id]) if details is None: return Response(_lt.get_label(errors.err_unknown_resource, lang), status=401) else: return Response(details) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_authorization, lang), status=401) except Exception as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_system_error, lang), status=500) finally: CONN_POOL.put_db_conn(conn)