import traceback from typing import Any import demistomock as demisto import requests import urllib3 from CommonServerPython import * # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ SECURITYSCORECARD_DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" """ CLIENT CLASS """ class SecurityScorecardClient(BaseClient): """Client class that interacts with the SecurityScorecard API Attributes: ``username`` (``str``): SecurityScorecard username/email. ``api_key`` (``str``): SecurityScorecard API token. ``max_fetch`` (``int``): Maximum alerts to fetch. """ def __init__(self, base_url, verify, proxy, headers, username, api_key, max_fetch=50): """ Args: base_url (str): SecurityScorecard base URL. verify (bool): Whether to verify certificates. proxy (bool): Whether to use Cortex XSOAR proxy. headers (dict): Dictionary holding the HTTP headers. """ super().__init__(base_url, verify=verify, proxy=proxy, headers=headers) self.username = username self.api_key = api_key self.max_fetch = max_fetch def get_portfolios(self) -> dict[str, Any]: return self.http_request_wrapper(method="GET", url_suffix="portfolios") def get_companies_in_portfolio( self, portfolio: str, grade: str | None, industry: str | None, vulnerability: str | None, issue_type: str | None, had_breach_within_last_days: int | None, ) -> dict[str, Any]: request_params: dict[str, Any] = assign_params( grade=grade, industry=industry, vulnerability=vulnerability, issue_type=issue_type, had_breach_within_last_days=had_breach_within_last_days, ) return self.http_request_wrapper(method="GET", url_suffix=f"portfolios/{portfolio}/companies", params=request_params) def get_company_score(self, domain: str) -> dict[str, Any]: return self.http_request_wrapper(method="GET", url_suffix=f"companies/{domain}") def get_company_factor_score(self, domain: str, severity_in: list[str] | None) -> dict[str, Any]: request_params: dict[str, Any] | None = {"severity_in": severity_in} if severity_in else None return self.http_request_wrapper(method="GET", url_suffix=f"companies/{domain}/factors", params=request_params) def get_company_events(self, domain: str, date_from: str, date_to: str) -> dict[str, Any]: request_params: dict[str, Any] = assign_params(date_from=date_from, date_to=date_to) return self.http_request_wrapper(method="GET", url_suffix=f"companies/{domain}/history/events", params=request_params) def get_company_event_findings(self, domain: str, date: str, issue_type: str, status: str) -> dict[str, Any]: request_params: dict[str, Any] = assign_params(group_status=status) return self.http_request_wrapper( method="GET", url_suffix=f"companies/{domain}/history/events/{date}/issues/{issue_type}", params=request_params ) def get_company_issue_findings(self, domain: str, issue_type: str) -> dict[str, Any]: return self.http_request_wrapper(method="GET", url_suffix=f"companies/{domain}/issues/{issue_type}") def get_company_historical_scores(self, domain: str, _from: str, to: str, timing: str) -> dict[str, Any]: request_params: dict[str, Any] = assign_params(to=to, timing=timing, domain=domain) # assign_params cannot accept 'from' as a parameter since it's a Python keyword if _from: request_params["from"] = _from return self.http_request_wrapper(method="GET", url_suffix=f"companies/{domain}/history/score", params=request_params) def get_company_historical_factor_scores(self, domain: str, _from: str, to: str, timing: str) -> dict[str, Any]: request_params: dict[str, Any] = assign_params(to=to, timing=timing) # Cannot use assign_params with reserved Python keyword 'from' if _from: request_params["from"] = _from return self.http_request_wrapper( method="GET", url_suffix=f"companies/{domain}/history/factors/score", params=request_params ) def get_issue_metadata(self, issue_type: str) -> dict[str, Any]: return self.http_request_wrapper(method="GET", url_suffix=f"metadata/issue-types/{issue_type}") def create_alert_subscription( self, event_type: str, delivery: dict[str, Any], ) -> dict[str, Any]: payload: dict[str, Any] = assign_params( event_type=event_type, delivery=delivery, ) return self.http_request_wrapper(method="POST", url_suffix="subscriptions", json_data=payload) def delete_alert(self, id: str) -> None: return self.http_request_wrapper(method="DELETE", url_suffix=f"subscriptions/{id}", return_empty_response=True) def get_subscriptions(self) -> dict[str, Any]: query_params: dict[str, Any] = assign_params( username=self.username, ) return self.http_request_wrapper(method="GET", url_suffix="subscriptions", params=query_params) def get_alerts_last_week(self, email: str, portfolio_id: str | None) -> dict[str, Any]: query_params: dict[str, Any] = assign_params(portfolio=portfolio_id) return self.http_request_wrapper( method="GET", url_suffix=f"users/by-username/{email}/notifications/recent", params=query_params ) def get_domain_services(self, domain: str) -> dict[str, Any]: return self.http_request_wrapper(method="GET", url_suffix=f"companies/{domain}/services") def fetch_alerts(self, page_size: int, page: int) -> dict[str, Any]: query_params: dict[str, Any] = assign_params( username=self.username, page_size=page_size, sort="date", order="asc", page=page ) return self.http_request_wrapper( method="GET", url_suffix=f"users/by-username/{self.username}/notifications/recent", params=query_params ) def http_request_wrapper( self, method: str, url_suffix: str | None = None, params: dict | None = None, json_data: dict | None = None, return_empty_response: bool | None = False, ): """Wrapper for the ``http_request`` function Args: ``self`` (``SecurityScorecardClient``). ``method`` (``str``): The HTTP method. ``url_suffix`` (``Optional[str]``): The URL suffix, appended to the base URL. Defaults to None. ``params`` (``Optional[dict]``): The query parameters sent in the HTTP request. Defaults to None. ``json_data`` (``Optional[dict]``): The payload to be sent in the HTTP request in JSON format. Defaults to None. Return: ``dict`` or ``str`` or ``requests.Response`` """ return super()._http_request( method=method, url_suffix=url_suffix, params=params, json_data=json_data, error_handler=self.error_handler, return_empty_response=return_empty_response, ) @staticmethod def error_handler(response: requests.Response): """ Error handler for the API requests Args: response (requests.Response): The server's response to the HTTP request. """ try: error_response_json = response.json().get("error") raise DemistoException(f'{error_response_json.get("message")} ({error_response_json.get("statusCode")})') except ValueError: raise DemistoException(f"Error parsing response as JSON. Response: {response.status_code} {response.content!s}") """ HELPER FUNCTIONS """ def get_last_run( last_run: str = demisto.getLastRun().get("last_run"), first_fetch: str = demisto.params().get("first_fetch", "2 days") ) -> datetime: """ Helper function to return the last incident fetch runtime as a `datetime` object. It uses the datetime of last_run from the demisto instance and first_fetch parameter. Args: ``last_run`` (``str``): last run datetime string of fetch ``first_fetch`` (``str``): first fetch from integration parameters Returns: ``datetime`` representing the last fetch occurred. """ # Check for existence of last run # When integration runs for the first time, it will not exist # Set 2 days by default if the first fetch parameter is not set if last_run: demisto.debug(f"Last run already exists: '{last_run}'") return arg_to_datetime(last_run).replace(tzinfo=None) # type: ignore else: demisto.debug(f"First fetch is defined as '{first_fetch}'") days_ago = first_fetch fetch_days_ago = arg_to_datetime(arg=days_ago, arg_name="first_fetch", required=False) demisto.debug(f"getLastRun is 'None' in Integration context, using parameter '{days_ago}' value '{fetch_days_ago}'") return fetch_days_ago.replace(tzinfo=None) # type: ignore def incidents_to_import(alerts: list[dict[str, Any]], last_run: datetime = get_last_run()) -> list[dict[str, Any]]: """ Helper function to filter events that need to be imported. It filters the events based on the `created_at` timestamp. Function will only be called if the SecurityScorecard API returns more than one alert. Args: ``alerts``(``List[Dict[str, Any]]``): A list of alerts to sort through. Returns: ``List[Dict[str, Any]]``: Alerts to import """ incidents: list[dict[str, Any]] = [] # Check if there are more than 0 alerts if alerts: # The alerts are sorted by ascending date so last alert is the most recent most_recent_alert = alerts[-1] most_recent_alert_created_date = most_recent_alert.get("created_at") most_recent_alert_datetime = arg_to_datetime(most_recent_alert_created_date).replace(tzinfo=None) # type: ignore for alert in alerts: demisto.debug(f"iterating alert id '{alert}'...") alert_id = alert.get("id") alert_created_at = alert.get("created_at") # alert_created_at includes a timezone whereas arg_to_datetime doesn't # therefore we need to eliminate tz info and set seconds=0 # preventing err "can't compare offset-naive and offset-aware datetimes" alert_datetime = arg_to_datetime(alert_created_at).replace(tzinfo=None).replace(second=0) # type: ignore company_name: str = alert.get("company_name") # type: ignore change_type: str = alert.get("change_type") # type: ignore demisto.debug(f"alert_created_at: {alert_created_at}") demisto.debug(f"alert_datetime: {alert_datetime}") demisto.debug(f"last_run: {last_run}") debug_msg = f"import alert '{alert_id}'? (last_run < alert_datetime): {(last_run < alert_datetime)}" # type: ignore demisto.debug(debug_msg) if alert_datetime > last_run: # type: ignore incident = {} incident["name"] = f"{company_name} {change_type.replace('_', ' ').title()}" incident["occurred"] = alert_datetime.strftime(format=DATE_FORMAT) # type: ignore incident["rawJSON"] = json.dumps(alert) incidents.append(incident) demisto.debug( f"Setting setLastRun as alert most recent: \ {most_recent_alert_datetime.strftime(format=DATE_FORMAT)}" # type: ignore ) demisto.setLastRun( { "last_run": most_recent_alert_datetime.strftime(format=DATE_FORMAT) # type: ignore } ) demisto.debug("Finished setLastRun") # If there are no alerts then we can't use the most recent alert timestamp # So we'll use the last run timestamp (last alert fetch modified date) else: demisto.debug(f"No alerts retrieved, setting last_run to last modified time ({last_run})") demisto.setLastRun(last_run) return incidents """ COMMAND FUNCTIONS """ def test_module(client: SecurityScorecardClient, incident_fetch_interval: str | None) -> str: """Tests API connectivity and authentication Runs the fetch-alerts mechanism to validate all integration parameters Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client Returns: ``str``: 'ok' if test passed, anything else will fail the test. """ demisto.debug("Initialized test module...") interval = arg_to_number(arg=incident_fetch_interval, arg_name="incident_fetch_interval", required=False) if interval > 1440 * 2: # type: ignore return "Test failed. Incident Fetch Interval is greater than 2 days." max_incidents = int(client.max_fetch) if max_incidents > 50: return "Test failed. Max Fetch is larger than 50." client.fetch_alerts(page_size=1, page=1) demisto.debug("Test module successful") return "ok" # region Methods # --------------- def portfolios_list_command(client: SecurityScorecardClient, args: dict[str, str]) -> CommandResults: """List all Portfolios you have access to. See https://securityscorecard.readme.io/reference#get_portfolios Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, str]``): Portfolio fetch limit Returns: ``CommandResults``: The results of the command. """ limit = arg_to_number( # type: ignore arg=args.get("limit", "50"), arg_name="limit", required=False ) portfolios = client.get_portfolios() portfolios_total = int(portfolios.get("total")) # type: ignore # Check that API returned more than 0 portfolios if portfolios_total == 0: return CommandResults( readable_output="No Portfolios were found in your account. Please create a new one and try again.", outputs_prefix=None, outputs=None, raw_response=portfolios, outputs_key_field=None, ) # API response is a dict with 'entries' entries = portfolios.get("entries") # If the number of portfolios returned is larger than the configured limit # filter the first elements if portfolios_total > limit: # type: ignore demisto.debug(f"portfolios_total ({portfolios_total}) > limit ({limit}), slicing number of entries") entries = entries[:limit] # type: ignore markdown = tableToMarkdown(f"Your SecurityScorecard Portfolios (first {limit})", entries, headers=["id", "name", "privacy"]) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Portfolio", outputs_key_field="id", outputs=entries, raw_response=portfolios, ) return results def portfolio_list_companies_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Retrieve all companies in portfolio. https://securityscorecard.readme.io/reference#get_portfolios-portfolio-id-companies Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, Any]``): Includes - portfolio ID - Grade filter - Industry filter - Vulnerability filter - Issue type filter - Filter breach days back Returns: ``CommandResults``: The results of the command. """ portfolio_id = args.get("portfolio_id") # type: ignore grade = args.get("grade") industry_arg = args.get("industry") vulnerability = args.get("vulnerability") issue_type = args.get("issue_type") # We need to capitalize the industry to conform to API industry = str.upper(industry_arg) if industry_arg else None # type: ignore had_breach_within_last_days = arg_to_number( # type: ignore arg=args.get("had_breach_within_last_days"), arg_name="had_breach_within_last_days", required=False ) response = client.get_companies_in_portfolio( portfolio=portfolio_id, # type: ignore grade=grade, industry=industry, vulnerability=vulnerability, issue_type=issue_type, had_breach_within_last_days=had_breach_within_last_days, # type: ignore ) # Check if the portfolio has more than 1 company total_portfolios = int(response.get("total")) # type: ignore if not total_portfolios > 0: return CommandResults( readable_output=f"No companies found in Portfolio '{portfolio_id}'. Please add a company to it and retry.", raw_response=response, outputs_key_field=None, ) companies = response.get("entries") title = f"**{total_portfolios}** companies found in Portfolio {portfolio_id}\n" markdown = tableToMarkdown( title, companies, headers=["domain", "name", "score", "last30days_score_change", "industry", "size"] ) results = CommandResults( outputs_prefix="SecurityScorecard.Portfolio.Company", readable_output=markdown, outputs=companies, raw_response=response, outputs_key_field="name", ) return results def company_score_get_command(client: SecurityScorecardClient, args: dict[str, str]) -> CommandResults: """Retrieve company overall score. See https://securityscorecard.readme.io/reference#get_companies-scorecard-identifier-factors Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, str]``): The domain to get the score for. Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") score = client.get_company_score(domain=domain) # type: ignore markdown = tableToMarkdown( f"Domain {domain} Scorecard", score, headers=["name", "grade", "score", "industry", "last30day_score_change", "size"] ) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Company.Score", outputs=score, raw_response=score, outputs_key_field="name", ) return results def company_factor_score_get_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Retrieve company factor score and scores See https://securityscorecard.readme.io/reference#get_companies-scorecard-identifier-factors Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, Any]``): The domain and severity filter Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") severity = args.get("severity") response = client.get_company_factor_score(domain=domain, severity_in=severity) # type: ignore entries = response["entries"] factor_scores = [] for entry in entries: score = { "name": entry.get("name"), "grade": entry.get("grade"), "score": entry.get("score"), "issues": len(entry.get("issue_summary")), "issue details": entry.get("issue_summary"), } factor_scores.append(score) markdown = tableToMarkdown(f"Domain {domain} Scorecard", factor_scores, headers=["name", "grade", "score", "issues"]) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Company.Factor", outputs=entries, raw_response=response, outputs_key_field="name", ) return results def company_history_score_get_command(client: SecurityScorecardClient, args: dict[str, str]) -> CommandResults: """Retrieve company historical scores See https://securityscorecard.readme.io/reference/get_companies-scorecard-identifier-history-score Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client. ``args`` (``Dict[str, str]``): Domain, start date, end date, timing. Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") _from = args.get("from") to = args.get("to") timing = args.get("timing") response = client.get_company_historical_scores(domain=domain, _from=_from, to=to, timing=timing) # type: ignore entries = response.get("entries") markdown = tableToMarkdown(f"Historical Scores for Domain [`{domain}`](https://{domain})", entries, headers=["date", "score"]) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Company.ScoreHistory", outputs=entries, raw_response=response, outputs_key_field="date", ) return results def company_events_get_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Retrieve company events See https://securityscorecard.readme.io/reference/get_companies-scorecard-identifier-history-events Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, Any]``): The domain, the initial date and the end (date_from, date_to) Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") date_from = args.get("date_from") date_to = args.get("date_to") response = client.get_company_events(domain=domain, date_to=date_to, date_from=date_from) # type: ignore entries = response["entries"] events = [] for entry in entries: event = { "ssc_event_id": entry.get("id"), "date": entry.get("date"), "status": entry.get("group_status"), "issue_count": entry.get("issue_count"), "score_impact": entry.get("total_score_impact"), "issue_type": entry.get("issue_type"), "severity": entry.get("severity"), "factor": entry.get("factor"), "ssc_detail_url": entry.get("detail_url"), } events.append(event) markdown = tableToMarkdown( f"Domain {domain} Events", events, headers=[ "ssc_event_id", "date", "status", "factor", "issue_type", "severity", "issue_count", "score_impact", "ssc_detail_url", ], ) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Company.Events", outputs=entries, raw_response=response, outputs_key_field="ssc_event_id", ) return results def company_event_findings_get_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Get an issue_type's historical findings in a scorecard See (example issue_type): https://securityscorecard.readme.io/reference/get_companies-scorecard-identifier-history-events-effective-date-issues-active-cve-exploitation-attempted-1 Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, Any]``): domain, date, issue_type, status Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") date = args.get("date") issue_type = args.get("issue_type") status = args.get("status") response = client.get_company_event_findings(domain=domain, date=date, issue_type=issue_type, status=status) # type: ignore entries = response["entries"] events = [] for entry in entries: # some issue types have domains, IPs and/or ports, but not all of them do if "domain" in entry: domain = entry.get("domain") elif "target" in entry: domain = entry.get("target") else: domain = "" if "ip" in entry: ip = entry.get("ip") elif "src_ip" in entry: ip = entry.get("src_ip") elif "ip_address" in entry: ip = entry.get("ip_address") elif "connection_attributes" in entry: ip = entry.get("connection_attributes").get("dst_ip") else: ip = "" if "protocol" in entry: protocol = entry.get("protocol") elif "scheme" in entry: protocol = entry.get("scheme") elif "connection_attributes" in entry: protocol = entry.get("connection_attributes").get("protocol") else: protocol = "" if "port" in entry: port = entry.get("port") elif "connection_attributes" in entry: port = entry.get("connection_attributes").get("dst_port") else: port = "" event = { "parent_domain": entry.get("parent_domain"), "count": entry.get("count"), "status": entry.get("group_status"), "first_seen_time": entry.get("first_seen_time"), "last_seen_time": entry.get("last_seen_time"), # the following details may or may not be populated "port": port, "domain_name": domain, "ip_address": ip, "protocol": protocol, "observations": entry.get("observations"), "issue_type": issue_type, } events.append(event) markdown = tableToMarkdown( f"Domain {domain} Findings for {issue_type}", events, headers=[ "parent_domain", "issue_type", "count", "status", "first_seen_time", "last_seen_time", "port", "domain_name", "ip_address", "protocol", "observations", ], ) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Company.Findings", outputs=entries, raw_response=response, outputs_key_field="issue_id", ) return results def company_history_factor_score_get_command(client: SecurityScorecardClient, args: dict[str, str]) -> CommandResults: """Retrieve company historical factor scores See https://securityscorecard.readme.io/reference#get_companies-scorecard-identifier-history-factors-score Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client. ``args`` (``Dict[str, str]``): Domain, start date, end date, timing. Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") _from = args.get("from") to = args.get("to") timing = args.get("timing") response = client.get_company_historical_factor_scores(domain=domain, _from=_from, to=to, timing=timing) # type: ignore entries = response["entries"] factor_scores = [] for entry in entries: factors = entry.get("factors") factor_row = "" for factor in factors: factor_name = factor.get("name").title().replace("_", " ") factor_score = factor.get("score") factor_row = factor_row + f"{factor_name}: {factor_score}\n" score = {"date": entry.get("date").split("T")[0], "factors": factor_row} factor_scores.append(score) markdown = tableToMarkdown(f"Historical Factor Scores for Domain {domain})", factor_scores) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Company.FactorHistory", outputs=entries, raw_response=response, outputs_key_field="date", ) return results def issue_metadata_get_command(client: SecurityScorecardClient, args: dict[str, str]) -> CommandResults: """Retrieve description and recommendation for an issue. See https://securityscorecard.readme.io/reference/get_metadata-issue-types-type-1 Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, str]``): The issue type to retrieve metadata for. Returns: ``CommandResults``: The results of the command. """ issue_type = args.get("issue_type") metadata = client.get_issue_metadata(issue_type=issue_type) # type: ignore markdown = tableToMarkdown( f"Issue Type {issue_type}", metadata, headers=["key", "severity", "factor", "title", "short_description", "long_description", "recommendation"], ) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Metadata.Issues", outputs=metadata, raw_response=metadata, outputs_key_field="key", ) return results @logger def alert_grade_change_create_command(client: SecurityScorecardClient, args: dict[str, str]) -> CommandResults: """Create an alert based on grade change. This function creates an alert subscription for grade changes in SecurityScorecard. It supports alerts for both overall and factor-specific grade changes. See POST /subscriptions Args: client (SecurityScorecardClient): The SecurityScorecard client instance. args (Dict[str, str]): The command arguments containing: - change_direction (str): The direction of the grade change ('drops' or 'raises'). - score_types (str): The types of scores to monitor (e.g., 'overall', 'network_security'). - target (str): The target entity for the alert (e.g., 'any_followed_company', 'my_scorecard'). - portfolio (str): The portfolio ID to monitor. Returns: CommandResults: The results of the command, including the created alert ID. Raises: DemistoException: If both 'portfolio' and 'target' are set, or if neither is set. """ change_direction = args.get("change_direction") score_types = argToList(args.get("score_types")) target = args.get("target") portfolio = args.get("portfolio") # Only one argument between portfolio and target should be defined # Return error if neither of them is defined or if both are defined if portfolio and target: raise DemistoException("Both 'portfolio' and 'target' argument have been set. Please remove one of them and try again.") elif not (target or portfolio): raise DemistoException("Either 'portfolio' or 'target' argument must be given") filters_changes_value = ( ("factor_grade_drop" if change_direction == "drops" else "factor_grade_raise") if "overall" not in score_types else ("grade_drop" if change_direction == "drops" else "grade_raise") ) name = ( f"Alert me when {target or 'portfolio'} {change_direction} in " f"{'factor(s) ' if 'overall' not in score_types else ''}grade" ) delivery: dict[str, Any] = { "workflow": { "steps": [ { "action": { "value": "alert_teammate", "recipientType": { "value": "self", }, }, }, ], "name": name, "filters": { "changes": { "value": filters_changes_value, "grade": { "value": "any", }, "factor": { "value": [], }, }, "scorecards": { "value": ( "followed" if target == "any_followed_company" else "my_scorecard" if target == "my_scorecard" else "in_portfolio" ), "portfolio_id": {"value": portfolio}, }, }, }, } if not portfolio: del delivery["workflow"]["filters"]["scorecards"]["portfolio_id"] if "overall" not in score_types: if "any_factor_score" in score_types: delivery["workflow"]["filters"]["changes"]["factor"]["value"] = [ "network_security", "dns_health", "patching_cadence", "endpoint_security", "ip_reputation", "application_security", "cubit_score", "hacker_chatter", "leaked_information", "social_engineering", ] else: delivery["workflow"]["filters"]["changes"]["factor"]["value"] = score_types else: del delivery["workflow"]["filters"]["changes"]["factor"] response = client.create_alert_subscription( event_type="scorecard.changed", delivery=delivery, ) demisto.debug(f"Response received: {response}") alert_id = response.get("id") markdown = f"Alert **{alert_id}** created" results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Alerts.GradeChangeAlert", outputs=alert_id, raw_response=response, outputs_key_field="id", ) return results def alert_score_threshold_create_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """ Create an alert based on a score threshold being met. This function creates an alert subscription in SecurityScorecard when a specified score threshold is met. The alert can be configured to trigger on changes in overall score or specific factor scores, and can be targeted to a specific portfolio or a general target. See POST /subscriptions Args: client (SecurityScorecardClient): The SecurityScorecard client instance used to interact with the API. args (Dict[str, Any]): A dictionary of arguments for the command. - change_direction (str): The direction of the score change ('rises_above' or 'drops_below'). - threshold (int): The score threshold value. - score_types (List[str]): The types of scores to monitor (e.g., 'overall', 'network_security'). - target (str): The target for the alert (e.g., 'any_followed_company', 'my_scorecard'). - portfolio (str): The portfolio ID to monitor. Returns: CommandResults: The results of the command, including the alert ID and raw response from the API. Raises: DemistoException: If both 'portfolio' and 'target' are provided, or if neither is provided. """ change_direction = args.get("change_direction") threshold = arg_to_number(args.get("threshold")) score_types = argToList(args.get("score_types")) target = args.get("target") portfolio = args.get("portfolio") # Only one argument between portfolio and target should be defined # Return error if neither of them is defined or if both are defined if portfolio and target: raise DemistoException("Both 'portfolio' and 'target' argument have been set. Please remove one of them and try again.") elif not (target or portfolio): raise DemistoException("Either 'portfolio' or 'target' argument must be given") # filters_changes_value can be # 'score_rise_threshold', 'score_drop_threshold', 'factor_score_rise_threshold', 'factor_score_drop_threshold' filters_changes_value = ( ("factor_score_drop_threshold" if change_direction == "drops_below" else "factor_score_rise_threshold") if "overall" not in score_types else ("score_drop_threshold" if change_direction == "drops_below" else "score_rise_threshold") ) name = ( f"Alert me when {target or 'portfolio'} " f"{'factor(s)' if 'overall' not in score_types else 'overall'} score {change_direction} {threshold} pts" ) delivery: dict[str, Any] = { "workflow": { "steps": [ { "action": { "value": "alert_teammate", "recipientType": { "value": "self", }, }, }, ], "name": name, "filters": { "changes": { "value": filters_changes_value, "threshold": { "value": f"{threshold}", }, "factor": { "value": [], }, }, "scorecards": { "value": ( "followed" if target == "any_followed_company" else "my_scorecard" if target == "my_scorecard" else "in_portfolio" ), "portfolio_id": {"value": portfolio}, }, }, }, } if not portfolio: del delivery["workflow"]["filters"]["scorecards"]["portfolio_id"] if "overall" not in score_types: if "any_factor_score" in score_types: delivery["workflow"]["filters"]["changes"]["factor"]["value"] = [ "network_security", "dns_health", "patching_cadence", "endpoint_security", "ip_reputation", "application_security", "cubit_score", "hacker_chatter", "leaked_information", "social_engineering", ] else: delivery["workflow"]["filters"]["changes"]["factor"]["value"] = score_types else: del delivery["workflow"]["filters"]["changes"]["factor"] response = client.create_alert_subscription( event_type="scorecard.changed", delivery=delivery, ) demisto.debug(f"Response received: {response}") alert_id = response.get("id") markdown = f"Alert **{alert_id}** created" results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.Alerts.ScoreThresholdAlert", outputs=alert_id, raw_response=response, outputs_key_field="id", ) return results def alert_delete_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Delete an alert See DELETE /subscriptions/:id Args: client (SecurityScorecardClient): SecurityScorecard client args (Dict[str, Any]): Dictionary of arguments specified in the command Returns: CommandResults: The results of the command. """ alert_id = args.get("alert_id") client.delete_alert(id=alert_id) # type: ignore markdown = f"Alert **{alert_id}** deleted" # type: ignore results = CommandResults(readable_output=markdown, raw_response=None, outputs_key_field=None) return results def alerts_list_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Retrieve alerts triggered in the last week See https://securityscorecard.readme.io/reference/get_users-by-username-username-notifications-recent-1 Args: client (SecurityScorecardClient): SecurityScorecard client args (Dict[str, Any]): Dictionary of arguments specified in the command Returns: CommandResults: The results of the command. """ email = client.username demisto.debug(f"email: {email}") portfolio_id = args.get("portfolio_id") demisto.debug(f"Sending request to retrieve alerts with arguments {args}") response = client.get_alerts_last_week(email=email, portfolio_id=portfolio_id) entries = response.get("entries") # type: ignore alerts: list[dict[str, str]] = [] for entry in entries: # type: ignore content: dict[str, str] = { "company": entry.get("company_name"), "domain": entry.get("domain"), "datetime": entry.get("created_at"), } change_data = entry.get("change_data") if change_data: try: for change in change_data: # content["change data"] = change content["alert id"] = change.get("workflow", {}).get("id", "N/A") content["trigger"] = change.get("score_change", {}).get("trigger_value", "N/A") content["grade"] = change.get("score_change", {}).get("grade", "N/A") content["score"] = change.get("score_change", {}).get("score", "N/A") # Handle additional optional fields content["factors"] = change.get("score_change", {}).get("factors", "N/A") content["issues"] = change.get("score_change", {}).get("issues", "N/A") except (json.JSONDecodeError, TypeError, KeyError) as e: demisto.error(f"Error processing change_data: {e!s}") # Old alerts system had the possibility of multiple portfolios, new rules system allows one portfolios = entry.get("portfolios") if portfolios: try: for portfolio in portfolios: content["target portfolio"] = portfolio.get("id") except (json.JSONDecodeError, TypeError, KeyError) as e: demisto.error(f"Error processing change_data: {e!s}") alerts.append(content) markdown = tableToMarkdown(f"Latest Alerts for user {email}", alerts) results = CommandResults( outputs_prefix="SecurityScorecard.Alerts.Alert", outputs_key_field="id", readable_output=markdown, outputs=alerts, raw_response=response, ) return results def company_services_get_command(client: SecurityScorecardClient, args: dict[str, str]) -> CommandResults: """Retrieve the service providers of a domain See https://securityscorecard.readme.io/reference#get_companies-domain-services Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, str]``): Domain. Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") response = client.get_domain_services(domain=domain) # type: ignore entries = response.get("entries") services = [] if entries: for entry in entries: # type: ignore categories = entry.get("categories") for category in categories: service = {} service["vendor_domain"] = entry.get("vendor_domain") service["category"] = category services.append(service) markdown = tableToMarkdown(f"Services for domain [{domain}](https://{domain})", services) results = CommandResults( outputs_prefix="SecurityScorecard.Company.Services", outputs=entries, readable_output=markdown, raw_response=response, outputs_key_field="category", ) else: results = CommandResults(readable_output=f"Error returning services for domain '{domain}'", raw_response=response) return results def fetch_alerts(client: SecurityScorecardClient): """ Fetch incidents/alerts from SecurityScorecard API See https://securityscorecard.readme.io/reference#get_users-by-username-username-notifications-recent The API is updated on a daily basis therefore `incidentFetchInterval` is set to 1440 (minutes per day) The API returns all alerts received in the last week. Every alert has a `"created_at"` parameter to notify when the alert was triggered. This method will create incidents only for alerts that occurred on the day the alert was created. Args: client (SecurityScorecardClient): SecurityScorecard client Returns: None: It calls demisto.incidents() to import incidents. """ # Set the query size max_incidents = arg_to_number(client.max_fetch) # type: ignore # Set initial page initial_page = 1 # Initial call will request the first page. results = client.fetch_alerts(page_size=max_incidents, page=initial_page) # type: ignore first_fetch_alerts = results.get("entries") size = results.get("size") # The number of fetches needed to retrieve all alerts # is the total number of alerts divided by the max fetch size fetches_required = int(size / max_incidents) # type: ignore demisto.debug(f"API returned {size} alerts. Fetches required to retrieve all alerts: {fetches_required}") # Check if the API returned any alerts if size > 0: # type: ignore # If there are no fetches required, import the alerts pulled from the initial request if fetches_required == 0: incidents = incidents_to_import(alerts=first_fetch_alerts) # type: ignore # Check if any incidents should be imported according to last run time timestamp if incidents: demisto.debug(f"{len(incidents)} Incidents will be imported") demisto.debug(f"Incidents: {incidents}") demisto.incidents(incidents) else: demisto.debug("No incidents will be imported.") demisto.incidents([]) # In case we cannot import all alerts in one go, # we paginate. if fetches_required > 0: alerts_to_import = [] # Add the alerts from the first fetch. first_fetch_incidents = incidents_to_import(alerts=first_fetch_alerts) # type: ignore if first_fetch_incidents: alerts_to_import.extend(first_fetch_incidents) demisto.debug(f"Adding {len(first_fetch_incidents)} alerts from first fetch to total alerts to import") demisto.debug(f"Total alerts currently in list: {len(alerts_to_import)}") # Iterate to bring the rest of the alerts for fetch_iteration in range(initial_page + 1, fetches_required + 2): demisto.debug(f"Fetch iteration {fetch_iteration} started...") results = client.fetch_alerts(page_size=max_incidents, page=fetch_iteration) # type: ignore alerts = results.get("entries") incidents = incidents_to_import(alerts=alerts) # type: ignore # Check if any incidents should be imported according to last run time timestamp if incidents: demisto.debug(f"Adding {len(incidents)} to total alerts to import") alerts_to_import.extend(incidents) demisto.debug(f"Total alerts currently in list: {len(alerts_to_import)}") else: demisto.debug("No incidents will be imported in this iteration.") demisto.debug(f"Fetch iteration {fetch_iteration} finished") demisto.debug(f"Total alerts to import: {len(alerts_to_import)}") demisto.debug(alerts_to_import) demisto.incidents(alerts_to_import) # Return no incidents if API returned no alerts else: demisto.debug("API returned no alerts. Returning empty incident list") demisto.incidents([]) def alert_rules_list_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Retrieve alert subscriptions for the user See https://securityscorecard.readme.io/reference/subscriptions (not available right now) Args: client (SecurityScorecardClient): SecurityScorecard client args (Dict[str, Any]): Dictionary of arguments specified in the command Returns: CommandResults: The results of the command. """ response = client.get_subscriptions() demisto.debug(f"Response received: {response}") entries = response.get("entries") alert_rules: list[dict[str, str]] = [] if entries: for entry in entries: target = entry.get("delivery", {}).get("workflow", {}).get("filters", {}).get("scorecards", {}).get("value", "N/A") if target == "by_id": target = "single scorecard" elif target == "in_portfolio": portfolio_id = ( entry.get("delivery", {}) .get("workflow", {}) .get("filters", {}) .get("scorecards", {}) .get("portfolio_id", {}) .get("value", "N/A") ) target = f"portfolio with id {portfolio_id}" elif target == "followed": target = "all followed scorecards" elif target == "my_scorecard": target = "my scorecard" content: dict[str, str] = { "Alert Rule ID": entry.get("id"), "Target": target, "Name": entry.get("delivery", {}).get("workflow", {}).get("name", "N/A"), "Updated At": entry.get("updated_at", "N/A"), "Paused At": entry.get("paused_at", "N/A"), } alert_rules.append(content) markdown = tableToMarkdown("Alert Rules", alert_rules) results = CommandResults( outputs_prefix="SecurityScorecard.AlertRules.Rule", outputs_key_field="id", readable_output=markdown, outputs=alert_rules, raw_response=response, ) return results def issue_details_get_command(client: SecurityScorecardClient, args: dict[str, Any]) -> CommandResults: """Retrieve issue details for a specific issue type and domain. See https://securityscorecard.readme.io/reference/get_companies-scorecard-identifier-issues-active-cve-exploitation-attempted Args: ``client`` (``SecurityScorecardClient``): SecurityScorecard client ``args`` (``Dict[str, Any]``): The domain and issue type Returns: ``CommandResults``: The results of the command. """ domain = args.get("domain") issue_type = args.get("issue_type") if not issue_type or not domain: raise ValueError("Both 'issue_type' and 'domain' arguments are required and cannot be None.") response = client.get_company_issue_findings(domain=domain, issue_type=issue_type) entries = response.get("entries", []) findings = [] for entry in entries: # some issue types have domains, IPs and/or ports, but not all of them do if "domain" in entry: domain = entry.get("domain") elif "target" in entry: domain = entry.get("target") else: domain = "" if "ip" in entry: ip = entry.get("ip") elif "src_ip" in entry: ip = entry.get("src_ip") elif "ip_address" in entry: ip = entry.get("ip_address") elif "connection_attributes" in entry: ip = entry.get("connection_attributes", {}).get("dst_ip") else: ip = "" if "protocol" in entry: protocol = entry.get("protocol") elif "scheme" in entry: protocol = entry.get("scheme") elif "connection_attributes" in entry: protocol = entry.get("connection_attributes", {}).get("protocol") else: protocol = "" if "port" in entry: port = entry.get("port") elif "connection_attributes" in entry: port = entry.get("connection_attributes", {}).get("dst_port") else: port = "" finding = { "id": entry.get("issue_id"), "parent_domain": entry.get("parent_domain"), "count": entry.get("count"), "status": entry.get("group_status"), "first_seen": entry.get("first_seen_time"), "last_seen": entry.get("last_seen_time"), # the following details may or may not be populated "port": port, "domain_name": domain, "ip_address": ip, "protocol": protocol, "observations": entry.get("observations"), "issue_type": issue_type, } findings.append(finding) if not findings: return CommandResults( readable_output=f"No findings were found for domain {domain} and issue type {issue_type}.", outputs_prefix="SecurityScorecard.IssueDetails", outputs_key_field="issue_id", outputs=[], ) markdown = tableToMarkdown( f"Domain {domain} -- Findings for {issue_type}", findings, headers=[ "id", "parent_domain", "issue_type", "count", "status", "first_seen", "last_seen", "port", "domain_name", "ip_address", "protocol", "observations", ], ) results = CommandResults( readable_output=markdown, outputs_prefix="SecurityScorecard.IssueDetails", outputs=entries, raw_response=response, outputs_key_field="issue_id", ) return results """ MAIN FUNCTION """ def main() -> None: """main function, parses params and runs command functions Args: None Returns: None """ params = demisto.params() # Credentials api_key = params.get("username").get("password") username = params.get("username").get("identifier") # SecurityScorecard API URL base_url = params.get("base_url", "https://api.securityscorecard.io/") # Default configuration verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) # Fetch configuration max_fetch = params.get("max_fetch") incident_fetch_interval = params.get("incidentFetchInterval") args: dict[str, str] = demisto.args() demisto.debug(f"Command being called is {demisto.command()}") try: headers: dict = { "Authorization": f"Token {api_key}", "X-SSC-Application-Name": "Cortex XSOAR", "X-SSC-Application-Version": "1.0.8", } # noqa: E501 client = SecurityScorecardClient( base_url=base_url, verify=verify_certificate, headers=headers, proxy=proxy, api_key=api_key, username=username, max_fetch=max_fetch, ) if demisto.command() == "test-module": return_results(test_module(client=client, incident_fetch_interval=incident_fetch_interval)) elif demisto.command() == "fetch-incidents": fetch_alerts(client=client) elif demisto.command() == "securityscorecard-portfolios-list": return_results(portfolios_list_command(client=client, args=args)) elif demisto.command() == "securityscorecard-portfolio-list-companies": return_results(portfolio_list_companies_command(client=client, args=args)) elif demisto.command() == "securityscorecard-company-score-get": return_results(company_score_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-company-factor-score-get": return_results(company_factor_score_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-company-history-score-get": return_results(company_history_score_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-company-events-get": return_results(company_events_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-company-findings-get": return_results(company_event_findings_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-company-history-factor-score-get": return_results(company_history_factor_score_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-alert-grade-change-create": return_results(alert_grade_change_create_command(client=client, args=args)) elif demisto.command() == "securityscorecard-alert-score-threshold-create": return_results(alert_score_threshold_create_command(client=client, args=args)) elif demisto.command() == "securityscorecard-alert-delete": return_results(alert_delete_command(client=client, args=args)) elif demisto.command() == "securityscorecard-alerts-list": return_results(alerts_list_command(client=client, args=args)) elif demisto.command() == "securityscorecard-company-services-get": return_results(company_services_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-issue-metadata": return_results(issue_metadata_get_command(client=client, args=args)) elif demisto.command() == "securityscorecard-alert-rules-list": return_results(alert_rules_list_command(client=client, args=args)) elif demisto.command() == "securityscorecard-issue-details-get": return_results(issue_details_get_command(client=client, args=args)) # Log exceptions and return errors except Exception as e: demisto.error(traceback.format_exc()) # print the traceback return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e!s}") """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()