from enum import Enum import json import os import re import time import magic from exchangelib import DELEGATE, Credentials, Account, HTMLBody, Message # from exchangelib.protocol import BASE64 import base64 as BASE64 import grpc import openpyxl import requests import cloudServiceDAO_pb2, cloudServiceDAO_pb2_grpc from bs4 import BeautifulSoup from datetime import datetime """" Relación de respuestas a correos acorde a estatus -------------------------------------------------------------------------------- SENDER | RECEIVER -------------------------------------------------------------------------------- CONTRATACIÓN | x SOLICITADO | ACTIVADO | ERROR DE ACTIVACIÓN -------------------------------------------------------------------------------- x CANCELACIÓN | CANCELADO | ERROR DE CANCELACIÓN -------------------------------------------------------------------------------- INSTALACIÓN DE SENSOR | INSTALADO | ERROR DE INSTALACIÓN -------------------------------------------------------------------------------- DESINSTALACIÓN DE SENSOR | DESINSTALADO | ERROR DE DESINSTALACIÓN -------------------------------------------------------------------------------- """ class TaskType(Enum): SERVICE = 1 LICENSE = 2 # **se requieren añadir a Dockerfile # cuenta de correo destinatario RECEIVER_EMAIL = os.environ["ORDERTAKER_RECEIVER_EMAIL"] # contraseña de destinatario RECEIVER_PASSWORD = os.environ["ORDERTAKER_RECEIVER_PASS"] # cuenta de correo remitente esperado (para filtrado) SENDER_EMAIL = os.environ["ORDERTAKER_SENDER_EMAIL"] # host de comunicación para con el microservicio de DAO dao_host = os.getenv("DAO_HOST", "dao.grpc") credentials = Credentials(RECEIVER_EMAIL, RECEIVER_PASSWORD) # Conexión a buzón de correo de servicio SGM account = Account( RECEIVER_EMAIL, credentials=credentials, autodiscover=True, access_type=DELEGATE ) inbox = account.inbox with open("ca.pem", "rb") as fp: ca_cert = fp.read() creds = grpc.ssl_channel_credentials(ca_cert) # Canal de comunicación hacia microservicio de DAO dao_stub = cloudServiceDAO_pb2_grpc.servicioDAOServiceStub( grpc.secure_channel(f"{dao_host}:8445", credentials=creds) ) # Objeto para identificación de mime-types mime = magic.Magic() def main(): while True: # Búsqueda de correos no leídos y por remitente unread_mails = account.inbox.filter(is_read=False, sender=SENDER_EMAIL) print( f"Encontrados {unread_mails.count()} correos filtrados por remitente y no leídos" ) for mail in unread_mails: # subject = mail.subject # Extracción de adjuntos por archivo, tipo de archivo XLSX y nombre attachments = [ a for a in mail.attachments if hasattr(a, 'content') # Verifica si el adjunto tiene un atributo 'content' and hasattr(a, 'name') # Verifica si el adjunto tiene un atributo 'name' and "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" in mime.from_buffer(getattr(a, 'content', '')) and ("SERVICIO" in getattr(a, 'name', '') or "LICENCIA" in getattr(a, 'name', '')) ] if not attachments: set_error( "No se encontró archivo adjunto esperado, ésto puede deberse a varias razones, Revise manualmente y vuelva a enviar la solicitud o consulte con mesa de servicio si:\n1.- No se adjuntó ningún archivo\2.- El formato del archivo no es de Excel (xlsx)\n3.- El nombre del archivo es distinto a 'CONTRATACIÓN'", mail=mail, ) return elif len(attachments) > 2: set_error( "Se encontraron demasiados archivos adjuntos con el formato de nombre esperado, por ello no se puede determinar cuáles son los archivo correctos. Verifique manualmente cada archivo o solicite el envío de un sólo archivo por tarea (servicio y/o licencia) a mesa de servicio", mail=mail, ) return for attachment in attachments: # Guardamos el único ajunto esperado with open(attachment[0].name, "wb") as f: f.write(attachment[0].content) workbook = openpyxl.load_workbook(attachment.name) sheet = workbook.active if "SERVICIO" in attachment.name: process_service(sheet, mail) elif "LICENCIA" in attachment.name: process_licenses(sheet, mail) # Marcar el correo como leído para evitar procesarlo nuevamente mail.is_read = True mail.save() # Espera 10 segundos antes de verificar nuevamente time.sleep(60) def process_service(sheet, mail): request_queue_id = ( sheet.cell(row=9, column=2).value if sheet.cell(row=9, column=1).value == "SOLICITUD" else None ) if not request_queue_id: set_error( err="No fue posible localizar información que ayude a identificar de forma automatizada a cual cliente se realizó la contratación del servicio en la respuesta a la solicitud de contratación de servicio o el formato encontrado es distinto al esperado. Revise manualmente el formato adjunto y relice las actualizaciones pertinentes en la base de datos de SGM o vuelva a solicitar la contratación:\n1.- Identificación de elemento en tabla 'request_queue', actualización de estatus de servicio\n2.- Asignación de valores API_HOST y API_KEY (API_SECRET_KEY/API_ID) a registros de licencia relacionados en tabla 'licenses'", task_type=TaskType.SERVICE, mail=mail, ) return error = ( sheet.cell(row=14, column=2).value if sheet.cell(row=14, column=1).value == "ERROR" else None ) if error: set_error(err=error, id=request_queue_id) return # Se espera que se marque correo como leído api_host = ( sheet.cell(row=11, column=2).value if sheet.cell(row=11, column=1).value == "*API_HOST" else None ) if not api_host: set_error( err=f"No fue posible localizar el valor 'API_HOST' requerido dentro de respuesta a la solicitud de contratación de servicio o cuenta con un fomato distinto al esperado. Revise manualmente si existe la información y realice las acciones pertinentes:\n1.- Identificación de registros de licencia en tabla 'licenses' a partir de registro 'request_queue_id'={request_queue_id} en tabla request_queue_id\n2.- Actualización de registros de licencias relacionadas en tabla 'licenses' con el valor de 'API_HOST'", task_type=TaskType.SERVICE, id=request_queue_id, ) return api_id = ( sheet.cell(row=12, column=2).value if sheet.cell(row=12, column=1).value == "*API_ID" else None ) api_secret_key = ( sheet.cell(row=13, column=2).value if sheet.cell(row=13, column=1).value == "*API_SECRET_KEY" else None ) if not api_id or not api_secret_key: set_error( err=f"No fue posible localizar los valores necesarios requeridos dentro de respuesta a la solicitud de contratación de servicio o cuenta con un fomato distinto al esperado. Revise manualmente si existe la información y realice las acciones pertinentes:\n1.- Identificación de registros de licencia en tabla 'licenses' a partir de registro 'request_queue_id'={request_queue_id} en tabla request_queue_id\n2.- Validar existencia de 'API_SECRET_KEY' y 'API_ID', concatenar en formato 'API_SECRET_KEY'/'API_ID' y añadir a los registros de licencias relacionados a request_queue_id en la tabla 'licenses'", task_type=TaskType.SERVICE, id=request_queue_id, ) return req = cloudServiceDAO_pb2.GetRequestQueueRequest( request_queue_id=int(request_queue_id) ) try: res = dao_stub.getRequestQueueService(req) except Exception as e: set_error( err=f"{e}: No fue posible recuperar contracted_service relacionado. Verifique que no existen conflictos de relación entre la tabla 'request_queue' y 'contracted_service'", task_type=TaskType.SERVICE, id=request_queue_id, ) return request_queue = res.requestqueue[0] req = cloudServiceDAO_pb2.GetLicensesRequest( contracted_services_id=request_queue.contracted_services_id ) try: res = dao_stub.GetLicenses(req) except Exception as e: set_error( err=f"{e}: No fue posible recuperar licencias relacionadas a 'contracted_services_id'={request_queue.contracted_services_id}, es posible que existiera un error al insertar registros de licencias al contratar el servicio. Reporte éste error", task_type=TaskType.SERVICE, id=request_queue_id, ) return for l in res.licenses: specs = l.specs specs = specs.replace("'", '"') lic = json.loads(specs) lic["api_host"] = api_host lic["token"] = f"{api_secret_key}/{api_id}" req = cloudServiceDAO_pb2.SetUpdateLicensesRequest( idlicenses=l.idlicenses, specs=str(lic) ) try: res = dao_stub.setUpdateLicenses(req) except Exception as e: set_error( err=f"{e}: No fue posible insertar información de contratación en registros de licencias. Reporte éste error", task_type=TaskType.SERVICE, id=request_queue_id, ) return req = cloudServiceDAO_pb2.SetUpdateRequestQueueRequest( request_queue_id=request_queue_id, supply_datetime=(datetime.datetime.now()).strftime("%Y-%m-%d %H:%M:%S"), supply_status_id=2, # SUCCESS service_status_id=4, # ACTIVE ) try: res = dao_stub.setUpdateRequestQueue(req) except Exception as e: set_error( err=f"No fue posible actualizar estatus de aprovisionamiento y de servicio. Reporte éste error", task_type=TaskType.SERVICE, id=request_queue_id, ) return def process_licenses(sheet, mail): licenses = {} movement = ( sheet.cell(row=3, column=2).value if sheet.cell(row=3, column=2).value == "MOVIMIENTO" else None ) if not movement: set_error( err="No se encontró el tipo de movimiento en el formato esperado. Reporte éste error", task_type=TaskType.LICENSE, mail=mail, ) return for index, row in enumerate(sheet.iter_rows(values_only=True), start=1): if movement == "DESINSTALACIÓN" and row[0] == "ID LICENCIA": licenses[sheet.cell(row=index, column=2).value] = { "device_id": sheet.cell(row=index + 1, column=2).value, "desinstalado": sheet.cell(row=index + 2, column=2).value, "error": sheet.cell(row=index + 3, column=2).value, } elif movement == "INSTALACIÓN" and row[0] == "ID LICENCIA": licenses[sheet.cell(row=index, column=2).value] = { "device_id": sheet.cell(row=index + 1, column=2).value, "error": sheet.cell(row=index + 2, column=2).value, } if not licenses: set_error( err="No se encontraron identificadores de licencias en el archivo adjunto bajo el formato esperado. Revise el formato manualmente extrayendo la información de cada sensor instalado o solicite a mesa de servicio el envío del archivo bajo el formato correcto", task_type=TaskType.LICENSE, ) return for id, value in licenses.items(): if value["error"]: set_error(err=value["error"], task_type=TaskType.LICENSE, id=id) continue if not value["device_id"]: set_error( err=f"No se encontró valor de campo 'device_id' para licencia con 'id'={id} y tampoco se presenta un valor de error dentro del formato. Notifique a mesa de servicio para una aclaración o revise si el formato es distinto al esperado", task_type=TaskType.LICENSE, id=id, ) if movement == "DESINSTALACIÓN" and value["desinstalado"] == "NO": set_error( err="Se solicitó la desinstalación de un sensor, sin embargo no se encontró ningún error ni confirmación positiva sobre dicha solicitud. Consulte con mesa de servicio si ocurrió algún imprevisto o la razón de dicha respuesta", task_type=TaskType.LICENSE, id=id, ) req = cloudServiceDAO_pb2.GetLicensesRequest(idlicenses=id) try: res = dao_stub.GetLicenses(req) except Exception as e: set_error( err=f"No fue posible recuperar registro de licencia relacionado. Reporte éste error", task_type=TaskType.LICENSE, id=id, ) return # Recuperando información de licencia desde BD license = res.licenses[0] specs = license.specs specs = specs.replace("'", '"') specs = json.loads(specs) if movement == "INSTALACIÓN": req = cloudServiceDAO_pb2.GetRequestQueueRequest( contracted_services_id=license.contracted_services_id ) res = dao_stub.getRequestQueueService(req) expire_datetime = res.requestqueue[0].supply_datetime expire_datetime = datetime.datetime.strptime( expire_datetime, "%d/%m/%Y %H:%M:%S" ) expire_datetime = (expire_datetime + timedelta(days=365)).strftime( "%Y-%m-%d %H:%M:%S" ) api_host = specs["api_host"] api_key = specs["token"] specs["expire_date"] = expire_datetime try: # Se requiere org_key de consola de administración de CBC r = requests.get( f"{api_host}/appservices/v5/orgs", timeout=120, verify=False, headers={ "X-AUTH-TOKEN": api_key, }, ) r.raise_for_status() except requests.RequestException as e: set_error( err=f"No fue posible recuperar información de organización desde consola de adminsitración de carbon black cloud, requerida para extraer información de sensores. Reporte éste error", task_type=TaskType.LICENSE, id=id, ) return org_key = json.loads(r.content) org_key = org_key["organization"]["orgKey"] try: r = requests.get( f"{api_host}/appservices/v6/orgs/{org_key}/devices/{value['device_id']}", timeout=120, verify=False, headers={ "X-AUTH-TOKEN": api_key, }, ) r.raise_for_status() except requests.RequestException as e: set_error( err=f"No fue posible recuperar información de sensor relacionado a 'device_id'={value['device_id']}. Reporte éste error", task_type=TaskType.LICENSE, id=id, ) sensor_info = json.loads(r.content) for l in specs["data"]: if not l["version"] == (len(specs["data"]) - 1): continue l["sensor_loginUserName"] = sensor_info["name"] l["sensor_loginUserName"] = sensor_info["login_user_name"] l["sensor_registeredTime"] = sensor_info["registered_time"] l["sensor_deviceType"] = sensor_info["sensor_kit_type"] l["sensor_osVersion"] = sensor_info["os_version"] l["sensor_sensorVersion"] = sensor_info["sensor_version"] l["sensor_macAddress"] = sensor_info["mac_address"] l["sensor_deviceId"] = value["device_id"] else: specs["data"].append( { "version": len(specs["data"]), "vm_name": "", "vm_ownerName": "", "vm_vdcName": "", "vm_dateCreated": "", "vm_href": "", "sensor_name": "", "sensor_loginUserName": "", "sensor_registeredTime": "", "sensor_deviceType": "", "sensor_osVersion": "", "sensor_sensorVersion": "", "sensor_macAddress": "", "sensor_deviceId": "", } ) req = cloudServiceDAO_pb2.SetUpdateLicensesRequest( idlicenses=id, specs=str(specs), ) try: res = dao_stub.setUpdateLicenses(req) except Exception as e: set_error( err=f"No fue posible actualizar la información de sensores en nuestro registro. Reporte éste error", task_type=TaskType.LICENSE, id=id, ) return def set_error(err: str, task_type: TaskType = None, mail=None, id: int = 0): if not id: # Reenviar a devsecops, cuando se recupera una respuesta imcompleta o incorrecta de contratación de servicio email = Message( account=account, folder=mail.folder, subject="Fwd: Error Carbon Black Cloud", body=err, to_recipients=["gustavo@vctra.xyz"], ) for attachment in mail.attachments: if hasattr(attachment, 'is_file'): email.attach( name=attachment.name, content=BASE64.encode( attachment.content ), # Codifica el contenido en base64 content_type=attachment.content_type, ) email.send_and_save() elif task_type == TaskType.SERVICE: # Se actualiza estatus (SUPPLY_ERROR) y log de request_queue_id req = cloudServiceDAO_pb2.SetUpdateRequestQueueRequest( request_queue_id=id, error_log=err, supply_status_id=3 ) try: dao_stub.setUpdateRequestQueue(req) except Exception as e: # No queremos que se interrumpa todo el proceso print(e) else: req = cloudServiceDAO_pb2.GetLicensesRequest(idlicenses=id) try: res = dao_stub.GetLicenses(req) except Exception as e: # No queremos que se interrumpa todo el proceso print(e) # Se inserta campo 'error'dentro de json de licencia specs = res.licenses[0].specs specs["error"] = err req = cloudServiceDAO_pb2.SetUpdateLicensesRequest(idlicenses=id, specs=specs) try: dao_stub.setUpdateRequestQueue(req) except Exception as e: # No queremos que se interrumpa todo el proceso print(e) main()