import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 """ IMPORTS """ # Std imports import re # 3-rd party imports from datetime import datetime import urllib3 # Local imports # N/A """ GLOBALS/PARAMS Attributes: INTEGRATION_NAME: Name of the integration as shown in the integration UI, for example: Microsoft Graph User. INTEGRATION_COMMAND_NAME: Command names should be written in all lower-case letters, and each word separated with a hyphen, for example: msgraph-user. INTEGRATION_CONTEXT_NAME: Context output names should be written in camel case, for example: MSGraphUser. """ INTEGRATION_NAME = "Keyfactor" INTEGRATION_COMMAND_NAME = "keyfactor" INTEGRATION_CONTEXT_NAME = "Keyfactor" # Disable insecure warnings urllib3.disable_warnings() class Client(BaseClient): def test_module(self) -> dict: """ Performs basic GET request to check if the API is reachable and authentication is successful. Returns: Response dictionary """ return self.get_enrollment_csr_context_template() def get_enrollment_csr_context_template(self) -> dict: """ Get enrollment CSR context my > Retrieve a list of existing certificate templates Args: N/A Returns: Json response as dictionary """ headers = {"x-keyfactor-api-version": "1", "x-keyfactor-requested-with": "APIClient", "Accept": "application/json"} return self._http_request(method="GET", url_suffix="Enrollment/CSR/Context/My", headers=headers) def post_enrollment_csr( self, csr_base64: str, cert_authority: str, include_chain: str, time_stamp: str, template: str, sans_ip4: str, keyAlgorithm: str, metadata: dict, ) -> dict: """ Post Enrollment CSR Send the certifcate CSR and return the certificate Args: csr_base64: The base-64 encoded CSR that will be passed in for enrollment. cert_authority: A string that sets the name of the certificate authority include_chain: A Boolean that sets whether to include the certificate chain in theresponse (true) or not (false). The default is false. time_stamp: The current date and time. The date and time should be given using the ISO 8601 UTC time format YYYY-MM-DDTHH:mm:ss.000Z (e.g.2021-05-19T16:23:01Z). template: A string that sets the name of the certificate template thatshould be used to issue the certificate. sans_ip4: Subject alternative name (SAN) ip addresses keyAlgorithm: A string indicating the cryptographic algorithm to use to generate the key metadata: An array of key/value pairs that set the values for the metadata fields Returns: Json response as dictionary """ if sans_ip4 == "": body = { "CSR": csr_base64, "CertificateAuthority": cert_authority, "includeChain": include_chain, "Timestamp": time_stamp, "Template": template, "Metadata": metadata, } else: ip4 = {"ip4": [sans_ip4]} body = { "CSR": csr_base64, "CertificateAuthority": cert_authority, "includeChain": include_chain, "Timestamp": time_stamp, "Template": template, "SANs": ip4, "Metadata": metadata, } headers = { "x-keyfactor-api-version": "1", "x-keyfactor-requested-with": "APIClient", "x-certificateformat": "PEM", "Accept": "application/json", } return self._http_request(method="POST", url_suffix="Enrollment/CSR", headers=headers, json_data=body) """ HELPER FUNCTIONS """ def get_enrollment_csr_context_template_ec(raw_response: dict) -> tuple[list, list]: """ Get raw response of Enrollment csr templates and parse to ec Args: raw_response: Enrollment csr templates list Returns: List of Enrollment csr templates entry context for human readable """ entry_context = [] human_readable = [] if raw_response: for template in raw_response["Templates"]: entry_context.append(assign_params(Name=template.get("Name"), CAs=template.get("CAs")[0].get("Name"))) human_readable.append(assign_params(Name=template.get("Name"), CAs=template.get("CAs")[0].get("Name"))) return entry_context, human_readable def post_enrollment_csr_command_ec(raw_response: dict) -> tuple[list, list]: """ Parse the post enrollment CSR > Parse the Certificate received from keyFactor Args: raw_response: Certificate Info received from keyFactor Returns: List of Certificate info """ entry_context = [] human_readable = [] if raw_response: certificateinfo: dict = raw_response["CertificateInformation"] regex = r"-----BEGIN CERTIFICATE-----[\w\W]*END CERTIFICATE-----" formated_certs: list = [] certificates: list = certificateinfo["Certificates"] for certs in certificates: matches = re.findall(regex, certs) formated_certs = formated_certs + matches new_formated_certs = [] for cert in formated_certs: new_formated_certs.append(cert.replace("\r\n", " ")) formated_chain = new_formated_certs[1] + " " + new_formated_certs[2] entry_context.append( assign_params( SerialNumber=certificateinfo.get("SerialNumber"), IssuerDN=certificateinfo.get("IssuerDN"), Thumbprint=certificateinfo.get("Thumbprint"), KeyfactorID=certificateinfo.get("KeyfactorID"), KeyfactorRequestId=certificateinfo.get("KeyfactorRequestId"), Certificates=certificateinfo.get("Certificates"), RequestDisposition=certificateinfo.get("RequestDisposition"), DispositionMessage=certificateinfo.get("DispositionMessage"), EnrollmentContext=certificateinfo.get("EnrollmentContext"), formated_certs=new_formated_certs, formated_cert=new_formated_certs[0], formated_chain=formated_chain, ) ) human_readable.append( assign_params( SerialNumber=certificateinfo.get("SerialNumber"), IssuerDN=certificateinfo.get("IssuerDN"), Thumbprint=certificateinfo.get("Thumbprint"), KeyfactorID=certificateinfo.get("KeyfactorID"), KeyfactorRequestId=certificateinfo.get("KeyfactorRequestId"), Certificates=certificateinfo.get("Certificates"), RequestDisposition=certificateinfo.get("RequestDisposition"), DispositionMessage=certificateinfo.get("DispositionMessage"), EnrollmentContext=certificateinfo.get("EnrollmentContext"), formated_certs=formated_certs, formated_cert=formated_certs[0], formated_chain=formated_chain, ) ) return entry_context, human_readable """ COMMANDS """ @logger def post_enrollment_csr_command( client: Client, csr_base64: dict, cert_authority: str, include_chain: str, template: str, metadata: str, keyAlgorithm: str, sans_ip4: str = "", ) -> tuple[object, dict, Union[list, dict]]: """ Post Enrollment CSR Send the certifcate CSR and return the certificate created Args: client: Client object with request csr_base64: The base-64 encoded CSR that will be passed in for enrollment. cert_authority: A string that sets the name of the certificate authority include_chain: A Boolean that sets whether to include the certificate chain in theresponse (true) or not (false). The default is false. template: A string that sets the name of the certificate template thatshould be used to issue the certificate. sans_ip4: Subject alternative name (SAN) ip addresses keyAlgorithm: A string indicating the cryptographic algorithm to use to generate the key metadata: An array of key/value pairs that set the values for the metadata fields Returns: human readable (markdown format), entry context and raw response """ import json if keyAlgorithm != "": csr_base64_changed = "" for csr in csr_base64["csrs"]: if csr["keyAlgorithm"] == keyAlgorithm: csr_base64_changed = csr["csr"] now_iso8601 = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z") raw_response: dict = client.post_enrollment_csr( csr_base64=csr_base64_changed, cert_authority=cert_authority, include_chain=include_chain, time_stamp=now_iso8601, template=template, keyAlgorithm=keyAlgorithm, sans_ip4=sans_ip4, metadata=json.loads(metadata), ) if raw_response: title = f"{INTEGRATION_NAME} - Post Enrollment CSR" entry_context, human_readable_ec = post_enrollment_csr_command_ec(raw_response) context_entry: dict = { f"{INTEGRATION_CONTEXT_NAME}.CertInfo.Lists(val.UniqueID && val.UniqueID == obj.UniqueID &&" f" val.UpdateDate && val.UpdateDate == obj.UpdateDate)": entry_context } human_readable = tableToMarkdown(name=title, t=human_readable_ec, removeNull=True) return human_readable, context_entry, raw_response else: return {}, {}, raw_response @logger def get_enrollment_csr_context_template_command(client: Client, *_) -> tuple[object, dict, Union[list, dict]]: """Get all Enrollment CSR Context My (templates) Args: client: Client object with request Returns: human readable (markdown format), entry context and raw response """ raw_response: dict = client.get_enrollment_csr_context_template() if raw_response: title = f"{INTEGRATION_NAME} - Get enrollment csr context my" entry_context, human_readable_ec = get_enrollment_csr_context_template_ec(raw_response) context_entry: dict = { f"{INTEGRATION_CONTEXT_NAME}.CSRTemplate.Lists(val.UniqueID && val.UniqueID == obj.UniqueID &&" f" val.UpdateDate && val.UpdateDate == obj.UpdateDate)": entry_context } human_readable = tableToMarkdown(name=title, t=human_readable_ec, removeNull=True) return human_readable, context_entry, raw_response else: return f"{INTEGRATION_NAME} - Could not find any results for given query", {}, {} @logger def test_module_command(client: Client, *_) -> tuple[None, None, str]: """Performs a basic GET request to check if the API is reachable and authentication is successful. Args: client: Client object with request *_: Usually demisto.args() Returns: 'ok' if test successful. Raises: DemistoException: If test failed. """ results = client.test_module() if "Templates" in results or "CertificateInformation" in results: return None, None, "ok" raise DemistoException(f"Test module failed, {results}") def main(): params = demisto.params() verify_ssl = not params.get("insecure", False) proxy = params.get("proxy") username = params.get("credentials", {}).get("identifier") password = params.get("credentials", {}).get("password") client = Client(base_url=params.get("host"), verify=verify_ssl, proxy=proxy, auth=(username, password)) command = demisto.command() commands = { "test-module": test_module_command, f"{INTEGRATION_COMMAND_NAME}-get-enrollment-csr": get_enrollment_csr_context_template_command, f"{INTEGRATION_COMMAND_NAME}-post-enrollment-csr": post_enrollment_csr_command, } try: readable_output, outputs, raw_response = commands[command](client=client, **demisto.args()) results = CommandResults(outputs=outputs, readable_output=readable_output, raw_response=raw_response) return_results(results) except Exception as e: err_msg = f"Error in {INTEGRATION_NAME} Integration [{e}]" return_error(err_msg, error=e) if __name__ in ("__main__", "__builtin__", "builtins"): main()