Unit 42 Feed

from CommonServerPython import *
from CommonServerUserPython import *
import urllib3

# Disable insecure warnings
urllib3.disable_warnings()

DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
INTEGRATION_NAME = "Unit 42 Feed"
API_LIMIT = 5000
TOTAL_INDICATOR_LIMIT = 100000

# Priority order for fetching Threat Objects first Files last
# Note: Threat Objects are handled separately in feed_types
INDICATOR_TYPE_PRIORITY = ["IP", "Domain", "URL", "File"]

# API endpoints
BASE_URL = "https://prod-us.tas.crtx.paloaltonetworks.com"
INDICATORS_ENDPOINT = "/api/v1/feeds/indicators"
THREAT_OBJECTS_ENDPOINT = "/api/v1/feeds/threat_objects"

# Retry configuration
RETRY_COUNT = 5
STATUS_CODES_TO_RETRY = list(range(429, 600))

# Mapping from API indicator types to XSOAR indicator types
INDICATOR_TYPE_MAPPING = {
    "ip": FeedIndicatorType.IP,
    "domain": FeedIndicatorType.Domain,
    "url": FeedIndicatorType.URL,
    "file": FeedIndicatorType.File,
    "filehash_sha256": FeedIndicatorType.File,
    "exploit": FeedIndicatorType.CVE,
    "vulnerability": FeedIndicatorType.CVE,
    "malware_family": ThreatIntel.ObjectsNames.MALWARE,
    "grayware": ThreatIntel.ObjectsNames.MALWARE,
    "actor": ThreatIntel.ObjectsNames.THREAT_ACTOR,
    "threat_actor": ThreatIntel.ObjectsNames.THREAT_ACTOR,
    "campaign": ThreatIntel.ObjectsNames.CAMPAIGN,
    "attack pattern": ThreatIntel.ObjectsNames.ATTACK_PATTERN,
    "attack_pattern": ThreatIntel.ObjectsNames.ATTACK_PATTERN,
    "technique": ThreatIntel.ObjectsNames.ATTACK_PATTERN,
    "malicious_behavior": ThreatIntel.ObjectsNames.ATTACK_PATTERN,
    "malicious behavior": ThreatIntel.ObjectsNames.ATTACK_PATTERN,
    "malicious_tool": ThreatIntel.ObjectsNames.TOOL,
}

VERDICT_TO_SCORE = {
    "malicious": Common.DBotScore.BAD,
    "suspicious": Common.DBotScore.SUSPICIOUS,
    "benign": Common.DBotScore.GOOD,
    "unknown": Common.DBotScore.NONE,
}

# Define valid regions enum
VALID_REGIONS = {
    "australia and oceania": "Australia And Oceania",
    "antarctica": "Antarctica",
    "north america": "North America",
    "south asia": "South Asia",
    "europe": "Europe",
    "central america and the caribbean": "Central America And The Caribbean",
    "africa": "Africa",
    "east and southeast asia": "East And Southeast Asia",
    "middle east": "Middle East",
    "central asia": "Central Asia",
    "south america": "South America",
}


def unit42_error_handler(res: requests.Response):
    """
    Custom error handler for Unit 42 API requests.
    Extracts and logs X-Request-ID header for failed requests (4xx/5xx errors).

    Args:
        res: Response object from failed request

    Returns:
        Error message string including X-Request-ID if available
    """
    request_id = res.headers.get("X-Request-ID", "N/A")
    demisto.debug(f"{INTEGRATION_NAME} API Error - X-Request-ID: {request_id}, Status: {res.status_code}, URL: {res.url}")

    error_msg = f"Error in API request [Status: {res.status_code}]\n"
    error_msg += f"[X-Request-ID: {request_id}]\n"
    error_msg += f"Response text - {res.text}"

    return_error(error_msg)


class Client(BaseClient):
    def __init__(self, headers, verify=False, proxy=False):
        """Implements class for Unit 42 feed.

        Args:
            headers: headers for the request.
            verify: boolean, if *false*, feed HTTPS server certificate is verified. Default: *false*
            proxy: boolean, if *false* feed HTTPS server certificate will not use proxies. Default: *false*
        """
        super().__init__(base_url=BASE_URL, headers=headers, verify=verify, proxy=proxy)

    def get_indicators(
        self,
        indicator_types: list | None = None,
        limit: int = API_LIMIT,
        start_time: str | None = None,
        next_page_token: str | None = None,
    ) -> dict:
        """Get indicators from the Unit 42 feed.

        Args:
            indicator_types: A list of indicator types to fetch (ip, filehash_sha256, domain, url)
            limit: Maximum number of indicators to return
            start_time: Start time for fetching indicators
            next_page_token: Token for pagination

        Returns:
            Dict containing indicators and pagination info
        """
        params: dict[str, Any] = {}
        if indicator_types:
            params["indicator_types"] = [i.lower().replace("file", "filehash_sha256") for i in indicator_types]
        if limit:
            params["limit"] = limit
        if start_time:
            params["start_time"] = start_time
        if next_page_token:
            params["page_token"] = next_page_token

        response = self._http_request(
            method="GET",
            url_suffix=INDICATORS_ENDPOINT,
            params=params,
            error_handler=unit42_error_handler,
            retries=RETRY_COUNT,
            status_list_to_retry=STATUS_CODES_TO_RETRY,
        )
        return response

    def get_threat_objects(self, limit: int = API_LIMIT, next_page_token: str | None = None) -> dict:
        """Get threat objects from the Unit 42 feed.

        Args:
            limit: Maximum number of threat objects to return
            next_page_token: Token for pagination

        Returns:
            Dict containing threat objects and pagination info
        """
        params: dict[str, Any] = {}
        if limit:
            params["limit"] = limit
        if next_page_token:
            params["page_token"] = next_page_token

        response = self._http_request(
            method="GET",
            url_suffix=THREAT_OBJECTS_ENDPOINT,
            params=params,
            error_handler=unit42_error_handler,
            retries=RETRY_COUNT,
            status_list_to_retry=STATUS_CODES_TO_RETRY,
        )
        return response


def create_publications(publications_data: list) -> list:
    """
    Creates the publications list of the indicator

    Args:
        publications_data: A list of all publications from threat object

    Returns:
        A list of publications of the indicator
    """
    publications = []

    for data in publications_data:
        timestamp = data.get("created_at", "")
        title = data.get("title", "")
        url = data.get("url", "")
        source = data.get("source", INTEGRATION_NAME)

        publications.append({"link": url, "title": title, "timestamp": timestamp, "source": source})

    return publications


def get_threat_object_score(threat_class: str) -> int:
    """
    Get the appropriate score for a threat object based on its class

    Args:
        threat_class: The threat object class (lowercase)

    Returns:
        Appropriate ThreatIntel score or Common.DBotScore.NONE as default
    """
    if threat_class not in INDICATOR_TYPE_MAPPING:
        return Common.DBotScore.NONE

    threat_type = INDICATOR_TYPE_MAPPING[threat_class]

    if threat_type == ThreatIntel.ObjectsNames.MALWARE:
        return ThreatIntel.ObjectsScore.MALWARE
    elif threat_type == ThreatIntel.ObjectsNames.THREAT_ACTOR:
        return ThreatIntel.ObjectsScore.THREAT_ACTOR
    elif threat_type == ThreatIntel.ObjectsNames.CAMPAIGN:
        return ThreatIntel.ObjectsScore.CAMPAIGN
    elif threat_type == ThreatIntel.ObjectsNames.ATTACK_PATTERN:
        return ThreatIntel.ObjectsScore.ATTACK_PATTERN

    return Common.DBotScore.NONE


def create_location_indicators_and_relationships(threat_obj: dict[str, Any], threat_actor_name: str) -> list[dict[str, Any]]:
    """
    Create location indicators from affected regions and origin field and build relationships

    Args:
        threat_obj: The threat object data
        threat_actor_name: Name of the threat actor to create relationships with

    Returns:
        List of location indicators with relationships
    """
    location_indicators: list[dict[str, Any]] = []

    # Handle affected regions
    affected_regions = demisto.get(threat_obj, "battlecard_details.threat_actor_details.affected_regions", [])

    # in case affected_regions is "null", return empty list.
    if not isinstance(affected_regions, list):
        return location_indicators

    for region in affected_regions:
        if isinstance(region, str) and region.strip():
            region_lower = region.strip().lower()

            # Use the standardized region name if it matches our enum
            standardized_region = VALID_REGIONS.get(region_lower)
            if not standardized_region:
                demisto.debug(f"Skipping region {region} as it is not in the valid regions enum")
                continue

            # Create EntityRelationship for the location
            entity_relationship = EntityRelationship(
                name=EntityRelationship.Relationships.TARGETS,
                entity_a=threat_actor_name,
                entity_a_type=ThreatIntel.ObjectsNames.THREAT_ACTOR,
                entity_b=standardized_region,
                entity_b_type=FeedIndicatorType.Location,
                source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                brand=INTEGRATION_NAME,
            )

            location_indicator = {
                "value": standardized_region,
                "type": FeedIndicatorType.Location,
                "score": Common.DBotScore.NONE,
                "service": INTEGRATION_NAME,
                "relationships": [entity_relationship.to_entry()],
                "fields": {
                    "geocountry": standardized_region,
                },
            }
            location_indicators.append(location_indicator)

    return location_indicators


def build_threat_object_description(threat_obj: dict[str, Any]) -> str:
    """
    Build a comprehensive description for a threat object including highlights, methods, and targets

    Args:
        threat_obj: The threat object data

    Returns:
        Formatted description string with sections for highlights, methods, and targets
    """
    description = threat_obj.get("description", "").replace("\\n", "\n")

    # Add highlights section if available
    highlights = demisto.get(threat_obj, "battlecard_details.highlights", "").replace("\\n", "\n")
    if highlights and highlights != "Highlights / Key Takeaways (external)":  # Do not add if it is only the default title
        description += "\n\n##"
        description += highlights

    # Add methods section if available (for threat actors)
    methods = demisto.get(threat_obj, "battlecard_details.threat_actor_details.methods", "").replace("\\n", "\n")
    if methods:
        description += "\n\n##"
        description += methods

    # Add targets section if available (for threat actors)
    targets = demisto.get(threat_obj, "battlecard_details.threat_actor_details.targets", "").replace("\\n", "\n")
    if targets:
        description += "\n\n##"
        description += targets

    return description


def create_vulnerabilities_relationships(threat_obj: dict[str, Any], threat_actor_name: str, threat_class: str) -> list[dict]:
    """
    Create vulnerabilities relationships from vulnerabilities associations

    Args:
        threat_obj: The threat object data
        threat_actor_name: Name of the threat actor
        threat_class: The threat object class

    Returns:
        List of EntityRelationship objects
    """
    relationships: list[dict] = []

    if threat_class not in INDICATOR_TYPE_MAPPING:
        demisto.debug(f"Skipping create_vulnerabilities_relationships for unknown threat_class {threat_class!r}")
        return relationships

    vulnerabilities = demisto.get(threat_obj, "battlecard_details.threat_actor_details.vulnerability_associations", [])

    for vulnerability in vulnerabilities:
        cve_id = vulnerability.get("cve")

        if cve_id:
            entity_relationship = EntityRelationship(
                name=EntityRelationship.Relationships.EXPLOITS,
                entity_a=threat_actor_name,
                entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                entity_b=cve_id.upper(),
                entity_b_type=FeedIndicatorType.CVE,
                source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                brand=INTEGRATION_NAME,
            )
            relationships.append(entity_relationship.to_entry())

    return relationships


def create_actor_relationships(threat_obj: dict[str, Any], malware_family_name: str, threat_class: str) -> list[dict]:
    """
    Create actor relationships from actor_associations

    Args:
        threat_obj: The threat object data
        malware_family_name: Name of the malware family
        threat_class: The threat object class

    Returns:
        List of EntityRelationship objects
    """
    relationships: list[dict] = []

    if threat_class not in INDICATOR_TYPE_MAPPING:
        demisto.debug(f"Skipping create_actor_relationships for unknown threat_class {threat_class!r}")
        return relationships

    actor_associations = demisto.get(threat_obj, "battlecard_details.malware_family_details.actor_associations", [])

    for relationship in actor_associations:
        aliases = relationship.get("aliases", [])
        name = relationship.get("name")

        if aliases:
            # Create a relationship for each alias
            for alias in aliases:
                entity_relationship = EntityRelationship(
                    name=EntityRelationship.Relationships.USED_BY,
                    entity_a=malware_family_name,
                    entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                    entity_b=string_to_table_header(alias),
                    entity_b_type=ThreatIntel.ObjectsNames.THREAT_ACTOR,
                    source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                    brand=INTEGRATION_NAME,
                )
                relationships.append(entity_relationship.to_entry())
        elif name:
            # Create a relationship using the name if no aliases exist
            entity_relationship = EntityRelationship(
                name=EntityRelationship.Relationships.USED_BY,
                entity_a=malware_family_name,
                entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                entity_b=string_to_table_header(name),
                entity_b_type=ThreatIntel.ObjectsNames.THREAT_ACTOR,
                source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                brand=INTEGRATION_NAME,
            )
            relationships.append(entity_relationship.to_entry())

    return relationships


def create_tools_relationships(threat_obj: dict[str, Any], threat_actor_name: str, threat_class: str) -> list[dict]:
    """
    Create tools relationships from tools associations

    Args:
        threat_obj: The threat object data
        threat_actor_name: Name of the threat actor
        threat_class: The threat object class

    Returns:
        List of EntityRelationship objects
    """
    relationships: list[dict] = []

    if threat_class not in INDICATOR_TYPE_MAPPING:
        demisto.debug(f"Skipping create_tools_relationships for unknown threat_class {threat_class!r}")
        return relationships

    tools_associations = demisto.get(threat_obj, "battlecard_details.threat_actor_details.tools", [])

    for tool in tools_associations:
        tool_name = tool.get("name")

        if tool_name:
            entity_relationship = EntityRelationship(
                name=EntityRelationship.Relationships.USES,
                entity_a=threat_actor_name,
                entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                entity_b=string_to_table_header(tool_name),
                entity_b_type=ThreatIntel.ObjectsNames.TOOL,
                source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                brand=INTEGRATION_NAME,
                fields={"tags": f"mitre-id: {tool.get('mitreid')}" if tool.get("mitreid") else ""},
            )
            relationships.append(entity_relationship.to_entry())

    return relationships


def create_malware_relationships(threat_obj: dict[str, Any], threat_actor_name: str, threat_class: str) -> list[dict]:
    """
    Create malware relationships from malware_associations

    Args:
        threat_obj: The threat object data
        threat_actor_name: Name of the threat actor
        threat_class: The threat object class

    Returns:
        List of EntityRelationship objects
    """
    relationships: list[dict] = []

    if threat_class not in INDICATOR_TYPE_MAPPING:
        demisto.debug(f"Skipping create_malware_relationships for unknown threat_class {threat_class!r}")
        return relationships

    malware_associations = demisto.get(threat_obj, "battlecard_details.threat_actor_details.malware_associations", [])

    for relationship in malware_associations:
        name = relationship.get("name")
        aliases = relationship.get("aliases", [])

        if name:
            # Create a relationship using the name
            entity_relationship = EntityRelationship(
                name=EntityRelationship.Relationships.USES,
                entity_a=threat_actor_name,
                entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                entity_b=string_to_table_header(name),
                entity_b_type=ThreatIntel.ObjectsNames.MALWARE,
                source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                brand=INTEGRATION_NAME,
            )
            relationships.append(entity_relationship.to_entry())
        elif aliases:
            # Create a relationship for each alias if no name exists
            for alias in aliases:
                entity_relationship = EntityRelationship(
                    name=EntityRelationship.Relationships.USES,
                    entity_a=threat_actor_name,
                    entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                    entity_b=string_to_table_header(alias),
                    entity_b_type=ThreatIntel.ObjectsNames.MALWARE,
                    source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                    brand=INTEGRATION_NAME,
                )
                relationships.append(entity_relationship.to_entry())

    return relationships


def create_attack_patterns_relationships(threat_obj: dict[str, Any], threat_actor_name: str, threat_class: str) -> list[dict]:
    """
    Create attack patterns relationships from attack patterns associations

    Args:
        threat_obj: The threat object data
        threat_actor_name: Name of the threat actor
        threat_class: The threat object class

    Returns:
        List of EntityRelationship objects
    """
    relationships: list[dict] = []

    if threat_class not in INDICATOR_TYPE_MAPPING:
        demisto.debug(f"Skipping create_attack_patterns_relationships for unknown threat_class {threat_class!r}")
        return relationships

    attack_patterns = demisto.get(threat_obj, "battlecard_details.attack_patterns", [])

    for pattern in attack_patterns:
        mitre_id = pattern.get("mitreid", "")
        pattern_name = pattern.get("name", "")

        # Skip items with a dot in the mitreid
        if "." in mitre_id:
            demisto.debug(f"Skipping attack pattern {pattern_name} with mitreid {mitre_id}")
            continue

        if pattern_name and pattern_name.endswith("(enterprise)"):
            # Remove (enterprise) suffix if present
            pattern_name = pattern_name.removesuffix("(enterprise)").strip()

            entity_relationship = EntityRelationship(
                name=EntityRelationship.Relationships.USES,
                entity_a=threat_actor_name,
                entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                entity_b=string_to_table_header(pattern_name),
                entity_b_type=ThreatIntel.ObjectsNames.ATTACK_PATTERN,
                source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                brand=INTEGRATION_NAME,
            )
            relationships.append(entity_relationship.to_entry())

    return relationships


def create_campaigns_relationships(threat_obj: dict[str, Any], threat_object_name: str, threat_class: str) -> list[dict]:
    """
    Create campaigns relationships from campaigns list

    Args:
        threat_obj: The threat object data
        threat_object_name: Name of the threat object
        threat_class: The threat object class

    Returns:
        List of EntityRelationship objects
    """
    relationships: list[dict] = []

    if threat_class not in INDICATOR_TYPE_MAPPING:
        demisto.debug(f"Skipping create_campaigns_relationships for unknown threat_class {threat_class!r}")
        return relationships

    campaigns = demisto.get(threat_obj, "battlecard_details.campaigns", [])

    for campaign in campaigns:
        if isinstance(campaign, str) and campaign.strip():
            entity_relationship = EntityRelationship(
                name=EntityRelationship.Relationships.RELATED_TO,
                entity_a=threat_object_name,
                entity_a_type=INDICATOR_TYPE_MAPPING[threat_class],
                entity_b=string_to_table_header(campaign),
                entity_b_type=ThreatIntel.ObjectsNames.CAMPAIGN,
                source_reliability=DBotScoreReliability.A_PLUS_PLUS,
                brand=INTEGRATION_NAME,
            )
            relationships.append(entity_relationship.to_entry())

    return relationships


def create_relationships_and_tags(
    indicator_value: str, indicator_type: str, threat_object_associations: list
) -> tuple[list[Any], list[str]]:
    """
    Create relationships and tags from threat object associations

    Args:
        indicator_value: The indicator value (entity_a)
        indicator_type: The indicator type for mapping
        threat_object_associations: List of threat object associations

    Returns:
        Tuple of List of EntityRelationship objects and tags
    """
    relationships: list[Any] = []
    tags: list[str] = []

    for assoc in threat_object_associations:
        if not assoc or not assoc.get("name") or not assoc.get("threat_object_class"):
            continue

        threat_name = assoc.get("name")
        threat_class = assoc.get("threat_object_class")

        tags.append(threat_name)

        if argToBoolean(demisto.params().get("create_relationships") or False):
            reliability = demisto.params().get("feedReliability", "A++ - Reputation script")

            # Map threat class to XSOAR threat intel object type
            entity_a_type = INDICATOR_TYPE_MAPPING.get(indicator_type, Common.Indicator)
            entity_b_type = INDICATOR_TYPE_MAPPING.get(threat_class, Common.Indicator)

            # Determine relationship type based on threat class
            if threat_class in ["actor", "threat_actor"]:
                relationship_name = EntityRelationship.Relationships.USED_BY
            elif threat_class == "campaign":
                relationship_name = EntityRelationship.Relationships.PART_OF
            elif threat_class in ["attack pattern", "technique"]:
                relationship_name = EntityRelationship.Relationships.USES
            elif threat_class == "exploit":
                relationship_name = EntityRelationship.Relationships.EXPLOITS
            elif threat_class in ["malicious behavior", "malicious_behavior"]:
                relationship_name = EntityRelationship.Relationships.INDICATOR_OF
            else:
                relationship_name = EntityRelationship.Relationships.RELATED_TO

            relationship = EntityRelationship(
                name=relationship_name,
                entity_a=indicator_value,
                entity_a_type=entity_a_type,
                entity_b=threat_name,
                entity_b_type=entity_b_type,
                source_reliability=reliability,
                brand=INTEGRATION_NAME,
            )

            relationships.append(relationship.to_entry())

    return relationships, tags


def map_indicator(indicator_data: dict, feed_tags: list = [], tlp_color: str | None = None) -> dict:
    """Map an indicator from the Unit 42 API to XSOAR format.

    Args:
        indicator_data: Indicator data from the API.
        feed_tags: List of tags to add to the indicator.
        tlp_color: Traffic Light Protocol color to add to the indicator.

    Returns:
        Indicator in XSOAR format.
    """
    indicator_value = indicator_data.get("indicator_value", "")
    indicator_type = indicator_data.get("indicator_type", "")

    # Map the indicator type to XSOAR type
    xsoar_indicator_type = INDICATOR_TYPE_MAPPING.get(indicator_type, Common.Indicator)

    # Create DBotScore
    verdict = str(indicator_data.get("verdict") or "")
    dbot_score = VERDICT_TO_SCORE.get(verdict, Common.DBotScore.NONE)

    # Create relationships and tags
    relationships: list[Any] = []
    tags: list[str] = []
    if indicator_data.get("threat_object_associations"):
        relationships, tags = create_relationships_and_tags(
            indicator_value, indicator_type, indicator_data.get("threat_object_associations") or []
        )

    # Create fields
    fields = {
        "updateddate": indicator_data.get("updated_at"),
        "creationdate": indicator_data.get("first_seen"),
        "reportedby": indicator_data.get("source"),
        "tags": list(set(feed_tags + tags)),
        "trafficlightprotocol": tlp_color,
    }
    if xsoar_indicator_type == FeedIndicatorType.File:
        fields["md5"] = demisto.get(indicator_data, "indicator_details.file_hashes.md5")
        fields["sha1"] = demisto.get(indicator_data, "indicator_details.file_hashes.sha1")
        fields["sha256"] = demisto.get(indicator_data, "indicator_details.file_hashes.sha256")
        fields["ssdeep"] = demisto.get(indicator_data, "indicator_details.file_hashes.ssdeep")
        fields["imphash"] = demisto.get(indicator_data, "indicator_details.file_hashes.imphash")
        fields["pehash"] = demisto.get(indicator_data, "indicator_details.file_hashes.pehash")
        fields["filetype"] = demisto.get(indicator_data, "indicator_details.file_type")
        fields["fileextension"] = demisto.get(indicator_data, "indicator_details.file_type", "").split(".")[-1]
        fields["size"] = demisto.get(indicator_data, "indicator_details.file_size")

    # Create the indicator object
    indicator: dict = {
        "value": indicator_value,
        "type": xsoar_indicator_type,
        "score": dbot_score,
        "service": INTEGRATION_NAME,
        "relationships": relationships,
        "fields": fields,
        "rawJSON": indicator_data,
    }

    return indicator


def map_threat_object(threat_object: dict, feed_tags: list = [], tlp_color: str | None = None) -> list:
    """Map a threat object from the Unit 42 API to XSOAR format.

    Args:
        threat_object: Threat object data from the API.
        feed_tags: List of tags to add to the threat object.
        tlp_color: Traffic Light Protocol color to add to the threat object.

    Returns:
        List of threat objects in XSOAR format.
    """
    result: list = []

    # Get basic threat object properties
    name = threat_object.get("name", "")
    threat_class = threat_object.get("threat_object_class", "").lower()

    # Map the threat object class to XSOAR type
    xsoar_indicator_type = INDICATOR_TYPE_MAPPING.get(str(threat_class), Common.Indicator)

    # Create relationships
    relationships, tags = create_relationships_and_tags(name, threat_class, threat_object.get("related_threat_objects", []))
    if argToBoolean(demisto.params().get("create_relationships") or False):
        relationships += create_campaigns_relationships(threat_object, name, threat_class)
        relationships += create_attack_patterns_relationships(threat_object, name, threat_class)
        relationships += create_malware_relationships(threat_object, name, threat_class)
        relationships += create_tools_relationships(threat_object, name, threat_class)
        relationships += create_vulnerabilities_relationships(threat_object, name, threat_class)
        relationships += create_actor_relationships(threat_object, name, threat_class)

        # Create location indicators and relationships
        location_indicators = create_location_indicators_and_relationships(threat_object, name)
        result.extend(location_indicators)

    fields = {
        "description": build_threat_object_description(threat_object),
        "lastseenbysource": threat_object.get("last_hit", ""),
        "reportedby": threat_object.get("sources", []),
        "aliases": [string_to_table_header(alias) for alias in threat_object.get("aliases", [])],
        "industrysectors": [
            string_to_table_header(industry) for industry in demisto.get(threat_object, "battlecard_details.industries", [])
        ],
        "primarymotivation": string_to_table_header(
            demisto.get(threat_object, "battlecard_details.threat_actor_details.primary_motivation", "")
        ),
        "publications": create_publications(threat_object.get("publications", [])),
        "geocountry": demisto.get(threat_object, "battlecard_details.threat_actor_details.origin", "").upper(),
        "tags": tags + feed_tags,
        "trafficlightprotocol": tlp_color,
        "ismalwarefamily": "True" if threat_class == "malware_family" else "False",
    }

    # Create the threat object
    result.append(
        {
            "value": name,
            "type": xsoar_indicator_type,
            "score": get_threat_object_score(threat_class),
            "service": INTEGRATION_NAME,
            "relationships": relationships,
            "fields": fields,
            "rawJSON": threat_object,
        }
    )

    return result


def parse_indicators(indicators_data: list, feed_tags: list = [], tlp_color: str | None = None) -> list:
    """Parse indicators from the Unit 42 API into XSOAR format.

    Args:
        indicators_data: List of indicators from the API.
        feed_tags: List of tags to add to the indicators.
        tlp_color: Traffic Light Protocol color to add to the indicators.

    Returns:
        List of parsed indicators in XSOAR format.
    """
    indicators = []

    if indicators_data and isinstance(indicators_data, list):
        for indicator_data in indicators_data:
            indicator = map_indicator(indicator_data, feed_tags, tlp_color)
            indicators.append(indicator)

    return indicators


def parse_threat_objects(threat_objects_data: list, feed_tags: list = [], tlp_color: str | None = None) -> list:
    """Parse threat objects from the Unit 42 API into XSOAR format.

    Args:
        threat_objects_data: List of threat objects from the API.
        feed_tags: List of tags to add to the threat objects.
        tlp_color: Traffic Light Protocol color to add to the threat objects.

    Returns:
        List of parsed threat objects in XSOAR format.
    """
    threat_objects = []

    if threat_objects_data and isinstance(threat_objects_data, list):
        for threat_object_data in threat_objects_data:
            new_threat_objects = map_threat_object(threat_object_data, feed_tags, tlp_color)
            threat_objects.extend(new_threat_objects)

    return threat_objects


def sort_indicator_types_by_priority(indicator_types: list) -> list:
    """
    Sort indicator types by priority (case-insensitive).

    Args:
        indicator_types: List of indicator types to sort

    Returns:
        Sorted list based on INDICATOR_TYPE_PRIORITY (IPs → Domains → URLs → Files)
    """
    # Create case-insensitive priority map (lower index = higher priority)
    priority_map = {indicator_type.lower(): idx for idx, indicator_type in enumerate(INDICATOR_TYPE_PRIORITY)}

    # Sort by priority using lowercase comparison (IPs=0, Domains=1, URLs=2, Files=3)
    return sorted(indicator_types, key=lambda t: priority_map.get(t.lower(), len(INDICATOR_TYPE_PRIORITY)))


def calculate_limit_per_type(limit: int | None, total_indicator_types: int) -> int:
    """
    Calculate the limit per type based on the provided limit and total indicator types.

    Algorithm:
    - If limit is None or < 0 -> use default limit (TOTAL_INDICATOR_LIMIT / total_indicator_types)
    - If limit * total_indicator_types > TOTAL_INDICATOR_LIMIT ->
        use default limit (TOTAL_INDICATOR_LIMIT / total_indicator_types)
    - Otherwise -> use the provided limit

    Args:
        limit: The requested limit per type (can be None or negative)
        total_indicator_types: Total number of indicator types (including threat objects if enabled)

    Returns:
        Calculated limit per type
    """
    # Calculate default limit per type (always TOTAL_INDICATOR_LIMIT / total_indicator_types)
    default_limit_per_type = (
        TOTAL_INDICATOR_LIMIT // total_indicator_types if total_indicator_types > 0 else TOTAL_INDICATOR_LIMIT
    )

    # If limit is None or negative, use default
    if limit is None or limit < 0:
        demisto.debug(f"UNIT42FEED: Limit is None or negative ({limit}), using default limit per type: {default_limit_per_type}")
        return default_limit_per_type

    # If limit * types exceeds total limit, use default
    total_expected = limit * total_indicator_types
    if total_expected > TOTAL_INDICATOR_LIMIT:
        demisto.debug(
            f"UNIT42FEED: Total expected ({total_expected}) exceeds maximum {TOTAL_INDICATOR_LIMIT}. "
            f"Using default limit per type: {default_limit_per_type}"
        )
        return default_limit_per_type

    # Otherwise, use the provided limit
    demisto.debug(f"UNIT42FEED: Using provided limit per type: {limit}")
    return int(limit)


def push_indicators_in_batches(indicators: list, batch_size: int = 2000) -> None:
    """
    Push indicators to the server immediately in fixed-size batches.

    This avoids holding large lists of already-fetched indicators in memory:
    each page fetched from the API is pushed to the server right away instead
    of being accumulated together with all other pages/types until the end
    of the fetch cycle.

    Args:
        indicators: List of indicators (or threat objects) to push
        batch_size: Maximum number of indicators to send in a single createIndicators call
    """
    for indicators_batch in batch(indicators, batch_size=batch_size):
        demisto.createIndicators(indicators_batch)


def fetch_indicator_type(
    client: Client, indicator_type: str, limit: int, start_time: str, feed_tags: list, tlp_color: str | None
) -> int:
    """
    Fetch indicators for a specific type with pagination and limit enforcement.

    Each fetched page (up to API_LIMIT indicators) is parsed and immediately pushed
    to the server in batches of 2000, instead of being accumulated in memory until
    the entire fetch cycle completes. This prevents out-of-memory issues when large
    numbers of indicators are fetched.

    Args:
        client: Client object
        indicator_type: Type to fetch (File, IP, URL, Domain)
        limit: Maximum number to fetch for this type
        start_time: Start time for fetching
        feed_tags: Tags to add to indicators
        tlp_color: TLP color

    Returns:
        Number of indicators fetched and pushed (count <= limit)
    """
    total_fetched = 0
    next_page_token = None

    while total_fetched < limit:
        # Calculate how many more we need
        remaining = limit - total_fetched
        page_limit = min(API_LIMIT, remaining)

        demisto.debug(f"UNIT42FEED: Fetching {indicator_type} page " f"(page_limit={page_limit}, total_so_far={total_fetched})")

        # Make API call
        response = client.get_indicators(
            indicator_types=[indicator_type], limit=page_limit, start_time=start_time, next_page_token=next_page_token
        )

        # Parse response
        if not response or not isinstance(response, dict):
            demisto.debug(f"UNIT42FEED: Invalid response for {indicator_type}, stopping")
            break

        data = response.get("data", [])
        if not data or not isinstance(data, list):
            demisto.debug(f"UNIT42FEED: No more data for {indicator_type}, stopping")
            break

        # Parse indicators for this page
        page_indicators = parse_indicators(data, feed_tags, tlp_color)

        # Ensure we don't exceed the limit (safety check, in case the API returns more than requested)
        if total_fetched + len(page_indicators) > limit:
            page_indicators = page_indicators[: limit - total_fetched]

        # Push this page to the server immediately, in batches of 2000
        push_indicators_in_batches(page_indicators)

        total_fetched += len(page_indicators)

        demisto.debug(
            f"UNIT42FEED: Parsed and pushed {len(page_indicators)} {indicator_type} indicators (total: {total_fetched})"
        )

        # Check for next page
        metadata = response.get("metadata", {})
        next_page_token = metadata.get("next_page_token") if isinstance(metadata, dict) else None

        if not next_page_token:
            demisto.debug(f"UNIT42FEED: No more pages for {indicator_type}")
            break

    return total_fetched


def fetch_threat_objects_with_limit(client: Client, limit: int, feed_tags: list, tlp_color: str | None) -> int:
    """
    Fetch threat objects with pagination and limit enforcement.

    Each fetched page (up to API_LIMIT threat objects) is parsed and immediately pushed
    to the server in batches of 2000, instead of being accumulated in memory until
    the entire fetch cycle completes. This prevents out-of-memory issues when large
    numbers of threat objects are fetched.

    Args:
        client: Client object
        limit: Maximum number to fetch
        feed_tags: Tags to add to threat objects
        tlp_color: TLP color

    Returns:
        Number of threat objects fetched and pushed (count <= limit)
    """
    total_fetched = 0
    next_page_token = None

    while total_fetched < limit:
        # Calculate how many more we need
        remaining = limit - total_fetched
        page_limit = min(API_LIMIT, remaining)

        demisto.debug(f"UNIT42FEED: Fetching threat objects page " f"(page_limit={page_limit}, total_so_far={total_fetched})")

        # Make API call
        response = client.get_threat_objects(limit=page_limit, next_page_token=next_page_token)

        # Parse response
        if not response or not isinstance(response, dict):
            demisto.debug("UNIT42FEED: Invalid response for threat objects, stopping")
            break

        data = response.get("data", [])
        if not data or not isinstance(data, list):
            demisto.debug("UNIT42FEED: No more threat objects data, stopping")
            break

        # Parse threat objects for this page (note: this may include extra location
        # indicators derived from a single threat object, so it can exceed len(data))
        page_objects = parse_threat_objects(data, feed_tags, tlp_color)

        # Ensure we don't exceed the limit (safety check, in case the API returns more than requested)
        if total_fetched + len(page_objects) > limit:
            page_objects = page_objects[: limit - total_fetched]

        # Push this page to the server immediately, in batches of 2000
        push_indicators_in_batches(page_objects)

        total_fetched += len(page_objects)

        demisto.debug(f"UNIT42FEED: Parsed and pushed {len(page_objects)} threat objects (total: {total_fetched})")

        # Check for next page
        metadata = response.get("metadata", {})
        next_page_token = metadata.get("next_page_token") if isinstance(metadata, dict) else None

        if not next_page_token:
            demisto.debug("UNIT42FEED: No more pages for threat objects")
            break

    return total_fetched


def test_module(client: Client) -> str:
    """Builds the iterator to check that the feed is accessible.
    Args:
        client: Client object.

    Returns:
        Outputs.
    """
    # Test connection by getting a small number of indicators
    try:
        client.get_indicators(limit=1)
        return "ok"
    except Exception as e:
        return f"Failed to connect to Unit 42 API. Check your Server URL and License. Error: {str(e)}"


def fetch_indicators(client: Client, params: dict, current_time: datetime) -> int:
    """Retrieves indicators from the feed with per-type limit enforcement.

    Indicators/threat objects are pushed to the server as soon as each page is
    fetched and parsed (see fetch_indicator_type and fetch_threat_objects_with_limit),
    instead of being accumulated in memory for the entire fetch cycle. This function
    only tracks and returns the total count fetched, to avoid holding all indicators
    in memory at once and causing out-of-memory issues.

    Args:
        client: Client object with request
        params: demisto.params()
        current_time: The current fetch time.
    Returns:
        int. Total number of indicators/threat objects fetched and pushed to the server.
    """
    total_fetched = 0

    # Get configuration
    feed_types = argToList(params.get("feed_types"))
    indicator_types = argToList(params.get("indicator_types"))
    feed_tags = argToList(params.get("feedTags", []))
    tlp_color = params.get("tlp_color")

    # Get start time
    default_start = (current_time - timedelta(hours=24)).strftime(DATE_FORMAT)
    last_run = demisto.getLastRun() or {}
    start_time = last_run.get("last_successful_run", default_start)

    # Calculate total types (including threat objects if enabled)
    total_types = 0
    if "Threat Objects" in feed_types:
        total_types += 1
    if "Indicators" in feed_types:
        total_types += len(set(indicator_types))

    # Parse limit from params and calculate limit per type
    requested_limit = arg_to_number(params.get("limit"))  # noqa: ucp-param-default  (calculate_limit_per_type handles None)
    limit_per_type = calculate_limit_per_type(requested_limit, total_types)

    demisto.debug(f"UNIT42FEED: Starting fetch with limit_per_type={limit_per_type}, feed_types={feed_types}")
    demisto.debug(f"UNIT42FEED: Total types: {total_types}, max total: {limit_per_type * total_types}")
    demisto.debug(f"UNIT42FEED: Indicator types: {indicator_types}, start_time={start_time}")

    # Track remaining quota for redistribution to the last type
    remaining_quota = 0

    # FETCH THREAT OBJECTS FIRST (if enabled) - Highest Priority
    if "Threat Objects" in feed_types:
        demisto.debug(f"UNIT42FEED: Fetching Threat Objects (limit: {limit_per_type})")

        fetched_count = fetch_threat_objects_with_limit(
            client=client, limit=limit_per_type, feed_tags=feed_tags, tlp_color=tlp_color
        )

        total_fetched += fetched_count

        # Track unused quota
        if fetched_count < limit_per_type:
            remaining_quota += limit_per_type - fetched_count

        demisto.debug(
            f"UNIT42FEED: Fetched {fetched_count}/{limit_per_type} threat objects. "
            f"Total: {total_fetched}, Unused quota: {remaining_quota}"
        )

    # FETCH INDICATORS (if enabled) - After Threat Objects
    if "Indicators" in feed_types:
        sorted_types = sort_indicator_types_by_priority(indicator_types)
        demisto.debug(f"UNIT42FEED: Fetching indicators in priority order: {sorted_types}")

        for idx, ind_type in enumerate(sorted_types):
            is_last_type = idx == len(sorted_types) - 1

            # For the last type, add remaining quota
            if is_last_type:
                type_limit = limit_per_type + remaining_quota
            else:
                type_limit = limit_per_type

            demisto.debug(
                f"UNIT42FEED: Fetching {ind_type} "
                f"(limit: {type_limit}{'[LAST TYPE - includes remaining quota]' if is_last_type else ''})"
            )

            # Fetch this indicator type
            fetched_count = fetch_indicator_type(
                client=client,
                indicator_type=ind_type,
                limit=type_limit,
                start_time=start_time,
                feed_tags=feed_tags,
                tlp_color=tlp_color,
            )

            total_fetched += fetched_count

            # Track unused quota for non-last types
            if not is_last_type and fetched_count < limit_per_type:
                remaining_quota += limit_per_type - fetched_count

            demisto.debug(
                f"UNIT42FEED: Fetched {fetched_count}/{type_limit} {ind_type} indicators. "
                f"Total: {total_fetched}, Unused quota: {remaining_quota}"
            )

    demisto.info(f"UNIT42FEED: Fetch complete. Total indicators: {total_fetched} (limit per type: {limit_per_type})")

    return total_fetched


def get_indicators_command(client: Client, args: dict, feed_tags: list = [], tlp_color: str | None = None) -> CommandResults:
    """Wrapper for retrieving indicators from the feed to the war-room.

    Args:
        client: Client object with request
        args: demisto.args()
        feed_tags: feed tags.
        tlp_color: Traffic Light Protocol color.
    Returns:
        Demisto Outputs.
    """
    limit = arg_to_number(args.get("limit", "10")) or 10  # Default to 10 if None
    indicator_types = argToList(args.get("indicator_types"))
    next_page_token = args.get("next_page_token")

    # Get indicators from the API
    response = client.get_indicators(indicator_types=indicator_types, limit=limit, next_page_token=next_page_token)

    indicators = []
    if response and isinstance(response, dict) and response.get("data"):
        data = response.get("data", [])
        if isinstance(data, list):
            indicators = parse_indicators(data, feed_tags, tlp_color)

    # Create human readable output
    headers = ["value", "type", "score"]
    human_readable = tableToMarkdown("Unit 42 Indicators:", indicators, headers=headers, removeNull=True)

    return CommandResults(
        readable_output=human_readable,
        outputs_prefix="Unit42.Indicator",
        outputs_key_field="value",
        outputs=indicators,
        raw_response=response,
    )


def get_threat_objects_command(client: Client, args: dict, feed_tags: list = [], tlp_color: str | None = None) -> CommandResults:
    """Wrapper for retrieving threat objects from the feed to the war-room.

    Args:
        client: Client object with request
        args: demisto.args()
        feed_tags: feed tags.
        tlp_color: Traffic Light Protocol color.
    Returns:
        Demisto Outputs.
    """
    limit = arg_to_number(args.get("limit", "10")) or 10  # Default to 10 if None
    next_page_token = args.get("next_page_token")

    # Get threat objects from the API
    response = client.get_threat_objects(limit=limit, next_page_token=next_page_token)

    threat_objects = []
    if response and isinstance(response, dict) and response.get("data"):
        data = response.get("data", [])
        if isinstance(data, list):
            threat_objects = parse_threat_objects(data, feed_tags, tlp_color)

    # Create human readable output
    headers = ["value", "type", "score"]
    human_readable = tableToMarkdown("Unit 42 Threat Objects:", threat_objects, headers=headers, removeNull=True)

    return CommandResults(
        readable_output=human_readable,
        outputs_prefix="Unit42.ThreatObject",
        outputs_key_field="value",
        outputs=threat_objects,
        raw_response=response,
    )


def main():  # pragma: no cover
    """
    The main function parses the params and runs the command functions
    """

    global _UCP_AUTH_PARAMS_INJECTED
    _UCP_AUTH_PARAMS_INJECTED = True  # type: ignore[name-defined]  # noqa: F821

    params = demisto.params()

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

    if (arg_to_number(params.get("feedFetchInterval", "720")) or 720) < 720:
        return_error("Feed Fetch Interval parameter must be set to at least 12 hours.")

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

    headers = {"Authorization": f"Bearer {demisto.getLicenseID()}"}

    try:
        client = Client(headers=headers, verify=verify_certificate, proxy=proxy)

        if command == "test-module":
            return_results(test_module(client))

        elif command == "fetch-indicators":
            demisto.debug("UNIT42FEED_DEBUG: Starting fetch-indicators command")
            now = datetime.now()
            demisto.debug(f"UNIT42FEED_DEBUG: Fetch start time: {now.strftime(DATE_FORMAT)}")

            # Indicators/threat objects are pushed to the server incrementally, page by page,
            # inside fetch_indicators (see fetch_indicator_type / fetch_threat_objects_with_limit)
            # to avoid holding the entire fetched dataset in memory at once.
            total_fetched = fetch_indicators(client, params, now)
            demisto.debug(f"UNIT42FEED_DEBUG: fetch_indicators fetched and pushed {total_fetched} indicators")

            demisto.setLastRun({"last_successful_run": now.strftime(DATE_FORMAT)})
            demisto.info(
                f"The fetch-indicators command completed successfully. Next run will fetch from: {now.strftime(DATE_FORMAT)}"
            )

        elif command == "unit42-get-indicators":
            return_results(get_indicators_command(client, demisto.args()))

        elif command == "unit42-get-threat-objects":
            return_results(get_threat_objects_command(client, demisto.args()))

    except Exception as e:
        return_error(f"Failed to execute {command} command. Error: {str(e)}")


if __name__ in ("__main__", "__builtin__", "builtins"):
    main()