Haseen Threat Intel

"""IMPORTS"""

from __future__ import annotations

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401
import re
import base64
import json
import hashlib
from datetime import datetime, UTC
from typing import Any

""" CONSTANTS """

INTEGRATION_NAME = "Haseen Threat Intel"

# STIX 2.x SCO/indicator object types mapped to Cortex FeedIndicatorType.
# Mirrors demisto's own StixParser mapping so downstream playbooks/automations
# receive identical indicator shapes.
STIX_2_TYPES_TO_CORTEX_TYPES: dict[str, str] = {
    "mutex": FeedIndicatorType.MUTEX,
    "windows-registry-key": FeedIndicatorType.Registry,
    "user-account": FeedIndicatorType.Account,
    "email-addr": FeedIndicatorType.Email,
    "autonomous-system": FeedIndicatorType.AS,
    "ipv4-addr": FeedIndicatorType.IP,
    "ipv6-addr": FeedIndicatorType.IPv6,
    "domain": FeedIndicatorType.Domain,
    "domain-name": FeedIndicatorType.Domain,
    "url": FeedIndicatorType.URL,
    "file": FeedIndicatorType.File,
    "md5": FeedIndicatorType.File,
    "sha-1": FeedIndicatorType.File,
    "sha-256": FeedIndicatorType.File,
    "file:hashes": FeedIndicatorType.File,
}

# STIX pattern keywords that carry the actual observable value.
HASH_TYPE_MAP: dict[str, str] = {
    "MD5": "md5",
    "SHA-1": "sha1",
    "SHA-256": "sha256",
    "SHA1": "sha1",
    "SHA256": "sha256",
}

# Non-indicator STIX SDOs we also surface as indicator-like objects so that
# relationship links (indicator -> malware -> attack-pattern -> ...) resolve to
# entities that actually exist in the indicators store. Type strings mirror
# ThreatIntel.ObjectsNames (same values CreateIndicatorsFromSTIX/StixParser use).
STIX_SDO_TYPES_TO_CORTEX_TYPES: dict[str, str] = {
    "malware": ThreatIntel.ObjectsNames.MALWARE,
    "attack-pattern": ThreatIntel.ObjectsNames.ATTACK_PATTERN,
    "threat-actor": ThreatIntel.ObjectsNames.THREAT_ACTOR,
    "intrusion-set": ThreatIntel.ObjectsNames.INTRUSION_SET,
    "campaign": ThreatIntel.ObjectsNames.CAMPAIGN,
    "tool": ThreatIntel.ObjectsNames.TOOL,
    "course-of-action": ThreatIntel.ObjectsNames.COURSE_OF_ACTION,
}

# Canonical Cortex scores for non-indicator SDOs (mirrors ThreatIntel.ObjectsScore).
SDO_TYPE_TO_SCORE: dict[str, int] = {
    "malware": ThreatIntel.ObjectsScore.MALWARE,
    "attack-pattern": ThreatIntel.ObjectsScore.ATTACK_PATTERN,
    "threat-actor": ThreatIntel.ObjectsScore.THREAT_ACTOR,
    "intrusion-set": ThreatIntel.ObjectsScore.INTRUSION_SET,
    "campaign": ThreatIntel.ObjectsScore.CAMPAIGN,
    "tool": ThreatIntel.ObjectsScore.TOOL,
    "course-of-action": ThreatIntel.ObjectsScore.COURSE_OF_ACTION,
}

# STIX relationship_type values we normalise into XSOAR's relationship
# vocabulary. `indicates` -> `indicated-by` is the canonical STIX 2.1 mapping
# (same special-case StixParser applies); everything else must already be a
# valid EntityRelationship name to be accepted.
STIX_RELATIONSHIP_TYPE_MAP: dict[str, str] = {
    "indicates": "indicated-by",
}

# Pattern operators we understand for value extraction.
INDICATOR_EQUALS_VAL_PATTERN = re.compile(r"(\w+)\s*=\s*'([^']+)'")

# STIX 'modified'/'created' timestamps may carry UTC offsets (+03:00) or
# variable precision. Parse to an aware datetime so delta comparisons are safe;
# a string compare only happens to work when every value is a zero-padded 'Z'
# timestamp, which Haseen does not guarantee.
STIX_TIMESTAMP_FORMATS = (
    "%Y-%m-%dT%H:%M:%S.%f%z",
    "%Y-%m-%dT%H:%M:%S%z",
    "%Y-%m-%dT%H:%M:%S.%fZ",
    "%Y-%m-%dT%H:%M:%SZ",
)

DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"


def _parse_stix_timestamp(value: str | None) -> datetime | None:
    """Parse a STIX timestamp to a timezone-aware datetime, or None."""
    if not value:
        return None
    text = str(value).strip()
    for fmt in STIX_TIMESTAMP_FORMATS:
        try:
            dt = datetime.strptime(text, fmt)
            # Normalise naive 'Z' timestamps to UTC-aware.
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=UTC)
            return dt
        except ValueError:
            continue
    return None


""" HELPER FUNCTIONS """


# Map Haseen x_attributes 'severity' values to a Cortex DbotScore.
SEVERITY_TO_SCORE: dict[str, int] = {
    "critical": 3,
    "high": 3,
    "medium": 2,
    "low": 1,
    "info": 0,
}


def _parse_x_attributes(obj: dict[str, Any]) -> tuple[dict[str, Any], int | None]:
    """
    Flatten a STIX object's `x_attributes` list into {name: value} fields.

    Attributes with duplicate names (e.g. two 'Threat Type' entries) are joined
    into a comma-separated string so no intel is lost. 'severity' also drives
    the Cortex indicator score.

    Returns:
        (fields dict, score int or None)
    """
    fields: dict[str, Any] = {}
    score: int | None = None
    x_attributes = obj.get("x_attributes") or []
    for attr in x_attributes:
        if not isinstance(attr, dict):
            continue
        name = (attr.get("name") or "").strip()
        value = attr.get("value")
        if not name or value is None:
            continue
        if name.lower() == "severity":
            score = SEVERITY_TO_SCORE.get(str(value).lower(), 0)
        if name in fields:
            # Accumulate duplicate keys into a comma-separated list.
            existing = fields[name]
            if isinstance(existing, list):
                existing.append(value)
            else:
                fields[name] = [existing, value]
        else:
            fields[name] = value
    # Normalise any list-valued fields to a comma-joined string.
    for k, v in fields.items():
        if isinstance(v, list):
            fields[k] = ", ".join(str(x) for x in v)
    return fields, score


def extract_value_from_pattern(pattern: str, stix_type: str) -> str | None:
    """
    Extract the observable value from a STIX 2.x indicator pattern.

    Supports the common `[ipv4-addr:value = '1.2.3.4']` shape as well as
    `value IN (...)` lists and file-hash patterns. Falls back to the object's
    `value` property (STIX 2.1 SCOs) elsewhere.

    Args:
        pattern: The STIX indicator pattern string.
        stix_type: The STIX object type (e.g. 'ipv4-addr').

    Returns:
        The extracted value string, or None if not determinable from the pattern.
    """
    if not pattern:
        return None

    # Handle file hashes: [file:hashes.'SHA-256' = 'abc...']
    if stix_type == "file":
        hash_match = re.search(r"hashes\.'([^']+)'\s*=\s*'([^']+)'", pattern)
        if hash_match:
            return hash_match.group(2)

    # Standard `= 'value'` extraction.
    equals_match = INDICATOR_EQUALS_VAL_PATTERN.search(pattern)
    if equals_match:
        return equals_match.group(2)

    # `value IN ('a', 'b')` lists — take the first, or join when URL/domain.
    in_match = re.search(r"value\s*IN\s*\(+('.*?')\)", pattern)
    if in_match:
        values = re.findall(r"'([^']+)'", in_match.group(1))
        if values:
            return values[0]

    return None


def _normalize_relationship_type(relationship_type: str) -> str | None:
    """
    Normalize a STIX relationship_type into XSOAR relationship vocabulary.

    Mirrors StixParser.parse_relationships: `indicates` maps to `indicated-by`,
    any other type must already be a valid EntityRelationship name or it is
    skipped (with a debug log).
    """
    if not relationship_type:
        return None
    relationship_type = STIX_RELATIONSHIP_TYPE_MAP.get(relationship_type, relationship_type)
    if relationship_type not in EntityRelationship.Relationships.RELATIONSHIPS_NAMES:
        return None
    return relationship_type


def _indicator_to_id_map(indicators: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    """
    Build a STIX object id -> {value, type} lookup from parsed indicators.

    Each indicator's rawJSON carries the source STIX object `id`, which is what
    `relationship` objects reference via source_ref / target_ref.
    """
    id_map: dict[str, dict[str, Any]] = {}
    for ind in indicators:
        raw = ind.get("rawJSON") or {}
        stix_id = raw.get("id")
        if stix_id:
            id_map.setdefault(stix_id, {"value": ind.get("value"), "type": ind.get("type")})
    return id_map


def _parse_relationships(
    relationships_lst: list[dict[str, Any]],
    id_map: dict[str, dict[str, Any]],
) -> dict[str, list[dict[str, Any]]]:
    """
    Build XSOAR EntityRelationship indicator dicts from STIX `relationship`
    objects — the exact output shape StixParser produces, which
    CreateIndicatorsFromSTIX (and the relationship graph) reads back.

    Returns a dict keyed by the A-side (source) indicator value, mapping to the
    list of relationship dicts that attach to it.

    Truly dangling references (either endpoint absent from the bundle window)
    are skipped with a debug log, matching StixParser.parse_relationships.
    """
    by_a_value: dict[str, list[dict[str, Any]]] = {}

    # Haseen's exporter emits each relationship object twice (duplicate STIX
    # ids). Dedup on (source_ref, relationship_type, target_ref) so the graph
    # doesn't render duplicate edges.
    seen: set[tuple[str, str, str]] = set()

    for rel in relationships_lst:
        relationship_type_raw = rel.get("relationship_type")
        if not isinstance(relationship_type_raw, str):
            demisto.debug(f"Invalid/unsupported relationship_type: {relationship_type_raw!r}")
            continue
        relationship_type = _normalize_relationship_type(relationship_type_raw)
        if relationship_type is None:
            demisto.debug(f"Invalid/unsupported relationship_type: {relationship_type_raw!r}")
            continue

        a_stixid = rel.get("source_ref", "")
        b_stixid = rel.get("target_ref", "")
        dedup_key = (a_stixid, relationship_type, b_stixid)
        if dedup_key in seen:
            continue
        seen.add(dedup_key)

        a_obj = id_map.get(a_stixid)
        b_obj = id_map.get(b_stixid)

        if not a_obj or not b_obj:
            # Truly dangling edge: one or both endpoints aren't present in this
            # bundle window and there is no loadable object to resolve them to.
            # Match StixParser.parse_relationships and skip the relationship
            # with a debug log. We deliberately do NOT synthesize a placeholder
            # endpoint here — fabricating a "ghost" SDO (whose only signal is a
            # STIX id stem) creates phantom indicators that (a) collide on XSOAR's
            # (value, type) dedupe key and (b) re-appear as empty-sourceInstances
            # stubs when the real object later shows up. A dangling reference
            # resolves cleanly once its object appears in a future bundle window.
            demisto.debug(f"Cant find {a_stixid=} or {b_stixid=} — skipping relationship (endpoint object not in bundle).")
            continue

        a_value, a_type = a_obj.get("value"), a_obj.get("type")
        b_value, b_type = b_obj.get("value"), b_obj.get("type")
        if not (a_value and a_type and b_value and b_type):
            continue

        entity_relation = EntityRelationship(
            name=relationship_type,
            entity_a=a_value,
            entity_a_type=a_type,
            entity_b=b_value,
            entity_b_type=b_type,
            fields={
                "firstseenbysource": rel.get("created"),
                "lastseenbysource": rel.get("modified"),
            },
        )
        indicator_relationship = entity_relation.to_indicator()
        by_a_value.setdefault(a_value, []).append(indicator_relationship)

    return by_a_value


def parse_stix_bundle(bundle: dict[str, Any], tags: list[str] | None = None) -> list[dict[str, Any]]:
    """
    Parse a STIX 2.x JSON bundle into Cortex indicator dictionaries.

    Args:
        bundle: A parsed STIX 2.x bundle dict (contains an 'objects' list).
        tags: Optional list of feed-level tags (from the `feedTags` integration
            parameter) applied to every emitted indicator, merged with any
            per-object `labels`.

    Returns:
        A list of indicator dicts in Cortex format, each carrying 'value',
        'type', 'score', 'fields', 'rawJSON', the ISO 'modified' timestamp and
        — when the object participates in a STIX relationship — a
        'relationships' list of XSOAR EntityRelationship dicts.
    """
    configured_tags: list[str] = list(tags) if tags else []
    indicators: list[dict[str, Any]] = []
    relationship_objects: list[dict[str, Any]] = []
    objects = bundle.get("objects", [])

    # Pass 1: parse indicator/SCO/SDO objects into indicators.
    for obj in objects:
        if not isinstance(obj, dict):
            continue
        stix_type = obj.get("type", "")

        if stix_type == "relationship":
            relationship_objects.append(obj)
            continue

        value: str | None = None
        cortex_type: str | None = None

        if stix_type == "indicator":
            # STIX indicator SDO: type is derived from the pattern's leading SCO.
            pattern = obj.get("pattern", "")
            pattern_type = pattern.strip().lstrip("[").split(":")[0] if pattern else ""
            cortex_type = STIX_2_TYPES_TO_CORTEX_TYPES.get(pattern_type)
            value = extract_value_from_pattern(pattern, pattern_type)
        elif stix_type in STIX_2_TYPES_TO_CORTEX_TYPES:
            # STIX 2.1 SCO: has a direct 'value' property (ipv4-addr, domain, etc.)
            cortex_type = STIX_2_TYPES_TO_CORTEX_TYPES[stix_type]
            value = obj.get("value")
            # File SCOs carry hashes instead of a single 'value'.
            if stix_type == "file" and not value:
                hashes = obj.get("hashes") or {}
                value = (
                    hashes.get("SHA-256")
                    or hashes.get("SHA-1")
                    or hashes.get("MD5")
                    or hashes.get("sha256")
                    or hashes.get("sha1")
                    or hashes.get("md5")
                )
        elif stix_type in STIX_SDO_TYPES_TO_CORTEX_TYPES:
            # Non-indicator SDO (malware, attack-pattern, threat-actor, ...):
            # surfaced as an indicator so relationship links can resolve to it.
            cortex_type = STIX_SDO_TYPES_TO_CORTEX_TYPES[stix_type]
            value = obj.get("name")
            # Haseen emits threat-actor objects UUID-keyed with no recoverable
            # real name — most carry only the generic attribution marker
            # 'Attributed'. That is a status flag, not an actor identity; emitting
            # it as a Threat Actor would collapse dozens of distinct actors into a
            # single meaningless 'Attributed' entity (XSOAR dedupes by value).
            # Skip these so only real-named actors survive.
            if (
                stix_type == "threat-actor"
                and isinstance(value, str)
                and value.strip().lower() in {"attributed", "unattributed", "unknown", "n/a", ""}
            ):
                demisto.debug(f"{INTEGRATION_NAME}: skipping threat-actor {obj.get('id')} (no real actor name, got {value!r})")
                continue

        if not value or not cortex_type:
            continue

        # Normalize domain-name SCOs that carry URLs.
        if cortex_type == FeedIndicatorType.Domain and isinstance(value, str) and "://" in value:
            cortex_type = FeedIndicatorType.URL

        modified = obj.get("modified") or obj.get("created") or datetime.now(UTC).strftime(DATE_FORMAT)

        custom_fields: dict[str, Any] = {}
        if obj.get("name"):
            custom_fields["name"] = obj["name"]
        if obj.get("description"):
            custom_fields["description"] = obj["description"]
        labels = obj.get("labels")
        merged_tags: list[str] = list(configured_tags)
        if labels:
            merged_tags.extend(labels if isinstance(labels, list) else [labels])
        if merged_tags:
            # De-duplicate while preserving order.
            custom_fields["tags"] = list(dict.fromkeys(merged_tags))

        # Fold x_attributes (severity, Threat Type, Role, Threat Detail, …) into
        # fields and derive the Cortex score from 'severity'.
        x_fields, score = _parse_x_attributes(obj)
        custom_fields.update(x_fields)

        # XSOAR resolves an indicator's true type from rawJSON. For SDOs the raw
        # STIX object's 'type' is lowercase/dashed ("attack-pattern", "malware"),
        # which contradicts the Cortex type ("Attack Pattern", "Malware") and is
        # not a valid indicator type — so persistence silently drops the record.
        # Rewrite rawJSON to be self-consistent (mirrors FeedTAXII / FeedMitreAttack).
        raw_json: dict[str, Any] = dict(obj)
        raw_json["type"] = cortex_type
        raw_json["value"] = value

        indicator: dict[str, Any] = {
            "value": value,
            "type": cortex_type,
            "fields": custom_fields,
            "rawJSON": raw_json,
            "modified": modified,
            "is_sdo": stix_type in STIX_SDO_TYPES_TO_CORTEX_TYPES,
        }

        if score is not None:
            indicator["score"] = score
        elif stix_type in SDO_TYPE_TO_SCORE:
            # SDOs (malware, attack-pattern, ...) get their canonical Cortex
            # score (mirrors ThreatIntel.ObjectsScore) when the feed doesn't
            # supply an explicit severity-based score via x_attributes.
            indicator["score"] = SDO_TYPE_TO_SCORE[stix_type]

        indicators.append(indicator)

    # Pass 2: build id -> {value, type} map and attach relationships. SDOs
    # (malware, attack-pattern, threat-actor, tool, ...) are parsed above as
    # first-class indicators — XSOAR treats any STIX object (SCO *or* SDO, plus
    # user-defined types) as an ingestible indicator, and the relationship graph
    # traverses both directions (e.g. IP ← indicator-of ← malware ← uses ←
    # attack-pattern), so they must be retained as indicator records.
    id_map = _indicator_to_id_map(indicators)
    relationships_by_a = _parse_relationships(relationship_objects, id_map)

    for ind in indicators:
        rels = relationships_by_a.get(ind["value"])
        if rels:
            ind["relationships"] = rels

    return indicators


# Volatile STIX fields that Haseen re-stamps on mass re-exports without a real
# content change. These must be excluded from the change-detection fingerprint,
# otherwise every re-export looks like a new modification.
_VOLATILE_STIX_FIELDS = frozenset(
    {
        "modified",
        "created",
        "revoked",
        "x_publish_timestamp",
        "x_ingest_timestamp",
        "x_last_seen",
        "first_seen",
        "last_seen",
    }
)


def _indicator_fingerprint(ind: dict[str, Any]) -> str:
    """
    Stable content hash of an indicator, ignoring volatile re-stamp fields and
    case-variant naming.

    Haseen re-exports the same SDO (malware/attack-pattern) in case-variant form
    ("XWorm" vs "Xworm") and re-stamps `modified` each export. Strip volatile
    fields and case-fold string values so those two quirks don't masquerade as a
    genuine change; only a real content edit changes the hash.
    """
    raw = ind.get("rawJSON") or {}
    stable = {k: v for k, v in raw.items() if k not in _VOLATILE_STIX_FIELDS}

    def _normalise(val: Any) -> Any:
        if isinstance(val, str):
            return val.casefold()
        if isinstance(val, list):
            return [_normalise(v) for v in val]
        if isinstance(val, dict):
            return {k: _normalise(v) for k, v in sorted(val.items())}
        return val

    payload = {
        "type": ind.get("type"),
        "value": _normalise(ind.get("value")),
        "raw": _normalise(stable),
        "fields": _normalise(ind.get("fields")),
    }
    # canonical JSON with sorted keys for reproducibility
    blob = json.dumps(payload, sort_keys=True, default=str, ensure_ascii=False)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()


def fetch_indicators_command(
    client: Client,
    first_fetch: str,
    limit: int,
    last_run: dict,
    tags: list[str] | None = None,
) -> tuple[list[dict[str, Any]], dict]:
    """
    Fetch indicators from the Haseen STIX bundle with id-keyed delta.

    Haseen re-stamps the STIX `modified` field on nearly every object on each
    bundle re-export (cluster stamps like N indicators sharing one export time),
    so `modified` alone is NOT a reliable change signal — a pure
    `modified > cutoff` filter would re-select almost the whole bundle every
    hour. Instead we track which STIX object ids have already been ingested and
    only emit an indicator when its id is new OR its `modified` has advanced
    past the value we last saw for that exact id.

    Args:
        client: Authenticated BaseClient wrapper.
        first_fetch: First-fetch lookback (parse_date_range spec).
        limit: Max indicators to return this run (-1 = unlimited).
        last_run: The previous fetch's last_run dict.
        tags: Optional feed-level tags applied to every fetched indicator.

    Returns:
        (indicators, updated_last_run).
    """
    # last_run carries two things:
    #   "seen": {stix_id -> content fingerprint}  (id-keyed watermark)
    #   "last_modified": high-water mark string (fallback / first-seed backing field)
    #
    # The fingerprint is a hash of the indicator's stable content *with volatile
    # fields stripped and the value case-normalised*. Haseen re-stamps `modified`
    # to a new batch time on mass re-exports WITHOUT changing object content, and
    # also re-exports the same SDO (malware/attack-pattern) multiple times with
    # case-variant names ("XWorm"/"Xworm"). Both quirks would otherwise look like
    # a genuine change every cycle. Hashing case-normalised, volatile-stripped
    # content collapses re-stamps AND case-variant dupes into one identity, so a
    # re-export emits nothing and only a real edit re-emits.
    seen: dict[str, str] = last_run.get("seen") or {}
    last_modified = last_run.get("last_modified") or parse_date_range(first_fetch, date_format=DATE_FORMAT)[0]

    bundle = client.fetch_bundle()

    all_indicators = parse_stix_bundle(bundle, tags=tags)
    demisto.debug(f"{INTEGRATION_NAME}: parsed {len(all_indicators)} indicators from bundle")

    # Fallback cut-off for objects that predate id-tracking (first boot or a
    # last_run that was reset). STIX-string ordered compare is unsafe across
    # '+03:00' offsets, so parse to aware datetimes.
    cutoff = _parse_stix_timestamp(last_modified) or datetime(1970, 1, 1, tzinfo=UTC)

    new_indicators: list[dict[str, Any]] = []
    latest_modified = last_modified
    latest_ts = cutoff
    updated_seen: dict[str, str] = dict(seen)

    for ind in all_indicators:
        raw = ind.get("rawJSON") or {}
        stix_id = raw.get("id")
        mod = ind.get("modified", "")
        mod_ts = _parse_stix_timestamp(mod)

        # Advance the high-water mark over every object (id-keyed or not) so the
        # fallback cut-off only matters on the very first seed.
        if mod_ts is not None and mod_ts > latest_ts:
            latest_ts = mod_ts
            latest_modified = mod

        prev_fp = updated_seen.get(stix_id) if isinstance(stix_id, str) else None
        cur_fp = _indicator_fingerprint(ind)

        if stix_id is None:
            # No STIX id (shouldn't happen for Haseen, but be safe): fall back to
            # the modified cut-off so we never drop a real indicator.
            emit = mod_ts is None or mod_ts > cutoff
        elif prev_fp is None:
            # Never seen this object before -> new -> emit + record fingerprint.
            emit = True
            updated_seen[stix_id] = cur_fp
        elif prev_fp != cur_fp:
            # Same id, changed content -> genuine edit -> emit + update fingerprint.
            emit = True
            updated_seen[stix_id] = cur_fp
        else:
            # Same id, identical content (re-stamp only) -> ignore.
            emit = False

        if emit:
            new_indicators.append(ind)

    # Prune the seen map to ids that still exist in the feed so it can't grow
    # without bound as the feed retires old objects.
    live_ids = {ind.get("rawJSON", {}).get("id") for ind in all_indicators}
    updated_seen = {k: v for k, v in updated_seen.items() if k in live_ids}

    if limit and limit > 0:
        new_indicators = new_indicators[:limit]

    if not new_indicators:
        demisto.debug(f"{INTEGRATION_NAME}: no new/updated indicators since {last_modified}; skipping ingest this cycle.")

    last_run["seen"] = updated_seen
    last_run["last_modified"] = latest_modified
    return new_indicators, last_run


def test_module(client: Client) -> str:
    """
    Validate connectivity and a successful bundle parse.
    """
    bundle = client.fetch_bundle()
    if not isinstance(bundle, dict) or "objects" not in bundle:
        raise DemistoException("Invalid STIX bundle returned by the feed (missing 'objects').")
    demisto.debug(f"{INTEGRATION_NAME}: test-module fetched {len(bundle.get('objects', []))} STIX objects")
    return "ok"


""" CLIENT """


class Client(BaseClient):
    """
    Thin authenticated client for the Haseen STIX 2.x feed endpoint.

    Per the Haseen API Integration Guide, authentication is:
      * A `token` passed as a URL query parameter (?token=<token>) on every
        request, plus
      * OPTIONAL HTTP Basic Auth (email as username, the API token as
        password) for "certain exports".

    The feed is rate-limited to 2 requests/hour per export type (429 on
    exceed, throttled for ~3600s), so fetch_bundle() must be called sparingly.
    """

    def __init__(
        self,
        url: str,
        token: str,
        verify: bool,
        proxy: bool,
        username: str | None = None,
        password: str | None = None,
    ):
        headers = {"Accept": "application/json"}
        if username and password:
            # Optional Basic Auth for exports that require it. The guide maps
            # username -> account email, password -> API token.
            b64 = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
            headers["Authorization"] = f"Basic {b64}"
        super().__init__(base_url=url, verify=verify, proxy=proxy, headers=headers)
        self._token = token

    def fetch_bundle(self) -> dict[str, Any]:
        """
        Download the STIX 2.x bundle from the configured feed URL.

        The token is appended as a `token` query parameter (not a header), per
        the Haseen guide.

        Returns:
            The parsed bundle dict.
        """
        params = {"token": self._token} if self._token else {}
        res = self._http_request(method="GET", url_suffix="", params=params, timeout=60, resp_type="json")
        # STIX bundles are either a bare bundle (has 'objects') or wrapped.
        if isinstance(res, dict) and "objects" in res:
            return res
        if isinstance(res, list):
            return {"objects": res}
        raise DemistoException(f"Unexpected feed payload shape: {type(res).__name__}")


""" MAIN """


def main() -> None:
    params = demisto.params()
    url = params.get("url", "").rstrip("/")
    token = params.get("api_token") or ""
    verify = not params.get("insecure", False)
    proxy = params.get("proxy", False)
    first_fetch = params.get("first_fetch", "7 days")
    limit = arg_to_number(params.get("limit", 0)) or 0
    # Optional Basic Auth credentials (email + token) for exports that require
    # them in addition to the token query parameter. Empty when not configured.
    username = (params.get("credentials") or {}).get("identifier", "")
    password = (params.get("credentials") or {}).get("password", "")
    feed_tags_raw = params.get("feedTags", "")
    feed_tags: list[str] = [t.strip() for t in str(feed_tags_raw).split(",") if t.strip()]

    if not url:
        raise DemistoException("The 'url' parameter is required.")

    command = demisto.command()
    client = Client(
        url=url,
        token=token,
        verify=verify,
        proxy=proxy,
        username=username or None,
        password=password or None,
    )

    try:
        if command == "test-module":
            return_results(test_module(client))
        elif command == "fetch-indicators":
            last_run = get_feed_last_run() or {}
            # The dedupe watermark (`seen`) is a large nested dict. Feed
            # frameworks sometimes drop unknown keys from last_run between
            # invocations, which would reset `seen` to {} every fetch and make
            # the delta logic re-emit the whole feed (the "Pulled 361" loop).
            # Persist `seen` in the integration context instead — it survives
            # container restarts and is not rewritten by the feed framework.
            ctx = demisto.getIntegrationContext() or {}
            saved_seen = ctx.get("haseen_seen")
            if isinstance(saved_seen, dict) and saved_seen:
                # Context is authoritative: prefer it, but fold in any seen ids
                # that only exist in the framework last_run (defensive merge).
                fb_seen = last_run.get("seen")
                if isinstance(fb_seen, dict):
                    merged = dict(saved_seen)
                    merged.update(fb_seen)
                    last_run["seen"] = merged
                else:
                    last_run["seen"] = dict(saved_seen)

            indicators, new_last_run = fetch_indicators_command(client, first_fetch, limit, last_run, tags=feed_tags)

            batch_size = 2000
            for i in range(0, len(indicators), batch_size):
                demisto.createIndicators(indicators[i : i + batch_size])
            set_feed_last_run(new_last_run)
            # Mirror the seen watermark into the integration context so it can
            # never be silently reset by the feed framework.
            ctx["haseen_seen"] = new_last_run.get("seen", {})
            demisto.setIntegrationContext(ctx)
        else:
            raise NotImplementedError(f"Command '{command}' is not implemented.")
    except Exception as e:
        return_error(f"Failed to execute {command} command. Error: {str(e)}")


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