CyrenThreatInDepth

Threat InDepth's actionable and contextualized intelligence helps enterprises improve their threat detection and response by providing unprecedented visibility into new email-borne security threats faster than other security vendors.

Data Enrichment & Threat Intelligence · Cyren Threat InDepth Threat Intelligence · Feed

Details

ID	`CyrenThreatInDepth`
Provider	Communitake
Category	Data Enrichment & Threat Intelligence
From Version	6.0.0
Docker Image	`demisto/python3:3.12.13.10116658`
Supported Modules	Agentix XSIAM

README

Threat InDepth’s correlated and contextualized intelligence helps enterprises improve their threat detection and response by providing unprecedented visibility into new email-borne security threats and actionable insights to make meaningful response decisions. By correlating insights gathered across email content, web traffic, and suspicious files; Cyren provides security teams with a multi-dimensional presentation of critical threat characteristics.

Benefits include

Access to Cyren’s GlobalView™ Threat Intelligence Cloud that provides the earliest visibility into new and evolving attacks on a global basis
Comprehensive, multi-dimensional presentation of critical threat characteristics to help analysts understand the evolving threat landscape
Timely, Correlated, & Contextualized intelligence that helps reduce mean-time-to-detect (MTTD) and mean-time-to-respond (MTTR) for security analysts
Improved threat detection for existing security products such as SIEM and SOAR solutions

Feeds included in Cyren Threat InDepth content pack

The Cyren Threat InDepth content pack includes access to these streams of indicators:

IP Reputation Intelligence
Phishing & Fraud URL Intelligence
Malware URL Intelligence
Malware File Intelligence

Configure Cyren Threat InDepth Threat Intelligence Feed In Cortex

Parameter	Description	Required
apikey	API JWT token that has been issued to you	True
feed_name	Name of the particular feed that matches your API JWT token	True
max_indicators	The maximum number of indicators to fetch	False
feed	Fetch indicators.	False
feedIncremental	Is incremental or not	False
feedReputation	The reputation to apply to the fetched indicators.	False
feedReliability	The reliability of the this feed.	True
tlp_color	The Traffic Light Protocol (TLP) designation to apply to indicators fetched from the feed. More information about the protocol can be found at https://us-cert.cisa.gov/tlp	False
feedExpirationPolicy		False
feedExpirationInterval		False
feedFetchInterval	Feed Fetch Interval	False
feedBypassExclusionList	Bypass exclusion list	False

The underlying Cyren Threat InDepth API provides you with an incremental feed, meaning it provides new
or modified indicators. It also works with an offset value that keeps track of your currently processed
indicators. Your current offset defaults at the globally known maximum offset on your first setup and
is being stored and updated for you in the integration instance context. The integration then uses the
“Maximum number of indicators” parameter as the count in each request. It is recommended to set it to
a high enough value so that you get all the feed indicators for maximum product value, to handle bursts
etc. (the value cannot be higher than 100.000 and it will be capped at that value if you set a higher one).

In case you want to want to reset the offset value, use the cyren-threat-indepth-reset-client-offset command.

You can retrieve the current offset value using the cyren-threat-indepth-get-client-offset command.

Commands

You can execute these commands from the XSOAR CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

Fetch indicators

Fetching Cyren Threat InDepth indicators

Required Permissions

A valid API JWT token and a matching feed name

Base Command

cyren-threat-indepth-get-indicators

Input

Argument Name	Description	Required
max_indicators	The maximum number of results to return.	True

Context Output

There is no context output for this command.

Command Example

!cyren-threat-indepth-get-indicators max_indicators=2

Human Readable Output

Indicators from Cyren Threat InDepth:

value	type	rawJSON	score
http://nu4vs0m.u5jkzm4r.i2wd30t.bpbp9c7d.b7ni2cio.auz8x15h.freshoff.eu	URL	payload: {"action": "+", "type": "url", "identifier": "f59ef036-a790-5193-b942-24a8618c936a", "first_seen": "2020-10-25T13:41:36.000Z", "last_seen": "2021-01-05T13:54:41.000Z", "detection": {"category": ["phishing"], "detection_ts": "2020-10-25T13:41:36.000Z"}, "meta": {"port": 80, "protocol": "http"}, "relationships": [{"relationship_type": "resolves to", "relationship_ts": "2020-10-25T13:41:36.000Z", "ip": "217.70.142.108", "related_entity_category": "phishing", "relationship_description": "resolves to phishing ip"}], "detection_methods": ["URL Categorization"], "url": "http://nu4vs0m.u5jkzm4r.i2wd30t.bpbp9c7d.b7ni2cio.auz8x15h.freshoff.eu"} offset: 57006380 timestamp: 2021-01-05T14:00:48.919Z	3

Reset Client Offset

This command allows you to update the stored client offset for the feed API.

Required Permissions

A valid API JWT token and a matching feed name

Base Command

cyren-threat-indepth-reset-client-offset

Input

Argument Name	Description	Required
offset	Optional The offset you want to use as your baseline for future fetches (if not provided, the global max offset from the API is used)	False

Context Output

There is no context output for this command.

Command Example

!cyren-threat-indepth-reset-client-offset
!cyren-threat-indepth-reset-client-offset offset = 34234234

Human Readable Output

Reset Cyren Threat InDepth ip_reputation feed client offset to 1000 (API provided max offset of 1000, was 500).

Get Client Offset

This command allows you to retrieve the stored client offset for the feed API.

Required Permissions

A valid API JWT token and a matching feed name

Base Command

cyren-threat-indepth-get-client-offset

Input

There is not input for this command.

Context Output

There is no context output for this command.

Command Example

!cyren-threat-indepth-get-client-offset

Human Readable Output

Cyren Threat InDepth ip_reputation feed client offset is 500 (API provided max offset of 1000).

Additional Information

Configuration parameters

feedReputation — Indicator Reputation
feedReliability — Source Reliability (required)
feedExpirationPolicy —
feedExpirationInterval —
feedIncremental — Incremental Feed
feedFetchInterval — Feed Fetch Interval
feedBypassExclusionList — Bypass exclusion list
tlp_color — Traffic Light Protocol Color
feedTags — Tags
feed — Fetch indicators
apikey — API Token (required)
feed_name — Feed Name (required)
max_indicators — Maximum number of indicators
insecure — Trust any certificate (not secure)
proxy — Use system proxy settings

Commands (3)

cyren-threat-indepth-get-client-offset
Prints the max client offset stored in the integration context.
cyren-threat-indepth-get-indicators
Fetching Cyren Threat InDepth indicators
cyren-threat-indepth-reset-client-offset
Resets the max client offset stored in the integration context.

Source

CyrenThreatInDepth.py CyrenThreatInDepth_test.py CyrenThreatInDepth.yml CyrenThreatInDepth_description.md

Download History Raw

import json
import traceback
from collections.abc import Callable
from enum import Enum

import demistomock as demisto
import requests
import urllib3
from CommonServerPython import *

from CommonServerUserPython import *

# Disable insecure warnings
urllib3.disable_warnings()


MAX_API_COUNT: int = 100000
BASE_URL = "https://api-feeds.cyren.com/v1/feed"
VERSION = "1.5.0"
BAD_IP_RISK_BOUNDARY = 80


class FeedPath(str, Enum):
    DATA = "data"
    INFO = "info"


class FeedName(str, Enum):
    IP_REPUTATION = "ip_reputation"
    PHISHING_URLS = "phishing_urls"
    MALWARE_URLS = "malware_urls"
    MALWARE_FILES = "malware_files"


class RelationshipIndicatorType(str, Enum):
    IP = "IP"
    URL = "URL"
    SHA256 = "SHA-256"
    UNKNOWN = "UNKNOWN"


class FeedAction(str, Enum):
    ADD = "+"
    UPDATE = "="
    REMOVE = "-"

    def get_human_readable_name(self):
        name = str(self.name)
        return name.lower()


class FeedCategory(str, Enum):
    SPAM = "spam"
    PHISHING = "phishing"
    MALWARE = "malware"
    CONFIRMED_CLEAN = "confirmed clean"


class FeedSource(str, Enum):
    PRIMARY = "primary"
    RELATED = "related"


def get_relationship_value_type(relationship: dict) -> tuple[RelationshipIndicatorType, str, str]:
    related_entity_type = relationship.get("related_entity_type", "")
    relationship_value = relationship.get("related_entity_identifier", "")
    if related_entity_type == "file":
        return RelationshipIndicatorType.SHA256, relationship_value, FeedIndicatorType.File
    elif related_entity_type == "ip":
        return RelationshipIndicatorType.IP, relationship_value, FeedIndicatorType.IP

    return RelationshipIndicatorType.UNKNOWN, "", ""


class FeedEntryBase:
    def __init__(self, entry: dict, feed_name: str):
        self.entry = entry
        self.feed_name = feed_name
        self.payload = self.entry.get("payload", {})
        self.meta = self.payload.get("meta", {})
        self.detection = self.payload.get("detection", {})
        self.categories = self.detection.get("category", [])
        self.action = FeedAction(self.payload.get("action"))
        self.relationships = self.payload.get("relationships", [])

    def to_indicator_objects(self) -> list[dict]:
        fields = self.get_fields()
        if any(self.relationships):
            relationship_indicators = []
            for relationship in self.relationships:
                relationship_type, relationship_value, _ = get_relationship_value_type(relationship)
                if not relationship_value:
                    continue

                relationship_indicators.append(
                    {
                        "indicatortype": relationship_type.value,
                        "relationshiptype": relationship.get("relationship_type", ""),
                        "timestamp": relationship.get("relationship_ts"),
                        "description": relationship.get("relationship_description", ""),
                        "value": relationship_value,
                        "entitycategory": relationship.get("related_entity_category", ""),
                    }
                )
            fields["cyrenfeedrelationships"] = relationship_indicators

        raw_json = self.entry.copy()
        raw_json["source_tag"] = FeedSource.PRIMARY
        raw_json["tags"] = self.get_tags()
        primary = {
            "value": self.get_value(),
            "type": self.get_type(),
            "rawJSON": raw_json,
            "score": self.get_score(),
            "fields": fields,
        }

        indicators = self.get_indicators_from_relationships(primary)
        indicators.append(primary)
        return indicators

    def get_indicators_from_relationships(self, primary_indicator: dict) -> list[dict]:
        indicators = []
        for relationship in self.relationships:
            relationship_type, relationship_value, indicator_type = get_relationship_value_type(relationship)
            if not relationship_value:
                continue

            fields = {
                "cyrenfeedrelationships": [
                    {
                        "indicatortype": self.get_relationship_indicator_type().value,
                        "relationshiptype": relationship.get("relationship_type"),
                        "timestamp": relationship.get("relationship_ts"),
                        "value": primary_indicator["value"],
                        "entitycategory": relationship.get("related_entity_category"),
                        "description": relationship.get("relationship_description"),
                    }
                ]
            }

            raw_json = {"payload": relationship, "source_tag": FeedSource.RELATED}
            indicators.append(
                {
                    "value": relationship_value,
                    "type": indicator_type,
                    "rawJSON": raw_json,
                    "score": self.get_relationship_score(primary_indicator, relationship),
                    "fields": fields,
                }
            )

        return indicators

    def get_score(self) -> int:
        if self.action in [FeedAction.ADD, FeedAction.UPDATE]:
            if FeedCategory.CONFIRMED_CLEAN in self.categories:
                return Common.DBotScore.NONE
            else:
                return Common.DBotScore.BAD

        if self.action == FeedAction.REMOVE:
            return Common.DBotScore.NONE

        return Common.DBotScore.BAD

    def get_relationship_score(self, primary_indicator: dict, relationship: dict) -> int:
        return Common.DBotScore.NONE

    def get_tags(self) -> list:
        detection_methods = self.payload.get("detection_methods", [])
        return self.categories + detection_methods

    def get_fields(self) -> dict:
        timestamp = self.entry.get("timestamp")
        fields = {"updateddate": timestamp, "indicatoridentification": self.payload.get("identifier")}

        if self.action == FeedAction.ADD:
            fields["published"] = timestamp

        return fields

    def get_offset(self) -> int:
        offset = self.entry.get("offset")
        if not offset:
            raise ValueError("expected an 'offset' field")

        return offset

    def get_type(self) -> str:
        raise NotImplementedError

    def get_value(self) -> str:
        raise NotImplementedError

    def get_relationship_indicator_type(self) -> RelationshipIndicatorType:
        raise NotImplementedError


class UrlFeedEntry(FeedEntryBase):
    def get_type(self) -> str:
        return FeedIndicatorType.URL

    def get_value(self) -> str:
        value = self.payload.get("url", "")
        value = value.rstrip("\n").rstrip("/")
        return value

    def get_tags(self) -> list:
        industries = self.detection.get("industry", [])
        brands = self.detection.get("brand", [])
        return super().get_tags() + industries + brands

    def get_relationship_indicator_type(self) -> RelationshipIndicatorType:
        return RelationshipIndicatorType.URL


class MalwareUrlFeedEntry(UrlFeedEntry):
    def get_relationship_score(self, primary_indicator: dict, relationship: dict) -> int:
        if primary_indicator["score"] < 2 or relationship.get("related_entity_type") != "file":
            return super().get_relationship_score(primary_indicator, relationship)

        return primary_indicator["score"]


class IpReputationFeedEntry(FeedEntryBase):
    def get_type(self) -> str:
        return FeedIndicatorType.IP

    def get_value(self) -> str:
        return self.payload.get("identifier")

    def get_score(self) -> int:
        if self.action in [FeedAction.ADD, FeedAction.UPDATE]:
            if FeedCategory.SPAM in self.categories:
                risk = self.detection.get("risk", 0)
                if risk >= BAD_IP_RISK_BOUNDARY:
                    return Common.DBotScore.BAD
                return Common.DBotScore.SUSPICIOUS
            if FeedCategory.PHISHING in self.categories or FeedCategory.MALWARE in self.categories:
                return Common.DBotScore.SUSPICIOUS
            if FeedCategory.CONFIRMED_CLEAN in self.categories:
                return Common.DBotScore.NONE

        if self.action == FeedAction.REMOVE:
            return Common.DBotScore.NONE

        return Common.DBotScore.BAD

    def get_relationship_indicator_type(self) -> RelationshipIndicatorType:
        return RelationshipIndicatorType.IP


class MalwareFileFeedEntry(FeedEntryBase):
    def get_type(self) -> str:
        return FeedIndicatorType.File

    def get_value(self) -> str:
        return self.payload.get("identifier")

    def get_tags(self) -> list:
        family_names = self.detection.get("family_name", [])
        return super().get_tags() + family_names

    def get_relationship_indicator_type(self) -> RelationshipIndicatorType:
        return RelationshipIndicatorType.SHA256


FEED_TO_ENTRY_CLASS: dict[str, Callable] = {
    FeedName.IP_REPUTATION: IpReputationFeedEntry,
    FeedName.PHISHING_URLS: UrlFeedEntry,
    FeedName.MALWARE_URLS: MalwareUrlFeedEntry,
    FeedName.MALWARE_FILES: MalwareFileFeedEntry,
}


FEED_TO_VERSION: dict[str, str] = {
    FeedName.IP_REPUTATION: "_v2",
    FeedName.PHISHING_URLS: "_v2",
    FeedName.MALWARE_URLS: "_v2",
    FeedName.MALWARE_FILES: "_v2",
}


FEED_OPTION_TO_FEED: dict[str, str] = {
    "IP Reputation": FeedName.IP_REPUTATION,
    "Phishing URLs": FeedName.PHISHING_URLS,
    "Malware URLs": FeedName.MALWARE_URLS,
    "Malware Hashes": FeedName.MALWARE_FILES,
}


class InvalidAPITokenException(Exception):
    pass


class InvalidAPIUrlException(Exception):
    pass


class Client(BaseClient):
    PARAMS = {"format": "jsonl"}
    INVALID_TOKEN_TEXTS = [
        "unable to parse claims from token",
        "unable to parse token from header",
        "token was revoked",
        "claims are invalid",
    ]

    def __init__(self, feed_name: str, api_token: str, *args, **kwargs):
        if not feed_name:
            raise ValueError("please specify a correct feed name")

        super().__init__(*args, **kwargs)
        self.feed_name = feed_name
        self.request_headers = {
            "Authorization": f"Bearer {api_token}",
            "Cyren-Client-Name": "Palo Alto Cortex XSOAR",
            "Cyren-Client-Version": VERSION,
        }

    def _is_invalid_token(self, response: requests.Response) -> bool:
        if response.status_code != 400:
            return False

        return any(invalid_text in response.text for invalid_text in self.INVALID_TOKEN_TEXTS)

    def _do_request(self, path: str, offset: int = -1, count: int = 0) -> requests.Response:
        params = self.PARAMS.copy()
        params["feedId"] = f"{self.feed_name}{FEED_TO_VERSION[self.feed_name]}"
        if offset > -1:
            params["offset"] = str(offset)
        if count > 0:
            params["count"] = str(count)

        demisto.debug(f"using path {path}, params {params} for request")

        try:
            response = self._http_request(
                method="GET",
                url_suffix=path,
                headers=self.request_headers,
                params=params,
                resp_type="",
                ok_codes=[200, 204, 201, 400, 404],
            )
        except requests.ConnectionError as e:
            raise requests.ConnectionError(f"Failed to establish a new connection: {e!s}")

        if self._is_invalid_token(response):
            raise InvalidAPITokenException

        if response.status_code == 404:
            raise InvalidAPIUrlException

        response.raise_for_status()

        return response

    def fetch_entries(self, offset: int, count: int) -> list[dict]:
        result = []
        response = self._do_request(path=FeedPath.DATA, offset=offset, count=count)

        for json_line in response.content.splitlines():
            try:
                result.append(json.loads(json_line))
            except json.JSONDecodeError as je:
                demisto.info(f"error while parsing single JSON feed line: {je!s} {json_line!r}")
                continue

        return result

    def get_offsets(self) -> dict[str, int]:
        response = self._do_request(path=FeedPath.INFO)
        try:
            return json.loads(response.content)
        except Exception as e:
            demisto.error(f"error while parsing JSON: {e!s} {response.content!r}")
            raise


def store_offset_in_context(offset: int) -> int:
    integration_context = get_integration_context()
    integration_context["offset"] = offset
    set_integration_context(integration_context)
    return offset


def get_offset_from_context() -> int:
    integration_context = get_integration_context()
    return integration_context.get("offset")


def test_module_command(client: Client) -> str:
    try:
        entries = client.fetch_entries(0, 10)
        if not any(entries):
            return "Test failed because no indicators could be fetched!"
    except InvalidAPITokenException:
        return "Test failed because of an invalid API token!"
    except InvalidAPIUrlException:
        return "Test failed because of an invalid API URL!"
    except Exception as e:
        return f"Test failed because of: {e!s}!"

    return "ok"


def get_indicators_command(client: Client, args: dict) -> CommandResults:
    max_indicators = int(args.get("max_indicators", 50))

    count = max_indicators
    end_offset = client.get_offsets().get("endOffset")
    if end_offset:
        offset = max(end_offset - max_indicators + 1, 0)
    else:
        offset = 0

    entries = client.fetch_entries(offset, count)
    indicators, _ = feed_entries_to_indicator(entries, client.feed_name)

    human_readable = tableToMarkdown(
        "Indicators from Cyren Threat InDepth:", indicators, headers=["value", "type", "rawJSON", "score"]
    )
    return CommandResults(readable_output=human_readable, raw_response=indicators)


def reset_offset_command(client: Client, args: dict) -> CommandResults:
    offset = int(args.get("offset", -1))
    offset_stored = get_offset_from_context()
    offset_api = int(client.get_offsets().get("endOffset", -1))
    if offset < 0 or offset > offset_api:
        offset = offset_api

    store_offset_in_context(offset)

    offset_stored_text = offset_stored or "not set before"
    readable_output = (
        f"Reset Cyren Threat InDepth {client.feed_name} feed client offset to {offset} "
        f"(API provided max offset of {offset_api}, was {offset_stored_text})."
    )
    return CommandResults(readable_output=readable_output, raw_response=offset)


def get_offset_command(client: Client, args: dict) -> CommandResults:
    offset_stored = get_offset_from_context()
    offset_api = int(client.get_offsets().get("endOffset", -1))
    offset_api_text = f"(API provided max offset of {offset_api})"
    if offset_stored:
        readable_output = f"Cyren Threat InDepth {client.feed_name} feed client offset is {offset_stored} {offset_api_text}."
    else:
        readable_output = f"Cyren Threat InDepth {client.feed_name} feed client offset has not been set yet {offset_api_text}."
    return CommandResults(readable_output=readable_output, raw_response=offset_stored)


def feed_entries_to_indicator(entries: list[dict], feed_name: str) -> tuple[list[dict], int]:
    indicators: list[dict] = []
    max_offset: int = -1
    for entry in entries:
        entry_obj = FEED_TO_ENTRY_CLASS[feed_name](entry, feed_name)
        max_offset = max(entry_obj.get_offset() + 1, max_offset)
        indicators = indicators + entry_obj.to_indicator_objects()

    return indicators, max_offset


def fetch_indicators_command(client: Client, initial_count: int, max_indicators: int, update_context: bool) -> list[dict]:
    offset = get_offset_from_context()
    count = max_indicators
    if not offset:
        offset = int(client.get_offsets().get("endOffset", -1))
        if initial_count > 0:
            offset = offset - initial_count + 1
            count = max_indicators + initial_count

    count = min(MAX_API_COUNT, count)

    entries = client.fetch_entries(offset, count)
    demisto.debug(f"pulled {len(entries)} for {client.feed_name}")
    indicators, max_offset = feed_entries_to_indicator(entries, client.feed_name)
    demisto.debug(f"about to ingest {len(indicators)} for {client.feed_name}")

    if update_context:
        store_offset_in_context(max_offset)

    return indicators


def main():
    params = demisto.params()
    api_token = params.get("apikey")

    feed_name = params.get("feed_name")
    feed_name = FEED_OPTION_TO_FEED.get(feed_name, "")

    max_indicators = int(params.get("max_indicators", MAX_API_COUNT))
    if max_indicators > MAX_API_COUNT:
        demisto.info(f"using a maximum value for max_indicators of {MAX_API_COUNT} instead of {max_indicators}!")
        max_indicators = MAX_API_COUNT

    proxy = params.get("proxy", False)
    verify_certificate = not params.get("insecure", False)

    demisto.info(f"using feed {feed_name}, max {max_indicators}")
    commands: dict[str, Callable] = {
        "cyren-threat-indepth-get-indicators": get_indicators_command,
        "cyren-threat-indepth-reset-client-offset": reset_offset_command,
        "cyren-threat-indepth-get-client-offset": get_offset_command,
    }

    command = demisto.command()
    demisto.info(f"Command being called is {command}")

    error = None
    try:
        client = Client(feed_name=feed_name, base_url=BASE_URL, verify=verify_certificate, api_token=api_token, proxy=proxy)

        if command == "fetch-indicators":
            indicators = fetch_indicators_command(
                client=client, initial_count=0, max_indicators=max_indicators, update_context=True
            )
            for b in batch(indicators, batch_size=2000):
                demisto.createIndicators(b)
        elif command == "test-module":
            return_outputs(test_module_command(client))
        else:
            return_results(commands[command](client, demisto.args()))
    except InvalidAPITokenException:
        error = "Invalid API token!"
    except InvalidAPIUrlException:
        error = "Invalid API URL!"
    except Exception as e:
        demisto.error(traceback.format_exc())
        error = f"Error failed to execute {command}, error: [{e}]"

    if error:
        return_error(error)


if __name__ == "__builtin__" or __name__ == "builtins":
    main()