# By: Md. Fahim Bin Amin # This file contains views related to the SolarWinds ServiceDesk integration. from data_syncers import syncer_task_instances from dbqueries import db_integrations from dbqueries.integrations import db_solarwinds_servicedesk from exceptions.user_exceptions import InvalidRequest, UnauthorizedRequest from integrations import solarwinds_servicedesk from modules.router import Router from objects.events import AcknowledgeEvent, CustomActionEvent, ResolveEvent, UrgencyAmendmentEvent from objects.task_payload import TaskPayload from rest_framework.decorators import api_view from rest_framework.response import Response from translators import label_translator as _lt from utils import constants, errors, info, integration_type_names as intt, logging, permissions, times, tokenizer, var_names from utils.db_connection import CACHE_CLIENT, CONN_POOL from validations import request_validator import configuration import jwt @api_view(['POST']) def get_solarwinds_accounts(request, conn=None): """ Get SolarWinds ServiceDesk accounts configured in TaskCall (not SWSD accounts). """ if request.method == 'POST': lang = request_validator.get_user_language(request) try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, []) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_INTEGRATION_ITSM_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) return Response( db_solarwinds_servicedesk.get_solarwinds_servicedesk_accounts( conn, times.get_current_timestamp(), org_id ) ) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError): 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_solarwinds_account_details(request, conn=None): """ Get the domain and username for a SolarWinds ServiceDesk integration. """ if request.method == 'POST': lang = request_validator.get_user_language(request) expected_fields = [var_names.integration_key] 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) integ_key = request.data[var_names.integration_key] if not permissions.has_org_permission(org_perm, permissions.ORG_INTEGRATION_ITSM_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) det = db_solarwinds_servicedesk.get_solarwinds_servicedesk_account_details( conn, times.get_current_timestamp(), org_id, integration_key=integ_key ) if var_names.password in det: del det[var_names.password] return Response(det) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError): 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_solarwinds_sd_meta_data(request, conn=None): """ Get SWSD metadata required for configuration: - Accounts - Categories - Priorities - Statuses Supports BOTH flows: New integration → domain + password in request Edit integration → integration_key only (credentials pulled from DB) """ lang = request_validator.get_user_language(request) try: conn = CONN_POOL.get_db_conn() if conn is None else conn logging.info(f"SWSD Meta request data {request.data}") user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_INTEGRATION_ITSM_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) # ---------------------------------------------------- # 1) Try NEW integration flow (domain + token sent) # ---------------------------------------------------- domain = request.data.get(var_names.vendor_endpoint_name) api_token = request.data.get(var_names.password) # ---------------------------------------------------- # 2) If missing → fallback to EDIT flow via DB # ---------------------------------------------------- if not domain or not api_token: integ_key = request.data.get(var_names.integration_key) if not integ_key: logging.error("SWSD META: Missing integration_key and no credentials sent") return Response(_lt.get_label(errors.err_invalid_request, lang), status=400) logging.info(f"SWSD meta using integration key {integ_key} to fetch stored credentials") sw = db_solarwinds_servicedesk.get_solarwinds_servicedesk_account_details( conn, times.get_current_timestamp(), org_id, integration_key=integ_key ) if not sw: raise InvalidRequest(errors.err_invalid_request) domain = sw.get(var_names.vendor_endpoint_name) api_token = sw.get(var_names.password) if not domain or not api_token: raise InvalidRequest(errors.err_invalid_request) # ---------------------------------------------------- # Normalize domain # ---------------------------------------------------- if not domain.startswith(("http://", "https://")): domain = "https://" + domain.rstrip("/") # ---------------------------------------------------- # Fetch metadata from SolarWinds # ---------------------------------------------------- accounts = solarwinds_servicedesk.get_accounts(domain, api_token) categories = solarwinds_servicedesk.get_categories(domain, api_token) priorities = solarwinds_servicedesk.get_priorities(domain, api_token) statuses = solarwinds_servicedesk.get_statuses(domain, api_token) # ---------------------------------------------------- # Validation # ---------------------------------------------------- if not isinstance(accounts, list): logging.error("SolarWinds get_accounts failed: %s", accounts) return Response(_lt.get_label(errors.err_invalid_request, lang), status=400) logging.info(f"PRIORITIES RAW: {priorities}") logging.info(f"STATUSES RAW: {statuses}") fmt_accounts = accounts fmt_categories = categories if isinstance(categories, list) else [] fmt_priorities = priorities if isinstance(priorities, list) else [] fmt_statuses = statuses if isinstance(statuses, list) else [] return Response([fmt_accounts, fmt_categories, fmt_priorities, fmt_statuses]) except (InvalidRequest, ValueError) as e: logging.exception(str(e)) return Response(_lt.get_label(errors.err_invalid_request, lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError): 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 update_solarwinds_credentials(request, conn=None): """ Update SWSD admin credentials (domain + optional email + API key). """ if request.method == 'POST': lang = request_validator.get_user_language(request) required_fields = [ var_names.vendor_endpoint_name, var_names.password # API token is required ] try: conn = CONN_POOL.get_db_conn() if conn is None else conn request_validator.validate_fields(request, required_fields) user_id, org_id, user_perm, org_perm = tokenizer.authorize_request(request) if not permissions.has_org_permission(org_perm, permissions.ORG_INTEGRATION_ITSM_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) domain = request.data[var_names.vendor_endpoint_name] admin_user = request.data.get(var_names.username, '') # Optional api_key = request.data[var_names.password] # Validate API token uid = solarwinds_servicedesk.get_admin_uid(api_key, domain) if uid is None: return Response(_lt.get_label(errors.err_invalid_request, lang), status=403) updated = db_solarwinds_servicedesk.update_solarwinds_servicedesk_credentials( conn, times.get_current_timestamp(), org_id, domain, admin_user, api_key, uid ) return Response(updated) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError): 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 process_incoming_webhook(request, integration_key, conn=None, cache=None): """ Process SolarWinds ServiceDesk incoming webhook. """ if request.method == 'POST': lang = request_validator.get_user_language(request) try: conn = CONN_POOL.get_db_conn() if conn is None else conn cache = CACHE_CLIENT if cache is None else cache if "incident" not in request.data: logging.error("SolarWinds webhook missing 'incident'") return Response(_lt.get_label(errors.err_invalid_request, lang), status=400) inc = request.data["incident"] tkt_id = inc["id"] tkt_title = inc.get("name", "") tkt_desc = inc.get("description", "") tkt_state = inc.get("state", "") tkt_priority = inc.get("priority", "") ts = times.get_current_timestamp() org_id, org_perm, serv_id, integ_id, integ_type_id, integ_details = \ db_integrations.get_integration_details(conn, ts, integration_key, intt.solarwinds_servicedesk) # Translate to internal urgency urgency = constants.critical_urgency if var_names.urgency_level in integ_details: for k, v in integ_details[var_names.urgency_level].items(): if v == tkt_priority: urgency = int(k) break # Translate to internal status status = constants.open_state if var_names.status in integ_details: for k, v in integ_details[var_names.status].items(): if v == tkt_state: status = k break synced = db_integrations.get_vendor_synced_open_instance_ids( conn, ts, org_id, integ_id, integ_type_id, str(tkt_id) ) if len(synced) == 0: payload = TaskPayload( ts, org_id, ts.date(), tkt_title, configuration.standard_timezone, ts.time(), text_msg=tkt_desc, urgency_level=urgency, trigger_method=constants.integrations_api, trigger_info=request.data, integration_id=integ_id, service_id=serv_id ) inst_id = Router(conn, cache, payload).run() event = CustomActionEvent( inst_id, ts, constants.internal, integration_id=integ_id, integration_type_id=integ_type_id, vendor_id=str(tkt_id), vendor_url=None, is_synced=True, configuration_name=integ_details[var_names.configuration_name] ) syncer_task_instances.execute_custom_action(conn, cache, event, org_id=org_id, is_sys_action=True) else: for inst_id, task_id, in_sync in synced: inst = syncer_task_instances.get_single_instance(conn, cache, ts, inst_id) if urgency != inst.task.urgency_level(): event = UrgencyAmendmentEvent(inst_id, ts, constants.integrations_api, urgency) syncer_task_instances.amend_urgency(conn, cache, [event], org_id, is_sys_action=True) if status != inst.status: if status == constants.acknowledged_state: event = AcknowledgeEvent(inst_id, ts, constants.integrations_api, None) syncer_task_instances.acknowledge( conn, cache, event, org_id, is_sys_action=True, skip_syncing=[intt.solarwinds_servicedesk], org_perm=org_perm ) elif status == constants.resolved_state: event = ResolveEvent(inst_id, ts, constants.integrations_api) syncer_task_instances.resolve( conn, cache, event, org_id, is_sys_action=True, skip_syncing=[intt.solarwinds_servicedesk], org_perm=org_perm ) return Response(info.msg_internal_success) except InvalidRequest: return Response(_lt.get_label(errors.err_invalid_request, lang), status=400) except Exception as e: logging.error(request.data) 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 solarwinds_account_entry(request, conn=None): """ Entry point for Step 1: Frontend requests to see whether user has existing SolarWinds SD accounts. Return: [] if no accounts exist ["domain1", "domain2"] if accounts exist """ 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_INTEGRATION_ITSM_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) # Fetch SWSD accounts stored in TaskCall accounts = db_solarwinds_servicedesk.get_solarwinds_servicedesk_accounts( conn, times.get_current_timestamp(), org_id ) # The frontend expects **just a list**: # [] or ["acme.samanage.com", "corp.samanage.com"] domains = [acc[var_names.vendor_endpoint_name] for acc in accounts] return Response(domains) except InvalidRequest as e: logging.exception(str(e)) return Response(_lt.get_label(str(e), lang), status=400) except (UnauthorizedRequest, jwt.ExpiredSignatureError, jwt.InvalidSignatureError): 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)