IOC Feed API

Python: poll & diff

import os, time, requests

API = "https://api.signalis.watch/api/feed/v1/iocs"
HEADERS = {"Authorization": f"Bearer {os.environ['SIGNALIS_API_KEY']}"}

def pull(since=None):
    """Yield every attacker IOC since a timestamp, walking the cursor. Honors backoff."""
    cursor = None
    while True:
        params = {"limit": 500}
        if since: params["since"] = since
        if cursor: params["cursor"] = cursor
        r = requests.get(API, headers=HEADERS, params=params, timeout=30)
        if r.status_code in (429, 503):
            time.sleep(int(r.headers.get("Retry-After", 5)))
            continue
        r.raise_for_status()
        body = r.json()
        yield from body["data"]
        if not body["has_more"]:
            break
        cursor = body["next_cursor"]

if __name__ == "__main__":
    # poll-and-diff: persist the max created_at and pass it as ?since next run
    for ioc in pull(since="2026-06-01T00:00:00Z"):
        print(ioc["type"], ioc["value"], "->", ioc["source_name"])

MISP: PyMISP pull

Maps Signalis IOC types to MISP attribute types; skips types with no clean mapping (cve, yara_rule, crypto addresses beyond BTC).

import os, requests
from pymisp import PyMISP, MISPEvent, MISPAttribute

SIG = "https://api.signalis.watch/api/feed/v1/iocs"
misp = PyMISP(os.environ["MISP_URL"], os.environ["MISP_KEY"], ssl=True)

# Signalis IOC type -> MISP attribute type. Types with no clean MISP mapping are skipped.
MAP = {
    "domain": "domain", "url": "url", "ip_v4": "ip-dst", "ip_v6": "ip-dst",
    "email": "email-src", "hash_md5": "md5", "hash_sha1": "sha1",
    "hash_sha256": "sha256", "hash_sha512": "sha512", "filename": "filename",
    "file_path": "filename", "asn": "AS", "user_agent": "user-agent",
    "bitcoin_address": "btc", "registry_key": "regkey", "mutex": "mutex",
}

def signalis(since):
    cursor = None
    while True:
        p = {"limit": 500, "since": since}
        if cursor: p["cursor"] = cursor
        b = requests.get(SIG, headers={"Authorization": f"Bearer {os.environ['SIGNALIS_API_KEY']}"},
                         params=p, timeout=30).json()
        yield from b["data"]
        if not b["has_more"]:
            break
        cursor = b["next_cursor"]

event = MISPEvent(); event.info = "Signalis attacker IOCs"
for ioc in signalis("2026-06-01T00:00:00Z"):
    t = MAP.get(ioc["type"])
    if not t:
        continue
    a = MISPAttribute(); a.type = t; a.value = ioc["value"]; a.to_ids = True
    a.comment = f"Signalis · {ioc['source_name']}"
    event.add_attribute(**a)
misp.add_event(event)

OpenCTI: generic import

A generic pull-and-create sketch via pycti. This is not a packaged connector (no dedup, bundling, or STIX relationships). A dedicated external-import connector is a separate effort. Adjust the observable-create call to your pycti version.

import os, requests
from pycti import OpenCTIApiClient

# GENERIC IMPORT (not a packaged connector; see the note below).
SIG = "https://api.signalis.watch/api/feed/v1/iocs"
octi = OpenCTIApiClient(os.environ["OPENCTI_URL"], os.environ["OPENCTI_TOKEN"])

OBS = {  # Signalis type -> OpenCTI observable type
    "domain": "Domain-Name", "url": "Url", "ip_v4": "IPv4-Addr", "ip_v6": "IPv6-Addr",
    "email": "Email-Addr", "hash_md5": "StixFile", "hash_sha1": "StixFile", "hash_sha256": "StixFile",
}

def signalis(since):
    cursor = None
    while True:
        p = {"limit": 500, "since": since}
        if cursor: p["cursor"] = cursor
        b = requests.get(SIG, headers={"Authorization": f"Bearer {os.environ['SIGNALIS_API_KEY']}"},
                         params=p, timeout=30).json()
        yield from b["data"]
        if not b["has_more"]:
            break
        cursor = b["next_cursor"]

for ioc in signalis("2026-06-01T00:00:00Z"):
    t = OBS.get(ioc["type"])
    if not t:
        continue
    octi.stix_cyber_observable.create(
        simple_observable_key=f"{t}.value", simple_observable_value=ioc["value"],
        x_opencti_description=f"Signalis · {ioc['source_name']}",
    )