From f1dc8eb6d9ace2f8a33ae472067c8e3a2ebdc3a4 Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:31:10 +0100 Subject: [PATCH 1/9] Threat intelligence add --- conf/default/processing.conf.default | 5 + conf/default/threat_intel.conf.default | 126 ++++ .../threatintelligence/__init__.py | 0 .../integrations/threatintelligence/base.py | 407 +++++++++++++ .../integrations/threatintelligence/cache.py | 69 +++ .../integrations/threatintelligence/config.py | 80 +++ .../malpedia_actor_provider.py | 196 ++++++ .../threatintelligence/malpedia_provider.py | 277 +++++++++ .../threatintelligence/registry.py | 96 +++ .../threatintelligence/threatfox_provider.py | 298 +++++++++ modules/processing/threatintelligence.py | 572 ++++++++++++++++++ utils/threatintelligence_diag.py | 122 ++++ web/analysis/views.py | 1 + web/templates/analysis/network/_dns.html | 9 + .../analysis/network/_dns_not_ajax.html | 9 + web/templates/analysis/network/_hosts.html | 9 + .../analysis/network/_hosts_not_ajax.html | 9 + web/templates/analysis/overview/index.html | 172 ++++++ 18 files changed, 2457 insertions(+) create mode 100644 conf/default/threat_intel.conf.default create mode 100644 lib/cuckoo/common/integrations/threatintelligence/__init__.py create mode 100644 lib/cuckoo/common/integrations/threatintelligence/base.py create mode 100644 lib/cuckoo/common/integrations/threatintelligence/cache.py create mode 100644 lib/cuckoo/common/integrations/threatintelligence/config.py create mode 100644 lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py create mode 100644 lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py create mode 100644 lib/cuckoo/common/integrations/threatintelligence/registry.py create mode 100644 lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py create mode 100644 modules/processing/threatintelligence.py create mode 100644 utils/threatintelligence_diag.py diff --git a/conf/default/processing.conf.default b/conf/default/processing.conf.default index 264fa4335d5..e7bba3f97f4 100644 --- a/conf/default/processing.conf.default +++ b/conf/default/processing.conf.default @@ -361,3 +361,8 @@ enabled = no # Code-similarity engine(s). See conf/integrations.conf [similarity]. [similarity] enabled = no + +# Threat-intelligence enrichment. Engine and API settings live in +# conf/threat_intel.conf. +[threatintelligence] +enabled = no diff --git a/conf/default/threat_intel.conf.default b/conf/default/threat_intel.conf.default new file mode 100644 index 00000000000..454f010a059 --- /dev/null +++ b/conf/default/threat_intel.conf.default @@ -0,0 +1,126 @@ +# Threat-intelligence enrichment. +# +# Layout: +# [threatintelligence] global behaviour shared by every engine +# [] one section per engine; "enabled" turns it on and +# all other keys are engine-local +# +# Engine sections inherit the global section, so an engine only needs to set +# what differs. Adding a new engine means adding a section here and one line +# in lib/cuckoo/common/integrations/threatintelligence/registry.py -- no +# changes to this file's structure. +# +# Enable the processing module itself in processing.conf [threatintelligence]. + +# ============================================================ +# Global behaviour +# ============================================================ +[threatintelligence] + +# Per-lookup deadline (seconds). +timeout = 20 +# Cap matches shown per host/domain (top N by confidence). 0 = no cap. +max_results = 5 +# Best-effort cross-analysis cache to limit duplicate API calls. +cache = yes +cache_ttl = 86400 +# Run lookups concurrently. +concurrent_lookups = yes +max_workers = 4 + +# ---- Indicator types looked up (global router gate) -------- +# Each engine only receives the types it supports. Indicators are +# reconstructed from the analysis artifacts: +# ip addresses - contacted host IPs +# domains - resolved / requested domains +# urls - full URLs reconstructed from HTTP traffic +# sha256 - the submitted sample's sha256 +ip addresses = yes +domains = yes +urls = no +sha256 = no + +# ---- Accuracy / false-positive control --------------------- +# A `url` IOC also matches a contacted domain/IP by HOST, so a dedicated +# malicious domain published as a url still tags. Dead-drop-resolver false +# positives (a legitimate shared host carrying url IOCs for MANY families) +# are auto-detected by family multiplicity on a single host and suppressed. +# Enable `urls` above to additionally match the exact URL path fetched. +# +# strict_ioc_type_match: require same-type matching only (a url IOC then +# never matches a bare domain/IP). +strict_ioc_type_match = no +# +# Secondary safety valve: if a single indicator still resolves to more than +# this many distinct families (e.g. a shared bulletproof IP), treat it as +# ambiguous shared infrastructure and suppress its tags. 0 = off. +max_families_per_indicator = 5 + +# ---- Promotion to CAPE detections -------------------------- +# Append a CAPE detection for indicator hits. This does NOT set the headline +# malfamily (it only adds a detection entry), so infrastructure attribution +# never overrides CAPE's own classification. The report card enriches +# regardless of these settings. +promote_to_detection = no +# Promote only C2 (botnet_cc) hits -- the high-confidence type. When yes, +# other threat types still enrich the card but are not promoted. +promote_c2_only = yes +# Minimum confidence (0-100) required to promote. +promote_minimum_confidence = 75 + +# ---- Threat actors (master gate) --------------------------- +# Attributing a sample to a named threat actor is a strong claim. Actor +# engines are gated by this toggle AND their own "enabled" key, and are +# intended to be driven by an authoritative, high-confidence source. +threat actors = no +# Build actor hints from the actors that matched families are attributed to. +# Family<->actor links are often community-sourced, so this is opt-in even +# when the master gate is on. +actor_attribution_from_families = no + +# ============================================================ +# Engines +# ============================================================ + +# ---- ThreatFox (abuse.ch) -- indicator engine -------------- +[threatfox] +enabled = no +# Auth-Key -- REQUIRED by abuse.ch. The engine reports itself unavailable +# when blank. Free key: https://auth.abuse.ch/ +api_key = +host = https://threatfox-api.abuse.ch/api/v1/ +# Wildcard (no) vs exact (yes); results are host-exact post-filtered either +# way, so leave as no. +exact_match = no +# ip:port precision: restrict ip:port IOC matches to the exact ports observed +# in the analysis. Off = surface any port ThreatFox knows for a contacted IP. +match_ports_only = no +# Minimum confidence_level (0-100) for a match to be shown. +minimum_confidence = 50 +# Rate-limit resilience: retry on HTTP 429 / 5xx / timeout with exponential +# backoff (honours Retry-After). +retries = 3 +backoff = 0.5 + +# ---- Malpedia (Fraunhofer FKIE) -- family engine ----------- +[malpedia] +enabled = yes +# APIToken -- OPTIONAL. get/family and find/family are public; a token only +# raises rate limits. Sent as "Authorization: apitoken ". +api_key = +host = https://malpedia.caad.fkie.fraunhofer.de/api +# How many recent reference reports to link per family card. +max_references = 4 +# Also describe families surfaced by indicator-engine infrastructure hits, +# not only CAPE's own detections. +from_threatfox = yes + +# ---- Malpedia actors -- actor engine (reference impl) ------ +# Reference implementation only, and additionally gated by the global +# "threat actors" toggle. Malpedia actor data is community/MISP curated; a +# dedicated high-confidence provider is preferred for confident attribution. +[malpedia_actors] +enabled = no +api_key = +host = https://malpedia.caad.fkie.fraunhofer.de/api +max_references = 4 diff --git a/lib/cuckoo/common/integrations/threatintelligence/__init__.py b/lib/cuckoo/common/integrations/threatintelligence/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/lib/cuckoo/common/integrations/threatintelligence/base.py b/lib/cuckoo/common/integrations/threatintelligence/base.py new file mode 100644 index 00000000000..9351b9d50f9 --- /dev/null +++ b/lib/cuckoo/common/integrations/threatintelligence/base.py @@ -0,0 +1,407 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Threat-intelligence framework — base abstractions. + +Two provider kinds share one registry and config section: + +* IndicatorProvider — resolves a network indicator (ip / domain / sha256) + to threat context. ThreatFox is the first. Produces IntelMatch records + that become the red tags rendered against hosts/domains on the network + page (``threat_type:malware_printable``). + +* FamilyProvider — resolves a malware FAMILY (a name or a Malpedia id) to a + descriptive card. Malpedia is the first. Produces FamilyCard records that + populate the collapsible "Threat Intelligence" section on the report page. + +The two are connected: families discovered by an IndicatorProvider hit +(e.g. ThreatFox returns ``win.cobalt_strike``) are fed to the family +providers, so an infrastructure match also yields a malware card. + +Zero external dependencies (stdlib + whatever CAPE already ships); any +optional package is imported lazily inside available()/lookup(). +""" + +import logging +import re +from typing import Dict, List, Optional, Set +from urllib.parse import urlparse + +log = logging.getLogger(__name__) + +# Indicator-type tokens. +IND_IP = "ip" +IND_DOMAIN = "domain" +IND_HASH = "hash" +IND_URL = "url" + +# Malpedia family ids look like ``platform.family_name`` (win.emotet, +# elf.mirai, apk.flubot, osx.x, js.x, py.x, jar.x, ...). +_MALPEDIA_ID_RE = re.compile(r"^[a-z0-9]{2,12}\.[a-z0-9_]+$") +_MALPEDIA_DETAILS = "https://malpedia.caad.fkie.fraunhofer.de/details/" + + +def _as_bool(value) -> bool: + if isinstance(value, bool): + return value + if value is None: + return False + return str(value).strip().lower() in ("1", "yes", "true", "on", "y") + + +def normalize_domain(domain: Optional[str]) -> str: + if not domain: + return "" + return domain.strip().rstrip(".").lower() + + +def ioc_host_part(ioc: Optional[str]) -> str: + """Reduce a ThreatFox IOC string to its bare host (ip or domain).""" + if not ioc: + return "" + s = ioc.strip() + if "://" in s: + s = s.split("://", 1)[1] + s = s.split("/", 1)[0] + if s.count(":") == 1: + head, _, tail = s.rpartition(":") + if tail.isdigit(): + s = head + return s.strip().lower() + + +def looks_like_malpedia_id(value: Optional[str]) -> bool: + return bool(value and _MALPEDIA_ID_RE.match(value.strip().lower())) + + +def url_match_key(url: Optional[str]) -> str: + """Normalize a URL for exact, path-sensitive comparison. + + Scheme is intentionally ignored: CAPE's HTTP parser labels every + reconstructed URL "http://" even when the real traffic was TLS, while + ThreatFox may store the same dead-drop-resolver URL as "https://". We + compare host (lower, minus default port) + path (minus trailing slash) + + query, so a URL IOC only matches the EXACT resource the sample + requested -- not merely its host. + """ + if not url: + return "" + s = url.strip() + if "://" in s: + s = s.split("://", 1)[1] + # split off fragment + s = s.split("#", 1)[0] + if "/" in s: + netloc, _, rest = s.partition("/") + path_q = "/" + rest + else: + netloc, path_q = s, "/" + netloc = netloc.lower() + # strip default ports + if netloc.endswith(":80"): + netloc = netloc[:-3] + elif netloc.endswith(":443"): + netloc = netloc[:-4] + # normalize a bare trailing slash so "/path" and "/path/" match, but keep + # the root "/" meaningful + path, sep, query = path_q.partition("?") + if len(path) > 1 and path.endswith("/"): + path = path[:-1] + return netloc + path + (sep + query if sep else "") + + +def malpedia_details_url(family_id: str) -> str: + return _MALPEDIA_DETAILS + family_id.strip().lower() + + +_MALPEDIA_ACTOR = "https://malpedia.caad.fkie.fraunhofer.de/actor/" + + +def malpedia_actor_url(actor_id: str) -> str: + return _MALPEDIA_ACTOR + actor_id.strip().lower() + + +def squash_name(name: Optional[str]) -> str: + """Lower-case and strip to alphanumerics for fuzzy name comparison. + + "Cobalt Strike" -> "cobaltstrike", "CASTLESTEALER" -> "castlestealer", + "win.castle_stealer" -> "wincastlestealer". + """ + return re.sub(r"[^a-z0-9]", "", (name or "").lower()) + + +# ============================================================ +# Indicator side +# ============================================================ +class IntelMatch: + """A single normalized indicator hit (ThreatFox-style fields + badge).""" + + def __init__(self, source, indicator, indicator_type, ioc=None, + threat_type=None, threat_type_desc=None, ioc_type=None, + ioc_type_desc=None, malware=None, malware_printable=None, + malware_alias=None, malware_malpedia=None, confidence_level=None, + first_seen=None, last_seen=None, reference=None, reporter=None, + tags=None, tag_category=None, tag_value=None, + ioc_id=None, indicator_url=None): + self.source = source + self.indicator = indicator + self.indicator_type = indicator_type + self.ioc = ioc + self.ioc_id = ioc_id + # indicator_url: a direct link to THIS indicator hit on the provider + # (e.g. the ThreatFox IOC page), as opposed to family/sample links. + self.indicator_url = indicator_url + self.threat_type = threat_type + self.threat_type_desc = threat_type_desc + self.ioc_type = ioc_type + self.ioc_type_desc = ioc_type_desc + self.malware = malware + self.malware_printable = malware_printable + self.malware_alias = malware_alias + self.malware_malpedia = malware_malpedia + self.confidence_level = confidence_level + self.first_seen = first_seen + self.last_seen = last_seen + self.reference = reference + self.reporter = reporter + self.tags = tags or [] + self.tag_category = tag_category or (threat_type or "") + self.tag_value = tag_value if tag_value is not None else (malware_printable or "") + + @property + def tag(self) -> str: + cat = (self.tag_category or "").strip().lower() + val = (self.tag_value or "").strip().lower() + if cat and val: + return f"{cat}:{val}" + return cat or val + + @property + def tooltip(self) -> str: + bits = [self.source] + if self.confidence_level is not None: + bits.append(f"confidence {self.confidence_level}%") + if self.ioc: + bits.append(f"ioc {self.ioc}") + if self.malware_printable: + bits.append(self.malware_printable) + if self.first_seen: + bits.append(f"first seen {self.first_seen}") + return " | ".join(bits) + + def dedup_key(self) -> str: + return f"{self.source}|{self.tag}" + + def to_dict(self) -> Dict: + return { + "source": self.source, "indicator": self.indicator, + "indicator_type": self.indicator_type, "ioc": self.ioc, + "ioc_id": self.ioc_id, "indicator_url": self.indicator_url, + "threat_type": self.threat_type, "threat_type_desc": self.threat_type_desc, + "ioc_type": self.ioc_type, "ioc_type_desc": self.ioc_type_desc, + "malware": self.malware, "malware_printable": self.malware_printable, + "malware_alias": self.malware_alias, "malware_malpedia": self.malware_malpedia, + "confidence_level": self.confidence_level, "first_seen": self.first_seen, + "last_seen": self.last_seen, "reference": self.reference, + "reporter": self.reporter, "tags": self.tags, + "tag": self.tag, "tooltip": self.tooltip, + } + + +class ProviderResult: + """Per-indicator result from one indicator provider.""" + + def __init__(self, status: str = "ok", error: Optional[str] = None): + self.status = status # ok | no_match | skipped | timeout | error | disabled + self.error = error + self.matches: List[IntelMatch] = [] + + +class IndicatorProvider: + """Base class for an indicator-lookup backend (e.g. ThreatFox).""" + + name = "base" + supported_indicators: Set[str] = set() + + def __init__(self, options: Dict): + self.options = options or {} + self.timeout = int(self.options.get("timeout", 20) or 20) + self.minimum_confidence = int(self.options.get("minimum_confidence", 0) or 0) + self.max_results = int(self.options.get("max_results", 5) or 0) + + def accepts_indicator(self, indicator_type: str) -> bool: + return indicator_type in self.supported_indicators + + def available(self) -> bool: + return True + + def lookup(self, indicator: str, indicator_type: str, ports=None) -> ProviderResult: + raise NotImplementedError + + def _select(self, matches: List[IntelMatch]) -> List[IntelMatch]: + kept = [m for m in matches + if (m.confidence_level is None or m.confidence_level >= self.minimum_confidence)] + best: Dict[str, IntelMatch] = {} + for m in kept: + key = m.dedup_key() + cur = best.get(key) + if cur is None or (m.confidence_level or 0) > (cur.confidence_level or 0): + best[key] = m + ordered = sorted(best.values(), key=lambda m: (m.confidence_level or 0), reverse=True) + if self.max_results and self.max_results > 0: + ordered = ordered[: self.max_results] + return ordered + + +# ============================================================ +# Family side +# ============================================================ +class FamilyCard: + """A descriptive malware-family card for the report-page section.""" + + def __init__(self, source, family_id=None, common_name=None, aliases=None, + description=None, references=None, attribution=None, updated=None, + url=None, provenance=None): + self.source = source + self.family_id = family_id + self.common_name = common_name or (family_id.split(".", 1)[-1] if family_id else None) + self.aliases = aliases or [] + self.description = description or "" + # references: list of {"url":..., "label":...} + self.references = references or [] + self.attribution = attribution or [] # associated actor names (groups later) + self.updated = updated + self.url = url or (malpedia_details_url(family_id) if family_id else None) + # provenance: how this family surfaced (detection / malfamily / threatfox) + self.provenance = set(provenance or ()) + + def to_dict(self) -> Dict: + return { + "source": self.source, + "family_id": self.family_id, + "common_name": self.common_name, + "aliases": self.aliases, + "description": self.description, + "references": self.references, + "attribution": self.attribution, + "updated": self.updated, + "url": self.url, + "provenance": sorted(self.provenance), + } + + +class FamilyProvider: + """Base class for a family-enrichment backend (e.g. Malpedia).""" + + name = "base" + + def __init__(self, options: Dict): + self.options = options or {} + self.timeout = int(self.options.get("timeout", 20) or 20) + + def available(self) -> bool: + return True + + def resolve(self, query: str) -> Optional[str]: + """Resolve a free-text family name to this provider's canonical id.""" + raise NotImplementedError + + def fetch(self, family_id: str, provenance=None) -> Optional[FamilyCard]: + """Fetch a card for a canonical family id.""" + raise NotImplementedError + + def enrich(self, query: str, is_id: bool = False, provenance=None) -> Optional[FamilyCard]: + """Resolve (unless already an id) then fetch a card.""" + family_id = query if is_id else self.resolve(query) + if not family_id: + return None + return self.fetch(family_id, provenance=provenance) + + +def reference_label(url: str) -> str: + """Human-friendly label for a reference link (its registrable host).""" + try: + host = urlparse(url).netloc or url + except Exception: + host = url + return host[4:] if host.startswith("www.") else host + + +# ============================================================ +# Threat-actor side (SCAFFOLDING — wired but disabled by default) +# ============================================================ +# Actor attribution must be HIGH CONFIDENCE. Showing a threat actor against a +# sample is a strong claim, so this path is intended for an authoritative, +# high-confidence actor provider and is gated off by default. A Malpedia actor +# provider exists as a reference implementation but its links derive from +# community/MISP curation (attribution-confidence is often moderate), so it too +# is disabled unless explicitly enabled for a specific, understood scenario. +class ActorCard: + """A descriptive threat-actor card for the report-page section.""" + + def __init__(self, source, actor_id=None, common_name=None, aliases=None, + description=None, country=None, references=None, families=None, + attribution_confidence=None, url=None, provenance=None): + self.source = source + self.actor_id = actor_id + self.common_name = common_name or actor_id + self.aliases = aliases or [] + self.description = description or "" + self.country = country + # references: list of {"url":..., "label":...} + self.references = references or [] + # families: associated malware family names/ids + self.families = families or [] + # attribution_confidence: provider-supplied 0-100 (or None) + self.attribution_confidence = attribution_confidence + self.url = url + # provenance: how the actor surfaced (e.g. family attribution, a + # high-confidence provider attribution, ...) + self.provenance = set(provenance or ()) + + def to_dict(self) -> Dict: + return { + "source": self.source, + "actor_id": self.actor_id, + "common_name": self.common_name, + "aliases": self.aliases, + "description": self.description, + "country": self.country, + "references": self.references, + "families": self.families, + "attribution_confidence": self.attribution_confidence, + "url": self.url, + "provenance": sorted(self.provenance), + } + + +class ActorProvider: + """Base class for a threat-actor enrichment backend. + + Mirrors FamilyProvider so actor engines slot into the same registry / + processing / template pattern as malware-family engines. + """ + + name = "base" + + def __init__(self, options: Dict): + self.options = options or {} + self.timeout = int(self.options.get("timeout", 20) or 20) + + def available(self) -> bool: + return True + + def resolve(self, query: str) -> Optional[str]: + """Resolve a free-text actor name to this provider's canonical id.""" + raise NotImplementedError + + def fetch(self, actor_id: str, provenance=None) -> Optional[ActorCard]: + """Fetch a card for a canonical actor id.""" + raise NotImplementedError + + def enrich(self, query: str, is_id: bool = False, provenance=None) -> Optional[ActorCard]: + actor_id = query if is_id else self.resolve(query) + if not actor_id: + return None + return self.fetch(actor_id, provenance=provenance) diff --git a/lib/cuckoo/common/integrations/threatintelligence/cache.py b/lib/cuckoo/common/integrations/threatintelligence/cache.py new file mode 100644 index 00000000000..33157b1b36f --- /dev/null +++ b/lib/cuckoo/common/integrations/threatintelligence/cache.py @@ -0,0 +1,69 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Best-effort, on-disk TTL cache for threat-intel lookups. + +Suppresses duplicate API calls for the same key ACROSS analyses (within an +analysis the processing module already dedupes), keeping usage inside +provider fair-use limits. Any cache error degrades to a live lookup, never +to a processing failure. One small JSON file per (kind, provider, key). +""" + +import hashlib +import json +import logging +import os +import time + +log = logging.getLogger(__name__) + + +def _default_cache_dir(): + try: + from lib.cuckoo.common.constants import CUCKOO_ROOT + return os.path.join(CUCKOO_ROOT, "storage", "threatintel_cache") + except Exception: + import tempfile + return os.path.join(tempfile.gettempdir(), "cape_threatintel_cache") + + +class IntelCache: + def __init__(self, enabled=True, ttl=86400, cache_dir=None): + self.enabled = bool(enabled) + self.ttl = int(ttl or 0) + self.cache_dir = cache_dir or _default_cache_dir() + if self.enabled: + try: + os.makedirs(self.cache_dir, exist_ok=True) + except OSError as err: + log.debug("Threat-intel cache disabled (mkdir failed): %s", err) + self.enabled = False + + def _path(self, kind, provider, key): + digest = hashlib.sha256(f"{kind}|{provider}|{key}".encode()).hexdigest() + return os.path.join(self.cache_dir, f"{digest}.json") + + def get(self, kind, provider, key): + if not self.enabled: + return None + try: + with open(self._path(kind, provider, key), "r", encoding="utf-8") as fh: + entry = json.load(fh) + except (OSError, ValueError): + return None + if self.ttl and (time.time() - entry.get("ts", 0)) > self.ttl: + return None + return entry.get("value") + + def set(self, kind, provider, key, value): + if not self.enabled: + return + path = self._path(kind, provider, key) + try: + tmp = f"{path}.{os.getpid()}.tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump({"ts": time.time(), "value": value}, fh) + os.replace(tmp, path) + except OSError as err: + log.debug("Threat-intel cache write failed: %s", err) diff --git a/lib/cuckoo/common/integrations/threatintelligence/config.py b/lib/cuckoo/common/integrations/threatintelligence/config.py new file mode 100644 index 00000000000..4ce001678f0 --- /dev/null +++ b/lib/cuckoo/common/integrations/threatintelligence/config.py @@ -0,0 +1,80 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Loader for conf/threat_intel.conf. + +The file is laid out as one global section plus one section per engine: + + [threatintelligence] global behaviour (timeouts, cache, gates, ...) + [threatfox] engine, "enabled" + engine-local keys + [malpedia] engine, "enabled" + engine-local keys + +Engine sections INHERIT the global section: an engine is handed +``globals | its own section``, so it only needs to declare what differs and +still sees shared settings such as ``timeout``. Because engine keys are +section-scoped they are unprefixed (``api_key``, not ``threatfox_api``), +which is what lets new engines be added without the flat namespace growing. + +Section names come from the registry, so adding an engine is a section here +plus a line in registry.py -- nothing else needs to know about it. +""" + +import logging +from typing import Dict, Iterable, Optional + +log = logging.getLogger(__name__) + +CONFIG_NAME = "threat_intel" +GLOBAL_SECTION = "threatintelligence" + + +class TIConfig: + """Global options plus per-engine sections, with inheritance.""" + + def __init__(self, global_options: Optional[Dict] = None, sections: Optional[Dict] = None): + self.globals: Dict = dict(global_options or {}) + self.sections: Dict[str, Dict] = dict(sections or {}) + + def section(self, name: str) -> Dict: + """Raw options for one engine section (empty when absent).""" + return dict(self.sections.get(name) or {}) + + def provider_options(self, name: str) -> Dict: + """Options handed to an engine: globals overlaid with its section.""" + merged = dict(self.globals) + merged.update(self.section(name)) + return merged + + def is_enabled(self, name: str) -> bool: + from lib.cuckoo.common.integrations.threatintelligence.base import _as_bool + + return _as_bool(self.section(name).get("enabled", False)) + + +def _read_section(cfg, name: str) -> Dict: + """Read one section, tolerating its absence. + + CAPE's Config.get() raises when a section is missing, which is expected + here: an operator may simply not have a section for an engine they do not + use, and that must degrade to "disabled", never to a processing failure. + """ + try: + return dict(cfg.get(name) or {}) + except Exception: + return {} + + +def load_config(section_names: Iterable[str] = ()) -> TIConfig: + """Load conf/threat_intel.conf into a TIConfig. + + Resolution is CAPE's standard chain for this file name: + conf/default/threat_intel.conf.default -> conf/threat_intel.conf -> + conf/threat_intel.conf.d/*.conf -> custom conf dir. + """ + from lib.cuckoo.common.config import Config + + cfg = Config(CONFIG_NAME) + globals_ = _read_section(cfg, GLOBAL_SECTION) + sections = {name: _read_section(cfg, name) for name in section_names} + return TIConfig(globals_, sections) diff --git a/lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py b/lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py new file mode 100644 index 00000000000..8b2b5f5f13e --- /dev/null +++ b/lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py @@ -0,0 +1,196 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Malpedia threat-actor provider — REFERENCE IMPLEMENTATION, OFF by default. + +Implements the actor-enrichment path so the report-page actor card is ready +for real providers. Malpedia's actor data is sourced from community / MISP +curation, so its family<->actor links and attribution-confidence are often +only moderate. Attributing a sample to a named actor is a strong claim, so +this provider stays disabled unless explicitly enabled for a specific, +understood scenario; a dedicated high-confidence actor provider is the +intended driver of actor cards. + +API (all GET, public; optional Authorization: apitoken ): + GET /api/find/actor/ -> resolve a name/synonym to actor id(s) + GET /api/get/actor/ -> actor meta + {value/common_name, description, meta:{synonyms, country, refs, + attribution-confidence, ...}, families:[...]} +""" + +import logging +from typing import List, Optional +from urllib.parse import quote + +from lib.cuckoo.common.integrations.threatintelligence.base import ( + ActorCard, ActorProvider, malpedia_actor_url, reference_label, squash_name, +) + +log = logging.getLogger(__name__) + +DEFAULT_HOST = "https://malpedia.caad.fkie.fraunhofer.de/api" + + +class MalpediaActorProvider(ActorProvider): + name = "malpedia" + + def __init__(self, options): + super().__init__(options) + self.host = (self.options.get("host") or DEFAULT_HOST).strip().rstrip("/") + self.apitoken = (self.options.get("api_key") or "").strip() or None + self.max_references = int(self.options.get("max_references", 4) or 0) + self._session = None + + def available(self) -> bool: + try: + import requests # noqa: F401 + except ImportError: + log.warning("Malpedia actor: 'requests' unavailable (unexpected in CAPE venv).") + return False + return True + + def _get_session(self): + if self._session is None: + import requests + s = requests.Session() + s.headers.update({"Accept": "application/json"}) + if self.apitoken: + s.headers.update({"Authorization": f"apitoken {self.apitoken}"}) + self._session = s + return self._session + + def _get(self, path): + import requests + url = f"{self.host}/{path.lstrip('/')}" + try: + resp = self._get_session().get(url, timeout=self.timeout) + except requests.exceptions.Timeout as err: + raise TimeoutError(f"Malpedia request exceeded {self.timeout}s") from err + if resp.status_code == 404: + return None + if resp.status_code != 200: + log.warning("Malpedia actor HTTP %s for %s", resp.status_code, path) + return None + try: + return resp.json() + except ValueError: + return None + + # -- resolution ------------------------------------------------------- + + def resolve(self, query: str) -> Optional[str]: + try: + data = self._get(f"find/actor/{quote(query.strip(), safe='')}") + except Exception as err: + log.warning("Malpedia actor resolve failed for %r: %s", query, err) + return None + candidates = self._harvest_actor_ids(data) + return self._best_candidate(query, candidates) + + @staticmethod + def _harvest_actor_ids(obj) -> List[str]: + """Collect actor-id-like strings from a loosely-typed find/actor blob. + + Iterative (explicit stack) rather than recursive: the blob comes from + an external API, so a pathologically deep response must not be able to + raise RecursionError. Children are pushed in reverse so that pop() + visits them in document order -- _best_candidate() returns the FIRST + acceptable id, so ordering is behaviourally significant. + """ + found = [] + stack = [obj] + while stack: + node = stack.pop() + if isinstance(node, str): + s = node.strip().lower() + # Actor ids are slugs (apt28, sofacy, lazarus_group); accept + # short slug-ish tokens, exclude obvious sentences/urls. + if s and " " not in s and "/" not in s and "." not in s and len(s) <= 64: + found.append(s) + elif isinstance(node, dict): + for k, v in reversed(list(node.items())): + stack.append(v) + stack.append(k) + elif isinstance(node, (list, tuple, set)): + stack.extend(reversed(list(node))) + seen, out = set(), [] + for f in found: + if f not in seen: + seen.add(f) + out.append(f) + return out + + @staticmethod + def _best_candidate(query, candidates) -> Optional[str]: + if not candidates: + return None + want = squash_name(query) + for cid in candidates: + if squash_name(cid) == want: + return cid + for cid in candidates: + cs = squash_name(cid) + if want and (want in cs or cs in want): + return cid + return None + + # -- fetch ------------------------------------------------------------ + + def fetch(self, actor_id, provenance=None) -> Optional[ActorCard]: + actor_id = actor_id.strip().lower() + try: + data = self._get(f"get/actor/{quote(actor_id, safe='')}") + except Exception as err: + log.warning("Malpedia actor fetch failed for %s: %s", actor_id, err) + return None + if not isinstance(data, dict): + return None + + meta = data.get("meta") if isinstance(data.get("meta"), dict) else {} + common = data.get("value") or data.get("common_name") or actor_id + aliases = [a for a in (meta.get("synonyms") or data.get("synonyms") or []) if a] + refs = self._select_references(meta.get("refs") or data.get("refs") or data.get("urls") or []) + families = self._actor_families(data) + confidence = _to_int(meta.get("attribution-confidence")) + + return ActorCard( + source=self.name, + actor_id=actor_id, + common_name=common, + aliases=aliases, + description=(data.get("description") or meta.get("description") or "").strip(), + country=meta.get("country"), + references=refs, + families=families, + attribution_confidence=confidence, + url=malpedia_actor_url(actor_id), + provenance=provenance, + ) + + def _select_references(self, urls) -> List[dict]: + urls = [u for u in urls if isinstance(u, str) and u.strip()] + if self.max_references and self.max_references > 0: + urls = urls[: self.max_references] + return [{"url": u, "label": reference_label(u)} for u in urls] + + @staticmethod + def _actor_families(data) -> List[str]: + fams = data.get("families") + out = [] + if isinstance(fams, dict): + out = list(fams.keys()) + elif isinstance(fams, list): + for f in fams: + if isinstance(f, str): + out.append(f) + elif isinstance(f, dict): + out.append(f.get("common_name") or f.get("family_id") or f.get("id") or "") + return [f for f in out if f] + + +def _to_int(value) -> Optional[int]: + try: + return int(value) + except (TypeError, ValueError): + return None diff --git a/lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py b/lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py new file mode 100644 index 00000000000..a93e08c5229 --- /dev/null +++ b/lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py @@ -0,0 +1,277 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Malpedia family-enrichment provider — direct REST client. + +Uses only ``requests`` (already in CAPE). No third-party Malpedia client. + +API (https://malpedia.caad.fkie.fraunhofer.de/usage/api), all GET, the +endpoints we use are public (no auth); an optional APIToken raises rate +limits and is sent as ``Authorization: apitoken ``: + + GET /api/find/family/ -> resolve a name/alias to family id(s) + GET /api/get/family/ -> family metadata + {common_name, description, alt_names[], attribution[], urls[], updated} + +ThreatFox already hands us canonical ids (win.cobalt_strike), so those skip +resolution; CAPE's own YARA/detection family names are resolved via +find/family. Focus is malware families. +""" + +import logging +import re +from typing import List, Optional +from urllib.parse import quote + +from lib.cuckoo.common.integrations.threatintelligence.base import ( + FamilyCard, FamilyProvider, looks_like_malpedia_id, malpedia_details_url, + reference_label, squash_name, +) + +log = logging.getLogger(__name__) + +DEFAULT_HOST = "https://malpedia.caad.fkie.fraunhofer.de/api" + +# Malpedia platform prefixes — used to harvest ids from loosely-typed +# find/family responses without over-matching arbitrary "a.b" strings. +_PLATFORMS = ("win", "elf", "apk", "osx", "jar", "js", "py", "vbs", "ps1", + "swf", "sh", "asp", "php", "pl", "rb", "go", "symbian", "ios") + + +class MalpediaProvider(FamilyProvider): + name = "malpedia" + + def __init__(self, options): + super().__init__(options) + self.host = (self.options.get("host") or DEFAULT_HOST).strip().rstrip("/") + self.apitoken = (self.options.get("api_key") or "").strip() or None + self.max_references = int(self.options.get("max_references", 4) or 0) + self._session = None + + def available(self) -> bool: + try: + import requests # noqa: F401 + except ImportError: + log.warning("Malpedia: 'requests' unavailable (unexpected in CAPE venv).") + return False + return True + + def _get_session(self): + if self._session is None: + import requests + s = requests.Session() + s.headers.update({"Accept": "application/json"}) + if self.apitoken: + s.headers.update({"Authorization": f"apitoken {self.apitoken}"}) + self._session = s + return self._session + + # -- HTTP ------------------------------------------------------------- + + def _get(self, path): + import requests + url = f"{self.host}/{path.lstrip('/')}" + try: + resp = self._get_session().get(url, timeout=self.timeout) + except requests.exceptions.Timeout as err: + raise TimeoutError(f"Malpedia request exceeded {self.timeout}s") from err + if resp.status_code == 404: + return None + if resp.status_code != 200: + log.warning("Malpedia HTTP %s for %s", resp.status_code, path) + return None + try: + return resp.json() + except ValueError: + log.warning("Malpedia returned non-JSON for %s", path) + return None + + def _get_raw(self, path): + """GET returning the raw response text (used for the .bib endpoint).""" + import requests + url = f"{self.host}/{path.lstrip('/')}" + try: + resp = self._get_session().get(url, timeout=self.timeout) + except requests.exceptions.Timeout as err: + raise TimeoutError(f"Malpedia request exceeded {self.timeout}s") from err + if resp.status_code != 200: + return None + return resp.text + + # -- resolution ------------------------------------------------------- + + def resolve(self, query: str) -> Optional[str]: + if looks_like_malpedia_id(query): + return query.strip().lower() + try: + data = self._get(f"find/family/{quote(query.strip(), safe='')}") + except Exception as err: + log.warning("Malpedia resolve failed for %r: %s", query, err) + return None + candidates = self._harvest_ids(data) + return self._best_candidate(query, candidates) + + @classmethod + def _harvest_ids(cls, obj) -> List[str]: + """Collect Malpedia family ids from a loosely-typed blob. + + Iterative (explicit stack) rather than recursive: the blob comes from + an external API, so a pathologically deep response must not be able to + raise RecursionError. Children are pushed in reverse so that pop() + visits them in document order -- _best_candidate() returns the FIRST + acceptable id, so ordering is behaviourally significant. + """ + found = [] + stack = [obj] + while stack: + node = stack.pop() + if isinstance(node, str): + s = node.strip().lower() + if "." in s and s.split(".", 1)[0] in _PLATFORMS and looks_like_malpedia_id(s): + found.append(s) + elif isinstance(node, dict): + for k, v in reversed(list(node.items())): + stack.append(v) + stack.append(k) + elif isinstance(node, (list, tuple, set)): + stack.extend(reversed(list(node))) + # preserve order, de-dup + seen, out = set(), [] + for f in found: + if f not in seen: + seen.add(f) + out.append(f) + return out + + @staticmethod + def _best_candidate(query, candidates) -> Optional[str]: + if not candidates: + return None + want = squash_name(query) + # 1) exact match on the family part (after the platform dot) + for cid in candidates: + if squash_name(cid.split(".", 1)[-1]) == want: + return cid + # 2) exact match on the whole id squashed + for cid in candidates: + if squash_name(cid) == want: + return cid + # 3) family part contains the query (or vice-versa) + for cid in candidates: + fam = squash_name(cid.split(".", 1)[-1]) + if want and (want in fam or fam in want): + return cid + # 4) give up on ambiguity rather than guess wrong + return None + + # -- fetch ------------------------------------------------------------ + + def fetch(self, family_id, provenance=None) -> Optional[FamilyCard]: + family_id = family_id.strip().lower() + try: + data = self._get(f"get/family/{quote(family_id, safe='.')}") + except Exception as err: + log.warning("Malpedia fetch failed for %s: %s", family_id, err) + return None + if not isinstance(data, dict): + return None + + references = self._select_references(data.get("urls") or [], family_id) + attribution = self._parse_attribution(data.get("attribution") or []) + common = data.get("common_name") or family_id.split(".", 1)[-1] + + return FamilyCard( + source=self.name, + family_id=family_id, + common_name=common, + aliases=[a for a in (data.get("alt_names") or []) if a], + description=(data.get("description") or "").strip(), + references=references, + attribution=attribution, + updated=data.get("updated"), + url=malpedia_details_url(family_id), + provenance=provenance, + ) + + def _select_references(self, urls, family_id) -> List[dict]: + """Order references by real publication date (most recent first). + + Malpedia's get/family ``urls`` is a flat list with NO dates and is + not chronologically ordered, so we cannot infer recency from it. + The per-family BibTeX endpoint (get/bib/family/) does carry a + date/year per reference, so we use it to sort. If the bib is + unavailable, we fall back to the native list order (best effort) and + omit the recency claim by leaving dates blank. + """ + urls = [u for u in urls if isinstance(u, str) and u.strip()] + dates = {} + try: + raw = self._get_raw(f"get/bib/family/{quote(family_id, safe='.')}") + if raw: + dates = _parse_bib_dates(raw) + except Exception as err: + log.debug("Malpedia bib lookup failed for %s: %s", family_id, err) + + if dates: + dated = sorted((u for u in urls if u in dates), key=lambda u: dates[u], reverse=True) + undated = [u for u in urls if u not in dates] + ordered = dated + undated + else: + # No dates available: keep Malpedia's native order rather than + # pretending to know recency. + ordered = urls + + if self.max_references and self.max_references > 0: + ordered = ordered[: self.max_references] + return [{"url": u, "label": reference_label(u), "date": dates.get(u)} for u in ordered] + + @staticmethod + def _parse_attribution(attribution) -> List[str]: + names = [] + for item in attribution: + if isinstance(item, dict): + for key in ("value", "common_name", "name", "actor"): + if item.get(key): + names.append(str(item[key])) + break + elif isinstance(item, str) and item.strip(): + names.append(item.strip()) + # de-dup, keep order + seen, out = set(), [] + for n in names: + if n.lower() not in seen: + seen.add(n.lower()) + out.append(n) + return out + + +# BibTeX field extractors (tolerant of {..} or "..", any whitespace/case). +_BIB_URL = re.compile(r'\burl\s*=\s*[{"]([^}"]+)[}"]', re.I) +_BIB_DATE = re.compile(r'\bdate\s*=\s*[{"]([^}"]+)[}"]', re.I) +_BIB_YEAR = re.compile(r'\byear\s*=\s*[{"]?\s*(\d{4})', re.I) + + +def _parse_bib_dates(text: str) -> dict: + """Map reference URL -> date string from a Malpedia family .bib file. + + Returns ISO-ish date strings ("2022-05-01") or bare years ("2022"), + which sort lexicographically in chronological order. Best-effort: any + entry without both a url and a date is skipped. + """ + out = {} + for entry in re.split(r'\n@', text or ""): + m_url = _BIB_URL.search(entry) + if not m_url: + continue + url = m_url.group(1).strip() + m_date = _BIB_DATE.search(entry) + date = m_date.group(1).strip() if m_date else None + if not date: + m_year = _BIB_YEAR.search(entry) + date = m_year.group(1) if m_year else None + if url and date: + prev = out.get(url) + if prev is None or date > prev: + out[url] = date + return out diff --git a/lib/cuckoo/common/integrations/threatintelligence/registry.py b/lib/cuckoo/common/integrations/threatintelligence/registry.py new file mode 100644 index 00000000000..28647720df3 --- /dev/null +++ b/lib/cuckoo/common/integrations/threatintelligence/registry.py @@ -0,0 +1,96 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Provider registry for the threat-intelligence framework. + +Two kinds of provider live here: + * indicator providers — ip/domain/sha256 -> context (ThreatFox, ...) + * family providers — malware family -> card (Malpedia, ...) + +Provider modules are imported LAZILY so a missing/broken optional +dependency cannot crash CAPE's plugin loader when a provider is disabled. + +Each provider owns a same-named SECTION in conf/threat_intel.conf, holding +"enabled" plus its engine-local keys; the global [threatintelligence] section +is inherited underneath it. + +Adding a provider: + 1. Implement the subclass (lazy-import deps inside available()/lookup()). + 2. Add it to _INDICATOR_MODULES / _FAMILY_MODULES / _ACTOR_MODULES below. + 3. Add a "[]" section (with "enabled") to conf/threat_intel.conf. +Nothing else needs to change: the section name is taken from the registry. +""" + +import importlib +import logging +from typing import Dict, List + +from lib.cuckoo.common.integrations.threatintelligence.base import ( + ActorProvider, FamilyProvider, IndicatorProvider, +) + +log = logging.getLogger(__name__) + +# Register additional indicator providers here, e.g.: +# "": "lib.cuckoo.common.integrations.threatintelligence..", +_INDICATOR_MODULES: Dict[str, str] = { + "threatfox": "lib.cuckoo.common.integrations.threatintelligence.threatfox_provider.ThreatFoxProvider", +} + +# Register additional family providers here (same dotted-path pattern). +_FAMILY_MODULES: Dict[str, str] = { + "malpedia": "lib.cuckoo.common.integrations.threatintelligence.malpedia_provider.MalpediaProvider", +} + +# Threat-actor engines. Enabled by "_actors = yes" AND the master +# "threat actors = yes" gate. Actor attribution must be high confidence, so +# the Malpedia reference engine is OFF by default (community/MISP sourced). +# Register a high-confidence actor provider here using the same pattern. +_ACTOR_MODULES: Dict[str, str] = { + "malpedia_actors": "lib.cuckoo.common.integrations.threatintelligence.malpedia_actor_provider.MalpediaActorProvider", +} + + +def _load(dotted: str): + module_path, class_name = dotted.rsplit(".", 1) + return getattr(importlib.import_module(module_path), class_name) + + +def all_provider_names() -> List[str]: + """Every registered engine name = every engine section in threat_intel.conf.""" + return list(_INDICATOR_MODULES) + list(_FAMILY_MODULES) + list(_ACTOR_MODULES) + + +def _enabled(modules: Dict[str, str], config): + """Instantiate the enabled engines from `modules`. + + `config` is a TIConfig: each engine is enabled by "enabled" in its own + section and receives globals overlaid with that section. + """ + out = [] + for name, dotted in modules.items(): + if not config.is_enabled(name): + continue + try: + provider = _load(dotted)(config.provider_options(name)) + except Exception as err: + log.warning("Threat-intel provider '%s' failed to load: %s", name, err) + continue + if not provider.available(): + log.warning("Threat-intel provider '%s' enabled but unavailable; skipping.", name) + continue + out.append(provider) + return out + + +def get_enabled_indicator_providers(config) -> List[IndicatorProvider]: + return _enabled(_INDICATOR_MODULES, config) + + +def get_enabled_family_providers(config) -> List[FamilyProvider]: + return _enabled(_FAMILY_MODULES, config) + + +def get_enabled_actor_providers(config) -> List[ActorProvider]: + return _enabled(_ACTOR_MODULES, config) diff --git a/lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py b/lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py new file mode 100644 index 00000000000..47095cdc4c9 --- /dev/null +++ b/lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py @@ -0,0 +1,298 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""ThreatFox (abuse.ch) indicator provider — direct REST client. + +Uses only ``requests`` (already in CAPE); no third-party ThreatFox client. + +API (https://threatfox.abuse.ch/api/), single endpoint, POST JSON: + POST https://threatfox-api.abuse.ch/api/v1/ + headers: Auth-Key: # REQUIRED since 2024 + body: {"query":"search_ioc","search_term":"","exact_match":bool} + body: {"query":"search_hash","hash":""} +Responses: {"query_status":"ok"|"no_result"|..., "data":[ {ioc fields} ]} + +Each ThreatFox row carries ``malware`` (a Malpedia family id, e.g. +win.cobalt_strike) — the processing module feeds those ids straight to the +family providers, so an infrastructure hit also yields a malware card. +""" + +import logging +from typing import List, Optional + +from lib.cuckoo.common.integrations.threatintelligence.base import ( + IND_DOMAIN, IND_HASH, IND_IP, IND_URL, + IntelMatch, IndicatorProvider, ProviderResult, + _as_bool, ioc_host_part, normalize_domain, url_match_key, +) + +log = logging.getLogger(__name__) + +DEFAULT_HOST = "https://threatfox-api.abuse.ch/api/v1/" + + +class ThreatFoxProvider(IndicatorProvider): + name = "threatfox" + supported_indicators = {IND_IP, IND_DOMAIN, IND_URL, IND_HASH} + + def __init__(self, options): + super().__init__(options) + self.host = (self.options.get("host") or DEFAULT_HOST).strip() + self.apikey = (self.options.get("api_key") or "").strip() or None + self.exact_match = _as_bool(self.options.get("exact_match", False)) + self.match_ports_only = _as_bool(self.options.get("match_ports_only", False)) + # When True, an IOC only matches an indicator of the SAME type. When + # False (default), url IOCs also match a contacted domain/IP by host, + # except on dead-drop-resolver hosts (auto-detected below). + self.strict_type_match = _as_bool(self.options.get("strict_ioc_type_match", False)) + # Rate-limit / transient-error resilience: retry on HTTP 429 / 5xx / + # timeout with exponential backoff (honours Retry-After when present). + self.retries = int(self.options.get("retries", 3) or 0) + self.backoff = float(self.options.get("backoff", 0.5) or 0.0) + self._session = None + + def available(self) -> bool: + try: + import requests # noqa: F401 + except ImportError: + log.warning("ThreatFox: 'requests' unavailable (unexpected in CAPE venv).") + return False + if not self.apikey: + log.warning( + "ThreatFox enabled but api_key (Auth-Key) is empty in the [threatfox] " + "section of threat_intel.conf. abuse.ch rejects unauthenticated " + "requests; get a free key at https://auth.abuse.ch/ . Provider skipped." + ) + return False + return True + + def _get_session(self): + if self._session is None: + import requests + s = requests.Session() + s.headers.update({"Auth-Key": self.apikey, "Accept": "application/json"}) + self._session = s + return self._session + + def lookup(self, indicator, indicator_type, ports=None) -> ProviderResult: + if not self.apikey: + return ProviderResult(status="disabled") + try: + if indicator_type == IND_HASH: + payload = {"query": "search_hash", "hash": indicator} + else: + payload = {"query": "search_ioc", "search_term": indicator, + "exact_match": bool(self.exact_match)} + data = self._post(payload) + except TimeoutError as err: + return ProviderResult(status="timeout", error=str(err)) + except Exception as err: + log.warning("ThreatFox lookup failed for %s: %s", indicator, err) + return ProviderResult(status="error", error=str(err)) + + if data is None: + return ProviderResult(status="error", error="no/invalid response") + + out = ProviderResult(status="ok") + parsed = self._parse(data, indicator, indicator_type, ports) + parsed = self._suppress_dead_drop(parsed, indicator, indicator_type) + out.matches = self._select(parsed) + if not out.matches: + out.status = "no_match" + return out + + def _post(self, payload): + import random + import time + + import requests + + attempt = 0 + while True: + try: + resp = self._get_session().post(self.host, json=payload, timeout=self.timeout) + except requests.exceptions.Timeout as err: + if attempt < self.retries: + self._sleep_backoff(attempt) + attempt += 1 + continue + raise TimeoutError(f"ThreatFox request exceeded {self.timeout}s") from err + except requests.exceptions.RequestException: + if attempt < self.retries: + self._sleep_backoff(attempt) + attempt += 1 + continue + raise + + # Rate limited / transient server error: back off and retry so a + # burst of lookups doesn't silently drop enrichment for some hosts. + if resp.status_code == 429 or 500 <= resp.status_code < 600: + if attempt < self.retries: + delay = self._retry_after(resp) + if delay is None: + delay = self.backoff * (2 ** attempt) + random.uniform(0, 0.25) + log.info("ThreatFox HTTP %s; retrying in %.1fs (%d/%d)", + resp.status_code, delay, attempt + 1, self.retries) + time.sleep(delay) + attempt += 1 + continue + log.warning("ThreatFox HTTP %s after %d retries for query=%s", + resp.status_code, self.retries, payload.get("query")) + return None + + if resp.status_code != 200: + log.warning("ThreatFox HTTP %s for query=%s", resp.status_code, payload.get("query")) + return None + + try: + body = resp.json() + except ValueError: + log.warning("ThreatFox returned non-JSON response.") + return None + status = (body or {}).get("query_status") + if status == "ok": + data = body.get("data") + return data if isinstance(data, list) else [] + if status in ("no_result", "no_results"): + return [] + if status in ("illegal_auth_key", "unauthorized", "missing_auth_key"): + log.warning("ThreatFox auth rejected (query_status=%s); check [threatfox] api_key.", status) + return None + log.debug("ThreatFox query_status=%s (treated as no match)", status) + return [] + + def _sleep_backoff(self, attempt): + import random + import time + time.sleep(self.backoff * (2 ** attempt) + random.uniform(0, 0.25)) + + @staticmethod + def _retry_after(resp): + val = resp.headers.get("Retry-After") + if not val: + return None + try: + return float(val) + except (TypeError, ValueError): + return None + + def _parse(self, data, indicator, indicator_type, ports) -> List[IntelMatch]: + matches: List[IntelMatch] = [] + want_ports = self._normalize_ports(ports) if self.match_ports_only else None + for row in data: + if not isinstance(row, dict): + continue + ioc = row.get("ioc") or "" + ioc_type = row.get("ioc_type") or "" + if not self._ioc_matches(ioc, ioc_type, indicator, indicator_type): + continue + if want_ports is not None and not self._port_matches(ioc, want_ports): + continue + ioc_id = row.get("id") + indicator_url = f"https://threatfox.abuse.ch/ioc/{ioc_id}/" if ioc_id else None + matches.append(IntelMatch( + source=self.name, indicator=indicator, indicator_type=indicator_type, + ioc=ioc, ioc_id=ioc_id, indicator_url=indicator_url, + threat_type=row.get("threat_type"), + threat_type_desc=row.get("threat_type_desc"), + ioc_type=row.get("ioc_type"), ioc_type_desc=row.get("ioc_type_desc"), + malware=row.get("malware"), malware_printable=row.get("malware_printable"), + malware_alias=row.get("malware_alias"), malware_malpedia=row.get("malware_malpedia"), + confidence_level=_to_int(row.get("confidence_level")), + first_seen=row.get("first_seen"), last_seen=row.get("last_seen"), + reference=row.get("reference"), reporter=row.get("reporter"), + tags=row.get("tags") or [], + )) + return matches + + # IOC matching. By default an IOC matches an indicator of its own type, + # AND a `url` IOC also matches a contacted domain/IP by HOST -- so a + # dedicated malicious domain stored on ThreatFox as a url (e.g. a Lumma C2 + # like https://curtainjors.fun/api) still tags the domain. The + # dead-drop-resolver false positive (a legitimate shared host like + # steamcommunity.com carrying url IOCs for many families) is handled + # separately by _suppress_dead_drop, which keys off family multiplicity + # rather than bluntly dropping all url IOCs. Set strict_ioc_type_match to + # require same-type matching only. + _IP_IOC_TYPES = {"ip:port", "ip"} + _HASH_IOC_TYPES = {"md5_hash", "sha256_hash", "sha1_hash", "sha384_hash", "sha512_hash"} + + def _ioc_matches(self, ioc, ioc_type, indicator, indicator_type) -> bool: + ioc_type = (ioc_type or "").strip().lower() + + if indicator_type == IND_HASH: + return (not ioc_type) or ioc_type in self._HASH_IOC_TYPES + + if indicator_type == IND_URL: + if ioc_type and ioc_type != "url": + return False + return url_match_key(ioc) == url_match_key(indicator) + + if indicator_type == IND_DOMAIN: + host_ok = ioc_host_part(ioc) == normalize_domain(indicator) + if ioc_type == "domain" or not ioc_type: + return host_ok + if ioc_type == "url" and not self.strict_type_match: + return host_ok # host-level url match; dead-drop guarded later + return False + + if indicator_type == IND_IP: + host_ok = ioc_host_part(ioc) == indicator.strip().lower() + if ioc_type in self._IP_IOC_TYPES or not ioc_type: + return host_ok + if ioc_type == "url" and not self.strict_type_match: + return host_ok + return False + + return False + + def _suppress_dead_drop(self, matches, indicator, indicator_type): + """Drop host-level url matches on dead-drop-resolver hosts. + + A url IOC matched a domain/IP by host (not exact path). If such url + IOCs on this single host span MORE THAN ONE malware family, the host + is almost certainly a legitimate shared service abused as a dead drop + resolver (steamcommunity.com, t.me, pastebin, ...), so those host-level + url matches are false positives and are removed. A single-family host + (a dedicated malicious domain) is kept. Same-type matches (domain/ip + IOCs) are always kept. + """ + if indicator_type not in (IND_DOMAIN, IND_IP): + return matches + url_hits = [m for m in matches if (m.ioc_type or "").strip().lower() == "url"] + if not url_hits: + return matches + families = {(m.malware or m.malware_printable or "").lower() for m in url_hits} + families.discard("") + if len(families) > 1: + log.info("ThreatFox: %s looks like a dead-drop resolver (url IOCs for %d " + "families); dropping host-level url matches.", indicator, len(families)) + return [m for m in matches if (m.ioc_type or "").strip().lower() != "url"] + return matches + + @staticmethod + def _normalize_ports(ports): + out = set() + for p in ports or []: + try: + out.add(int(p)) + except (TypeError, ValueError): + continue + return out + + @staticmethod + def _port_matches(ioc, want_ports) -> bool: + host = ioc.split("://", 1)[-1].split("/", 1)[0] + if host.count(":") == 1: + _, _, tail = host.rpartition(":") + if tail.isdigit(): + return int(tail) in want_ports + return True + + +def _to_int(value) -> Optional[int]: + try: + return int(value) + except (TypeError, ValueError): + return None diff --git a/modules/processing/threatintelligence.py b/modules/processing/threatintelligence.py new file mode 100644 index 00000000000..8121357599c --- /dev/null +++ b/modules/processing/threatintelligence.py @@ -0,0 +1,572 @@ +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Threat-intelligence processing module. + +Runs late (order=22) so network analysis and family detections are already +populated. Two phases: + +1. Indicator phase (ThreatFox, ...): looks up contacted host IPs / domains + (and optionally the sample sha256) per the global indicator toggles, + annotates network hosts/domains in-place with red tags, and harvests the + Malpedia family ids that those infrastructure hits reference. + +2. Family phase (Malpedia, ...): takes the malware families surfaced by + CAPE's own detections / malfamily AND by the indicator hits, fetches a + descriptive card for each (name, aliases, description, recent reference + reports), and writes them for the collapsible "Threat Intelligence" + report-page section. + +Output (results["threatintelligence"]): + { + "providers": {"indicator": ["threatfox"], "family": ["malpedia"]}, + "indicators": {"by_indicator": {...}, "count": N}, + "families": [ {family_id, common_name, aliases, description, + references:[{url,label}], attribution, url, + provenance:[...]}, ... ], + "stats": {...}, + } +""" + +import logging +from concurrent.futures import ThreadPoolExecutor + +from lib.cuckoo.common.abstracts import Processing +from lib.cuckoo.common.integrations.threatintelligence.base import ( + IND_DOMAIN, IND_HASH, IND_IP, IND_URL, _as_bool, looks_like_malpedia_id, + normalize_domain, squash_name, +) +from lib.cuckoo.common.integrations.threatintelligence.cache import IntelCache +from lib.cuckoo.common.integrations.threatintelligence.config import load_config +from lib.cuckoo.common.integrations.threatintelligence.registry import ( + all_provider_names, + get_enabled_actor_providers, get_enabled_family_providers, get_enabled_indicator_providers, +) + +log = logging.getLogger(__name__) + + +def _looks_like_ip(value): + """Cheap check: is this host an IP literal (so it's not a domain)?""" + if not value: + return False + v = str(value).strip().strip("[]") + if ":" in v and "." not in v: + return True # IPv6 + parts = v.split(".") + return len(parts) == 4 and all(p.isdigit() and 0 <= int(p) <= 255 for p in parts) + + +class ThreatIntelligence(Processing): + """Enrich an analysis with external threat intelligence.""" + + order = 22 + + def run(self): + self.key = "threatintelligence" + + try: + config = load_config(all_provider_names()) + except Exception as err: + log.warning("Could not read threat_intel.conf: %s", err) + return {} + # Global behaviour; each engine additionally gets its own section + # overlaid on top of these (see TIConfig.provider_options). + options = config.globals + + indicator_providers = get_enabled_indicator_providers(config) + family_providers = get_enabled_family_providers(config) + # Threat-actor engines are additionally gated by a master "threat + # actors" toggle (default off) because actor attribution is a strong, + # high-confidence-only claim. + actor_enabled = _as_bool(options.get("threat actors", False)) + actor_providers = get_enabled_actor_providers(config) if actor_enabled else [] + if not indicator_providers and not family_providers and not actor_providers: + return {} + + self.cache = IntelCache( + enabled=_as_bool(options.get("cache", True)), + ttl=int(options.get("cache_ttl", 86400) or 0), + ) + self.concurrent = _as_bool(options.get("concurrent_lookups", True)) + self.max_workers = int(options.get("max_workers", 4) or 4) + self.max_results = int(options.get("max_results", 5) or 0) + # Safety valve for shared/abused infrastructure: if one indicator + # matches more than this many distinct families it is treated as an + # ambiguous shared host and its tags are suppressed. 0 = no limit. + self.max_families_per_indicator = int(options.get("max_families_per_indicator", 5) or 0) + + # Global indicator-type toggles (his requested layout). + lookups = { + IND_IP: _as_bool(options.get("ip addresses", True)), + IND_DOMAIN: _as_bool(options.get("domains", True)), + IND_URL: _as_bool(options.get("urls", False)), + IND_HASH: _as_bool(options.get("sha256", False)), + } + + by_indicator, discovered_families = {}, [] + if indicator_providers: + by_indicator, discovered_families = self._indicator_phase( + indicator_providers, lookups) + + # Optionally promote indicator hits to CAPE detections. This APPENDS a + # detection entry (via add_family_detection) and never sets the + # headline malfamily, so ThreatFox infrastructure attribution does not + # override CAPE's own classification. Off by default; C2-only by + # default when enabled. + promoted = [] + if by_indicator and _as_bool(options.get("promote_to_detection", False)): + promoted = self._promote_detections(by_indicator, options) + + families = [] + if family_providers: + from_threatfox = _as_bool(config.section("malpedia").get("from_threatfox", True)) + hints = self._gather_family_hints(discovered_families if from_threatfox else []) + families = self._family_phase(family_providers, hints) + + # Actor phase (gated): only when master toggle on AND an actor engine + # is enabled. Hints are the actors that the matched families are + # attributed to (and, in future, explicit high-confidence provider + # attributions). Off by default. + actors = [] + if actor_providers: + actor_hints = self._gather_actor_hints(families, options) + actors = self._actor_phase(actor_providers, actor_hints) + + if not by_indicator and not families and not actors: + return {} + + return { + "providers": { + "indicator": [p.name for p in indicator_providers], + "family": [p.name for p in family_providers], + "actor": [p.name for p in actor_providers], + }, + "indicators": {"by_indicator": by_indicator, "count": len(by_indicator)}, + "families": families, + "actors": actors, + "promoted_detections": promoted, + "stats": { + "indicators_with_intel": len(by_indicator), + "families": len(families), + "actors": len(actors), + }, + } + + # ================================================================ + # Indicator phase + # ================================================================ + def _indicator_phase(self, providers, lookups): + network = self.results.get("network") or {} + if not isinstance(network, dict): + network = {} + + ip_ports, domains, urls = {}, set(), set() + if lookups[IND_IP]: + for host in network.get("hosts") or []: + if isinstance(host, dict) and host.get("ip"): + ip = str(host["ip"]).strip() + ip_ports.setdefault(ip, set()).update(host.get("ports") or []) + # Accurate ip:port construction: ports actually contacted in the + # capture (TCP/UDP destinations). These drive precise ip:port + # matching when match_ports_only is enabled. + for proto in ("tcp", "udp"): + for conn in network.get(proto) or []: + if isinstance(conn, dict) and conn.get("dst") and conn.get("dport"): + ip_ports.setdefault(str(conn["dst"]).strip(), set()).add(conn["dport"]) + if lookups[IND_DOMAIN]: + # Gather domains from EVERY network surface so nothing contacted is + # skipped: resolved domains, DNS requests + answers, HTTP Host + # headers, reverse-resolved host names, and TLS SNI. + for d in network.get("domains") or []: + if isinstance(d, dict) and d.get("domain"): + domains.add(normalize_domain(d["domain"])) + for req in network.get("dns") or []: + if isinstance(req, dict): + if req.get("request"): + domains.add(normalize_domain(req["request"])) + for ans in req.get("answers") or []: + data = ans.get("data") if isinstance(ans, dict) else None + if data and (ans.get("type") in ("CNAME", "NS", "PTR")): + domains.add(normalize_domain(data)) + for h in network.get("http") or []: + if isinstance(h, dict) and h.get("host") and not _looks_like_ip(h["host"]): + domains.add(normalize_domain(h["host"])) + for host in network.get("hosts") or []: + if isinstance(host, dict) and host.get("hostname"): + domains.add(normalize_domain(host["hostname"])) + for tls in network.get("tls") or []: + if isinstance(tls, dict) and tls.get("sni"): + domains.add(normalize_domain(tls["sni"])) + if lookups[IND_URL]: + # Full URLs reconstructed from HTTP traffic (network.http already + # builds scheme://host[:port]/path in the "uri" field). + for h in network.get("http") or []: + if isinstance(h, dict) and h.get("uri"): + urls.add(str(h["uri"]).strip()) + domains.discard("") + urls.discard("") + + indicators = [(ip, IND_IP, sorted(ports)) for ip, ports in ip_ports.items()] + indicators += [(dom, IND_DOMAIN, None) for dom in domains] + indicators += [(url, IND_URL, None) for url in urls] + if lookups[IND_HASH]: + sha = self._sample_sha256() + if sha: + indicators.append((sha, IND_HASH, None)) + + if not indicators: + return {}, [] + + tasks = [(p, ind, it, pt) for p in providers for (ind, it, pt) in indicators + if p.accepts_indicator(it)] + if not tasks: + return {}, [] + + log.info("ThreatIntelligence: %d indicator lookup(s) via %s", + len(tasks), ", ".join(p.name for p in providers)) + results_list = self._run_tasks(tasks, self._lookup_indicator) + + by_indicator = {} + for (_p, indicator, _it, _pt), match_dicts in zip(tasks, results_list): + if match_dicts: + by_indicator.setdefault(indicator, []).extend(match_dicts) + for indicator, items in list(by_indicator.items()): + # Suppress shared/abused-infrastructure over-matches (a single + # indicator attributed to many families) as likely false tags. + families = {it.get("malware") or it.get("tag") for it in items} + if self.max_families_per_indicator and len(families) > self.max_families_per_indicator: + log.info("ThreatIntelligence: suppressing %s -> %d families " + "(likely shared/abused host, not asserting tags)", + indicator, len(families)) + del by_indicator[indicator] + continue + by_indicator[indicator] = self._dedup_cap(items, self.max_results) + + # Annotate network dicts in-place for the web UI red tags. + self._annotate_hosts(network, by_indicator) + self._annotate_domains(network, by_indicator) + self._annotate_urls(network, by_indicator) + + # Harvest discovered families (Malpedia ids) from the hits, carrying the + # indicator type so provenance can record WHERE it was seen + # (threatfox_domain, threatfox_ip_address, ...). + discovered = [] + for items in by_indicator.values(): + for it in items: + fam_id = it.get("malware") + if fam_id: + discovered.append((fam_id, it.get("malware_printable"), + it.get("source"), it.get("indicator_type"))) + return by_indicator, discovered + + def _lookup_indicator(self, task): + provider, indicator, indicator_type, ports = task + ck = f"{indicator_type}:{indicator}" + cached = self.cache.get("indicator", provider.name, ck) + if cached is not None: + return cached + try: + result = provider.lookup(indicator, indicator_type, ports=ports) + except Exception: + log.exception("ThreatIntelligence: %s crashed on %s", provider.name, indicator) + return [] + match_dicts = [m.to_dict() for m in result.matches] + if result.status in ("ok", "no_match"): + self.cache.set("indicator", provider.name, ck, match_dicts) + return match_dicts + + @staticmethod + def _dedup_cap(items, max_results): + best = {} + for it in items: + key = f"{it.get('source')}|{it.get('tag')}" + cur = best.get(key) + if cur is None or (it.get("confidence_level") or 0) > (cur.get("confidence_level") or 0): + best[key] = it + ordered = sorted(best.values(), key=lambda d: (d.get("confidence_level") or 0), reverse=True) + return ordered[:max_results] if max_results and max_results > 0 else ordered + + def _annotate_hosts(self, network, by_indicator): + for host in network.get("hosts") or []: + if isinstance(host, dict): + hits = by_indicator.get(str(host.get("ip") or "").strip()) + if hits: + host["threatintel"] = hits + + def _annotate_domains(self, network, by_indicator): + for req in network.get("dns") or []: + if isinstance(req, dict): + hits = by_indicator.get(normalize_domain(req.get("request"))) + if hits: + req["threatintel"] = hits + for d in network.get("domains") or []: + if isinstance(d, dict): + hits = by_indicator.get(normalize_domain(d.get("domain"))) + if hits: + d["threatintel"] = hits + + def _annotate_urls(self, network, by_indicator): + for h in network.get("http") or []: + if isinstance(h, dict): + hits = by_indicator.get(str(h.get("uri") or "").strip()) + if hits: + h["threatintel"] = hits + + def _sample_sha256(self): + target = self.results.get("target") or {} + f = target.get("file") if isinstance(target, dict) else None + if isinstance(f, dict): + return f.get("sha256") + return None + + # ================================================================ + # Family phase + # ================================================================ + def _gather_family_hints(self, discovered_families): + """Build de-duplicated family hints from detections + indicator hits. + + Each hint: {"query","is_id","provenance":set}. Dedup key is the + Malpedia id (when known) or the squashed family name. + """ + hints = {} + + def hint_key(query, is_id): + # Key on the squashed family-part so a free-text detection name + # ("CastleStealer") and a canonical id ("win.castle_stealer") + # collapse to ONE hint -> one Malpedia lookup. The platform is + # dropped because names carry none; same-family-part across + # platforms (rare) intentionally merges. + base_part = query.split(".", 1)[-1] if is_id else query + return squash_name(base_part) + + def add(query, is_id, source): + if not query: + return + key = hint_key(query, is_id) + if not key: + return + entry = hints.get(key) + if entry is None: + hints[key] = {"query": query.strip(), "is_id": is_id, "provenance": {source}} + else: + entry["provenance"].add(source) + # Prefer a canonical id over a free-text name for the same family. + if is_id and not entry["is_id"]: + entry["query"], entry["is_id"] = query.strip(), True + + # CAPE detections (list of blocks, or a bare string). + detections = self.results.get("detections") + if isinstance(detections, str): + add(detections, False, "cape_detection") + elif isinstance(detections, list): + for block in detections: + if isinstance(block, dict) and block.get("family"): + add(block["family"], False, "cape_detection") + elif isinstance(block, str): + add(block, False, "cape_detection") + + # Top-level malfamily hints. + for key in ("malfamily", "malfamily_tag"): + val = self.results.get(key) + if isinstance(val, str) and val: + add(val, False, "cape_detection") + + # Families surfaced by indicator hits, tagged with WHERE seen, e.g. + # threatfox_domain / threatfox_ip_address / threatfox_url. + for fam_id, printable, source, indicator_type in discovered_families: + prov = self._indicator_provenance(source or "threatfox", indicator_type) + if looks_like_malpedia_id(fam_id): + add(fam_id, True, prov) + elif printable: + add(printable, False, prov) + + return list(hints.values()) + + @staticmethod + def _indicator_provenance(source, indicator_type): + label = {IND_IP: "ip_address", IND_DOMAIN: "domain", + IND_URL: "url", IND_HASH: "sha256"}.get(indicator_type) + return f"{source}_{label}" if label else source + + # ThreatFox threat_types considered "C2" (command-and-control). Only these + # promote when promote_c2_only is set. + _C2_THREAT_TYPES = {"botnet_cc"} + + def _promote_detections(self, by_indicator, options): + """Append CAPE detections for indicator hits (never sets malfamily). + + With promote_c2_only (default), only C2 (botnet_cc) hits promote; + other threat types still enrich the TI card but are not promoted. The + family is added via add_family_detection, which appends to + results["detections"] without touching the headline malfamily. + """ + try: + from lib.cuckoo.common.utils import add_family_detection + except Exception: + log.warning("ThreatIntelligence: add_family_detection unavailable; cannot promote.") + return [] + + c2_only = _as_bool(options.get("promote_c2_only", True)) + min_conf = int(options.get("promote_minimum_confidence", 100) or 0) + + promoted, seen = [], set() + for indicator, items in by_indicator.items(): + for it in items: + threat_type = (it.get("threat_type") or "").strip().lower() + if c2_only and threat_type not in self._C2_THREAT_TYPES: + continue + conf = it.get("confidence_level") + if min_conf and (conf is None or conf < min_conf): + continue + family = it.get("malware_printable") or it.get("malware") + if not family: + continue + key = (family, indicator) + if key in seen: + continue + seen.add(key) + add_family_detection( + self.results, family, "ThreatIntelligence", + f"ThreatFox {threat_type or 'indicator'}: {indicator}") + promoted.append({"family": family, "indicator": indicator, + "threat_type": threat_type, "confidence_level": conf}) + if promoted: + log.info("ThreatIntelligence: promoted %d indicator hit(s) to detections.", len(promoted)) + return promoted + + def _family_phase(self, providers, hints): + if not hints: + return [] + tasks = [(p, h) for p in providers for h in hints] + log.info("ThreatIntelligence: %d family lookup(s) via %s", + len(tasks), ", ".join(p.name for p in providers)) + results_list = self._run_tasks(tasks, self._lookup_family) + + # Keep one card per (family, source) so multiple intel sources for the + # SAME family (from multiple family providers) coexist and render in one + # family block. Sorted so same family_id is adjacent for the template's + # {% regroup %} (then ordered by source within a family). + cards = {} + for (_p, hint), card in zip(tasks, results_list): + if not card: + continue + fam_id = card.get("family_id") or card.get("common_name") + source = card.get("source") or "?" + key = (fam_id, source) + prov = set(card.get("provenance") or []) | set(hint["provenance"]) + existing = cards.get(key) + if existing is None: + card["provenance"] = sorted(prov) + cards[key] = card + else: + existing["provenance"] = sorted(set(existing["provenance"]) | prov) + return sorted( + cards.values(), + key=lambda c: ((c.get("common_name") or "").lower(), c.get("family_id") or "", c.get("source") or ""), + ) + + def _lookup_family(self, task): + provider, hint = task + key = ("id:" + hint["query"].lower()) if hint["is_id"] else ("name:" + squash_name(hint["query"])) + cached = self.cache.get("family", provider.name, key) + if cached is not None: + return cached or None # {} is a cached miss + try: + card = provider.enrich(hint["query"], is_id=hint["is_id"], provenance=hint["provenance"]) + except Exception: + log.exception("ThreatIntelligence: family provider %s crashed on %s", + provider.name, hint["query"]) + return None + value = card.to_dict() if card else {} + self.cache.set("family", provider.name, key, value) + return value or None + + # ================================================================ + # Actor phase (gated; see run()) + # ================================================================ + def _gather_actor_hints(self, families, options): + """Build de-duplicated actor hints. + + For now, hints are the actors that the matched malware families are + attributed to (Malpedia family ``attribution``). This is community- + sourced, so it is opt-in via ``actor_attribution_from_families`` and + only ever runs inside the master ``threat actors`` gate. When a + high-confidence actor provider is registered, its explicit actor + attributions become additional, preferred hints here. + """ + hints = {} + + def add(name, source, provenance): + if not name: + return + key = squash_name(name) + if not key: + return + entry = hints.get(key) + if entry is None: + hints[key] = {"query": name.strip(), "is_id": False, "provenance": set(provenance or ()) | {source}} + else: + entry["provenance"].update(provenance or ()) + entry["provenance"].add(source) + + if _as_bool(options.get("actor_attribution_from_families", True)): + for fam in families: + for actor in fam.get("attribution") or []: + add(actor, f"family:{fam.get('source')}", fam.get("provenance")) + + return list(hints.values()) + + def _actor_phase(self, providers, hints): + if not hints: + return [] + tasks = [(p, h) for p in providers for h in hints] + log.info("ThreatIntelligence: %d actor lookup(s) via %s", + len(tasks), ", ".join(p.name for p in providers)) + results_list = self._run_tasks(tasks, self._lookup_actor) + + cards = {} + for (_p, hint), card in zip(tasks, results_list): + if not card: + continue + actor_id = card.get("actor_id") or card.get("common_name") + source = card.get("source") or "?" + key = (actor_id, source) + prov = set(card.get("provenance") or []) | set(hint["provenance"]) + existing = cards.get(key) + if existing is None: + card["provenance"] = sorted(prov) + cards[key] = card + else: + existing["provenance"] = sorted(set(existing["provenance"]) | prov) + return sorted( + cards.values(), + key=lambda c: ((c.get("common_name") or "").lower(), c.get("actor_id") or "", c.get("source") or ""), + ) + + def _lookup_actor(self, task): + provider, hint = task + key = ("id:" + hint["query"].lower()) if hint["is_id"] else ("name:" + squash_name(hint["query"])) + cached = self.cache.get("actor", provider.name, key) + if cached is not None: + return cached or None + try: + card = provider.enrich(hint["query"], is_id=hint["is_id"], provenance=hint["provenance"]) + except Exception: + log.exception("ThreatIntelligence: actor provider %s crashed on %s", + provider.name, hint["query"]) + return None + value = card.to_dict() if card else {} + self.cache.set("actor", provider.name, key, value) + return value or None + + # ================================================================ + # Shared task runner + # ================================================================ + def _run_tasks(self, tasks, fn): + if self.concurrent and len(tasks) > 1 and self.max_workers > 1: + workers = min(self.max_workers, len(tasks)) + with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="threatintel") as pool: + return list(pool.map(fn, tasks)) + return [fn(t) for t in tasks] diff --git a/utils/threatintelligence_diag.py b/utils/threatintelligence_diag.py new file mode 100644 index 00000000000..89878bfb562 --- /dev/null +++ b/utils/threatintelligence_diag.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# This file is part of CAPE Sandbox - http://www.capesandbox.com +# See the file 'docs/LICENSE' for copying permission. + +"""Diagnostic for the threat-intelligence integration. + +Run from the CAPE root inside CAPE's venv: + + python3 utils/threatintelligence_diag.py + python3 utils/threatintelligence_diag.py 139.180.203.104 # indicator + python3 utils/threatintelligence_diag.py evil-domain.com # indicator + python3 utils/threatintelligence_diag.py win.castle_stealer # family (Malpedia) + python3 utils/threatintelligence_diag.py "Cobalt Strike" # family (resolve) + +No argument: checks config + engine availability. +""" + +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) +sys.path.insert(0, os.getcwd()) + + +def _looks_like_ip(value): + parts = value.split(".") + return len(parts) == 4 and all(p.isdigit() and 0 <= int(p) <= 255 for p in parts) + + +def main(): + print("=" * 60) + print("CAPE threat-intelligence diagnostic") + print("=" * 60) + + try: + from lib.cuckoo.common.integrations.threatintelligence.config import load_config + from lib.cuckoo.common.integrations.threatintelligence.registry import all_provider_names + config = load_config(all_provider_names()) + options = config.globals + except Exception as err: + print(f"[FAIL] Could not read threat_intel.conf: {err}") + return + if not options: + print("[FAIL] [threatintelligence] section empty or missing in threat_intel.conf.") + return + + def _show(key, value): + if key.endswith(("_api", "_key", "_token", "_secret")) and value: + value = "" + print(f" {key} = {value!r}") + + print("[ OK ] threat_intel.conf read. Global [threatintelligence]:") + for key in sorted(options): + _show(key, options[key]) + for name in all_provider_names(): + section = config.section(name) + if not section: + print(f" [{name}] section absent (engine treated as disabled)") + continue + print(f" [{name}] enabled={config.is_enabled(name)}") + for key in sorted(section): + if key != "enabled": + _show(key, section[key]) + + from lib.cuckoo.common.integrations.threatintelligence.registry import ( + get_enabled_family_providers, get_enabled_indicator_providers, + ) + ind = get_enabled_indicator_providers(config) + fam = get_enabled_family_providers(config) + for p in ind: + print(f"[ OK ] Indicator engine '{p.name}' available. Types: {sorted(p.supported_indicators)}") + for p in fam: + print(f"[ OK ] Family engine '{p.name}' available.") + if not ind and not fam: + print("[FAIL] No engines available (none enabled, or enabled-but-unavailable,") + print(" e.g. threatfox_api / Auth-Key not set).") + return + + if len(sys.argv) <= 1: + print("\nPass an indicator (IP/domain) or a family (name or Malpedia id) to test it.") + return + + from lib.cuckoo.common.integrations.threatintelligence.base import ( + IND_DOMAIN, IND_IP, looks_like_malpedia_id, + ) + arg = sys.argv[1].strip() + + # Family if it looks like a Malpedia id or contains a space / no dotted-host shape. + treat_as_family = looks_like_malpedia_id(arg) or (" " in arg) + + if not treat_as_family and ind: + itype = IND_IP if _looks_like_ip(arg) else IND_DOMAIN + print(f"\n[indicator] {arg!r} as {itype}") + for p in ind: + if not p.accepts_indicator(itype): + continue + res = p.lookup(arg, itype) + print(f" {p.name}: status={res.status}" + (f" error={res.error}" if res.error else "")) + for m in res.matches: + print(f" TAG [{m.tag}] confidence={m.confidence_level} ioc={m.ioc} family={m.malware}") + + if fam: + print(f"\n[family] resolving/fetching {arg!r}") + for p in fam: + is_id = looks_like_malpedia_id(arg) + card = p.enrich(arg, is_id=is_id) + if not card: + print(f" {p.name}: no card (unresolved or unknown family)") + continue + print(f" {p.name}: {card.common_name} ({card.family_id})") + if card.aliases: + print(f" aliases: {', '.join(card.aliases[:8])}") + if card.description: + print(f" description: {card.description[:200]}{'...' if len(card.description) > 200 else ''}") + for ref in card.references: + print(f" ref: {ref['label']} {ref['url']}") + print("\nDone.") + + +if __name__ == "__main__": + main() diff --git a/web/analysis/views.py b/web/analysis/views.py index ae8dbf6ea8e..18ba39da562 100644 --- a/web/analysis/views.py +++ b/web/analysis/views.py @@ -2903,6 +2903,7 @@ def report(request, task_id): "network.hosts": 1, "reversinglabs": 1, "tcr_config_lookup": 1, + "threatintelligence": 1, "_id": 0, } if CUSTOM_SERVICES: diff --git a/web/templates/analysis/network/_dns.html b/web/templates/analysis/network/_dns.html index c959dae3290..0a6b0521b68 100644 --- a/web/templates/analysis/network/_dns.html +++ b/web/templates/analysis/network/_dns.html @@ -21,6 +21,15 @@
DNS Reque {% if config.display_pt_portal %} [PT] {% endif %} +{# threatintel-domain-tags-v1 BEGIN #} + {% if p.threatintel %} + + {% for ti in p.threatintel %} + {{ ti.tag }}{% if ti.indicator_url %}[TF]{% endif %} + {% endfor %} + + {% endif %} +{# threatintel-domain-tags-v1 END #} {% for a in p.answers %} diff --git a/web/templates/analysis/network/_dns_not_ajax.html b/web/templates/analysis/network/_dns_not_ajax.html index 4bc50fb85fd..f50859746c5 100644 --- a/web/templates/analysis/network/_dns_not_ajax.html +++ b/web/templates/analysis/network/_dns_not_ajax.html @@ -16,6 +16,15 @@ {% if config.display_pt_portal %} [PT] {% endif %} +{# threatintel-domain-tags-v1 BEGIN #} + {% if p.threatintel %} + + {% for ti in p.threatintel %} + {{ ti.tag }}{% if ti.indicator_url %}[TF]{% endif %} + {% endfor %} + + {% endif %} +{# threatintel-domain-tags-v1 END #} {% for a in p.answers %} diff --git a/web/templates/analysis/network/_hosts.html b/web/templates/analysis/network/_hosts.html index 5b4f8db6110..4326aca1a53 100644 --- a/web/templates/analysis/network/_hosts.html +++ b/web/templates/analysis/network/_hosts.html @@ -30,6 +30,15 @@
Hosts[PT] {% endif %} +{# threatintel-host-tags-v1 BEGIN #} + {% if host.threatintel %} + + {% for ti in host.threatintel %} + {{ ti.tag }}{% if ti.indicator_url %}[TF]{% endif %} + {% endfor %} + + {% endif %} +{# threatintel-host-tags-v1 END #} {{host.country_name}} diff --git a/web/templates/analysis/network/_hosts_not_ajax.html b/web/templates/analysis/network/_hosts_not_ajax.html index 8296b99931a..aa34730a6d1 100644 --- a/web/templates/analysis/network/_hosts_not_ajax.html +++ b/web/templates/analysis/network/_hosts_not_ajax.html @@ -25,6 +25,15 @@ {% if config.display_pt_portal %} [PT] {% endif %} +{# threatintel-host-tags-v1 BEGIN #} + {% if host.threatintel %} + + {% for ti in host.threatintel %} + {{ ti.tag }}{% if ti.indicator_url %}[TF]{% endif %} + {% endfor %} + + {% endif %} +{# threatintel-host-tags-v1 END #} {{host.country_name}} diff --git a/web/templates/analysis/overview/index.html b/web/templates/analysis/overview/index.html index 691bffc146c..0a5c86f0dc2 100644 --- a/web/templates/analysis/overview/index.html +++ b/web/templates/analysis/overview/index.html @@ -154,6 +154,178 @@
Parent File Info
{% include "analysis/overview/_statistics.html" %} {% endif %} +{# threatintel-overview-section-v1 BEGIN #} + +{% if analysis.threatintelligence.families or analysis.threatintelligence.actors %} +
+
+
+
Threat Intelligence
+
+
+ {% if analysis.threatintelligence.families %} +
+ {% regroup analysis.threatintelligence.families by family_id as ti_groups %} + {% for grp in ti_groups %} + {% with first=grp.list.0 %} +
+
+

+ +

+
+
+
+ {% if first.provenance %} +

Detection sources: + {% for p in first.provenance %}{{ p }}{% endfor %} +

+ {% endif %} + {% for card in grp.list %} + {% if not forloop.first %}
{% endif %} +
+ {{ card.source }} + {% for alias in card.aliases %}{{ alias }}{% endfor %} +
+ {% if card.description %} +

{{ card.description }}

+ {% else %} +

No description available from {{ card.source }}.

+ {% endif %} + {% comment %} + Malpedia's actor attribution is community/MISP-sourced and frequently + mislabels commodity malware as APT activity, so it is intentionally NOT + shown on the malware card. Actor attribution is surfaced only via the + dedicated, high-confidence threat-actor cards. + {% endcomment %} + {% if card.references %} +
+ Recent references: +
    + {% for ref in card.references %} +
  • {{ ref.url }}{% if ref.date %} ({{ ref.date }}){% endif %}
    • + {% endfor %} +
+
+ {% endif %} + {% if card.url %} +
+ View on {{ card.source|capfirst }} +
+ {% endif %} + {% endfor %} +
+
+
+ {% endwith %} + {% endfor %} +
+ {% endif %}{# /families #} + + {# Threat actors — scaffolding; populated only when a (high-confidence) actor engine is enabled #} + {% if analysis.threatintelligence.actors %} + +
+ + Threat actors +
+
+ {% regroup analysis.threatintelligence.actors by actor_id as ta_groups %} + {% for grp in ta_groups %} + {% with first=grp.list.0 %} +
+
+

+ +

+
+
+
+ {% if first.provenance %} +

Attribution sources: + {% for p in first.provenance %}{{ p }}{% endfor %} +

+ {% endif %} + {% for card in grp.list %} + {% if not forloop.first %}
{% endif %} +
+ {{ card.source }} + {% if card.attribution_confidence is not None %}confidence {{ card.attribution_confidence }}%{% endif %} + {% for alias in card.aliases %}{{ alias }}{% endfor %} +
+ {% if card.description %} +

{{ card.description }}

+ {% else %} +

No description available from {{ card.source }}.

+ {% endif %} + {% if card.families %} +

Associated malware: + {% for fam in card.families %}{{ fam }}{% endfor %} +

+ {% endif %} + {% if card.references %} +
+ References: +
    + {% for ref in card.references %} +
  • {{ ref.url }}
    • + {% endfor %} +
+
+ {% endif %} + {% if card.url %} +
+ View on {{ card.source|capfirst }} +
+ {% endif %} + {% endfor %} +
+
+
+ {% endwith %} + {% endfor %} +
+ {% endif %}{# /actors #} +
+
+
+{% endif %} +{# threatintel-overview-section-v1 END #} + {% if analysis.signatures %} {% include "analysis/overview/_signatures.html" %} {% endif %} From 39e8825d3a811e8785f91edf629951ecbae5c3a1 Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:50:48 +0100 Subject: [PATCH 2/9] Update copyright year in cache.py --- lib/cuckoo/common/integrations/threatintelligence/cache.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/cuckoo/common/integrations/threatintelligence/cache.py b/lib/cuckoo/common/integrations/threatintelligence/cache.py index 33157b1b36f..fd139c87099 100644 --- a/lib/cuckoo/common/integrations/threatintelligence/cache.py +++ b/lib/cuckoo/common/integrations/threatintelligence/cache.py @@ -1,4 +1,4 @@ -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission. From f2bf59b64a76040498f76f496a5abc9a1c7f59d8 Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:51:16 +0100 Subject: [PATCH 3/9] Update copyright year to 2026 --- lib/cuckoo/common/integrations/threatintelligence/base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/cuckoo/common/integrations/threatintelligence/base.py b/lib/cuckoo/common/integrations/threatintelligence/base.py index 9351b9d50f9..029674b520c 100644 --- a/lib/cuckoo/common/integrations/threatintelligence/base.py +++ b/lib/cuckoo/common/integrations/threatintelligence/base.py @@ -1,4 +1,4 @@ -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission. From c20fcd5f840cbb999a2685aeabf386a7396998de Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:51:42 +0100 Subject: [PATCH 4/9] Update copyright year to 2026 in config.py --- lib/cuckoo/common/integrations/threatintelligence/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/cuckoo/common/integrations/threatintelligence/config.py b/lib/cuckoo/common/integrations/threatintelligence/config.py index 4ce001678f0..e7a3e32aa0c 100644 --- a/lib/cuckoo/common/integrations/threatintelligence/config.py +++ b/lib/cuckoo/common/integrations/threatintelligence/config.py @@ -1,4 +1,4 @@ -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission. From 6812f71a4789a7209bdc9a649d2fe78530ded531 Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:52:01 +0100 Subject: [PATCH 5/9] Update malpedia_actor_provider.py --- .../integrations/threatintelligence/malpedia_actor_provider.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py b/lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py index 8b2b5f5f13e..792be7052f0 100644 --- a/lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py +++ b/lib/cuckoo/common/integrations/threatintelligence/malpedia_actor_provider.py @@ -1,4 +1,4 @@ -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission. From 2233d9fea8e0c618dbbdf2f8a7b32845c250566c Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:52:19 +0100 Subject: [PATCH 6/9] Update malpedia_provider.py --- .../common/integrations/threatintelligence/malpedia_provider.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py b/lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py index a93e08c5229..bf3b0bdd2e1 100644 --- a/lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py +++ b/lib/cuckoo/common/integrations/threatintelligence/malpedia_provider.py @@ -1,4 +1,4 @@ -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission. From 9108711ab9546835580a17a09f7e0b44de2067f1 Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:52:38 +0100 Subject: [PATCH 7/9] Update registry.py --- lib/cuckoo/common/integrations/threatintelligence/registry.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/cuckoo/common/integrations/threatintelligence/registry.py b/lib/cuckoo/common/integrations/threatintelligence/registry.py index 28647720df3..be807a8be1e 100644 --- a/lib/cuckoo/common/integrations/threatintelligence/registry.py +++ b/lib/cuckoo/common/integrations/threatintelligence/registry.py @@ -1,4 +1,4 @@ -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission. From 2d697e097c433396b9d021634b346dc7eb9da92f Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:53:04 +0100 Subject: [PATCH 8/9] Update copyright year to 2026 in threatfox_provider.py --- .../integrations/threatintelligence/threatfox_provider.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py b/lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py index 47095cdc4c9..1cdb405c7c6 100644 --- a/lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py +++ b/lib/cuckoo/common/integrations/threatintelligence/threatfox_provider.py @@ -1,4 +1,4 @@ -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission. From cb929a20f0329574e5f5c2d1be0f554a0e2a8047 Mon Sep 17 00:00:00 2001 From: kevross33 Date: Thu, 30 Jul 2026 13:53:24 +0100 Subject: [PATCH 9/9] Update copyright year in threatintelligence_diag.py --- utils/threatintelligence_diag.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/utils/threatintelligence_diag.py b/utils/threatintelligence_diag.py index 89878bfb562..e6d0a850c8f 100644 --- a/utils/threatintelligence_diag.py +++ b/utils/threatintelligence_diag.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2024 CAPE developers. +# Copyright (C) 2010-2015 Cuckoo Foundation, 2016-2026 CAPE developers. # This file is part of CAPE Sandbox - http://www.capesandbox.com # See the file 'docs/LICENSE' for copying permission.