Metadata-Version: 2.4
Name: ciaops
Version: 3.2.0
Summary: Python library - modules for processing data from the TI, ASM, DRP and IMC systems collected in one library. This library simplifies work with the products API and gives you the flexibility to customize the search and retrieval of data from the system.
Author-email: Group-IB <integration@group-ib.com>
License-Expression: MIT
Keywords: group-ib,threat intelligence,digital risk protection,attack surface management,incident management,cybersecurity,ti,drp,asm,imc,api client
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Topic :: Security
Classifier: Typing :: Typed
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Natural Language :: English
Classifier: Operating System :: OS Independent
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.31.0
Requires-Dist: urllib3>=2.0.2
Provides-Extra: misp
Requires-Dist: pyaml>=25.7.0; extra == "misp"
Dynamic: license-file

# ciaops

[![Python](https://img.shields.io/badge/python-v3.9+-blue?logo=python)](https://python.org/downloads/release/python-390/)

**ciaops** - Python library to communicate with **Group-IB Products** (TI, DRP, ASM, IMC) via **API**.

## **License**

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

## **Content**

- [ciaops](#ciaops)
  - [**License**](#license)
  - [**Content**](#content)
  - [**Installation**](#installation)
  - [**Usage**](#usage)
    - [Initialization](#initialization)
    - [Threads and rate limits](#threads-and-rate-limits)
    - [Collection constants](#collection-constants)
    - [Collections mapping](#collections-mapping)
    - [Portions generator](#portions-generator)
    - [TIPoller extra methods](#tipoller-extra-methods)
      - [Available collections](#available-collections)
      - [Find feed by ID](#find-feed-by-id)
      - [Download file](#download-file)
      - [Download PDF reports](#download-pdf-reports)
      - [IP scoring](#ip-scoring)
      - [Domain scoring](#domain-scoring)
      - [MITRE ATT\&CK](#mitre-attck)
      - [Global search](#global-search)
    - [DRPPoller methods](#drppoller-methods)
    - [ASMPoller methods](#asmpoller-methods)
    - [IMCPoller methods](#imcpoller-methods)
      - [Consuming incidents](#consuming-incidents)
      - [Filtering incidents](#filtering-incidents)
      - [Reading an incident](#reading-an-incident)
      - [Drilling into the evidence](#drilling-into-the-evidence)
      - [Resolving matches to TI records](#resolving-matches-to-ti-records)
    - [Close session](#close-session)
  - [Parsing](#parsing)
    - [Parse portion method](#parse-portion-method)
    - [Get IoCs method](#get-iocs-method)
  - [Utilities](#utilities)
    - [ParserHelper](#parserhelper)
      - [find\_by\_template](#find_by_template)
      - [find\_element\_by\_key](#find_element_by_key)
      - [unpack\_iocs](#unpack_iocs)
    - [Validator](#validator)
  - [Portal Links](#portal-links)
    - [Simple collections (single ID in URL)](#simple-collections-single-id-in-url)
    - [Multi-part URL templates](#multi-part-url-templates)
    - [Custom prefix override](#custom-prefix-override)
    - [Embedding portal links in parsed output](#embedding-portal-links-in-parsed-output)
    - [Inspecting the full URL map](#inspecting-the-full-url-map)
  - [Adapter utilities](#adapter-utilities)
    - [ConfigParser](#configparser)
    - [FileHandler](#filehandler)
  - [Examples](#examples)
    - [Full version of program](#full-version-of-program)
  - [API logic](#api-logic)
    - [Sequence update logic](#sequence-update-logic)
      - [API response](#api-response)
      - [Iteration steps](#iteration-steps)
      - [Stop the iteration](#stop-the-iteration)
    - [Search logic](#search-logic)
      - [Global search](#global-search-1)
      - [Iteration steps](#iteration-steps-1)
      - [Stop the iteration](#stop-the-iteration-1)
  - [Records limits](#records-limits)
  - [Recommended TTL](#recommended-ttl)
  - [Troubleshooting](#troubleshooting)
    - [401 response code](#401-response-code)
    - [403 response code](#403-response-code)
    - [504 response code or timeout](#504-response-code-or-timeout)
  - [FAQ](#faq)

<br>

## **Installation**

Lib deps: **requests**, **urllib3**.

ciaops lib is available on PyPI:

```
pip install ciaops
```

Or use a Portal WHL archive. Replace `X.X.X` with current lib version:

```
pip install ./ciaops-X.X.X-py3-none-any.whl
```

<br>

## **Usage**

### Initialization

Initialize a **poller** with your credentials. TLS certificate verification is enabled by default.

Call `set_verify()` only when you need to override the default: pass `False` to disable verification (e.g. local testing behind a corporate proxy), or pass a path to a custom CA bundle.

```python
from ciaops import TIPoller, DRPPoller, ASMPoller, IMCPoller

# Threat Intelligence
ti = TIPoller(username='example@example.corp', api_key='API_KEY', api_url='TI_API_URL')

# Digital Risk Protection
drp = DRPPoller(username='example@example.corp', api_key='API_KEY', api_url='DRP_API_URL')

# Attack Surface Management
asm = ASMPoller(username='example@example.corp', api_key='API_KEY', api_url='ASM_API_URL')

# Incident Management Center
imc = IMCPoller(username='example@example.corp', api_key='API_KEY', api_url='TI_API_URL')

# Override only when needed:
ti.set_verify(False)                  # disable TLS check (local testing only)
ti.set_verify('/path/to/ca-bundle')   # custom CA bundle
```

Proxy setup (all pollers share the same interface):

```python
ti.set_proxies(
    proxy_protocol='https',
    proxy_ip='10.0.0.1',
    proxy_port='3128',
    proxy_username='user',       # optional
    proxy_password='secret',     # optional
)
```

Use pollers as context managers to ensure the session is always closed:

```python
with TIPoller(username='...', api_key='...', api_url='...') as ti:
    generator = ti.create_update_generator('apt/threat', sequpdate=16172928022293)
    for portion in generator:
        print(portion.parse_portion())
```

Tag your integration in the User-Agent header with `set_product()` — required by some on-premise deployments:

```python
ti.set_product(
    product_type='SIEM',
    product_name='MySIEM',
    product_version='2.1',
    integration_name='ciaops-siem',
    integration_version='1.0',
)
```

### Threads and rate limits

**Use one poller per thread.** A poller owns an HTTP session, the
`set_keys` mappings and (for IMC) the rate limiter it hands to every walk it
drives; a feed generator owns cursor state. None of it is synchronized.

More importantly, the rate limits are **per account, not per connection**
(IMC: 1 request/second; TI collections vary). Running the same account from
several threads therefore buys no throughput — each thread paces itself
independently, so together they exceed the limit and collect `429`s, which the
retry adapter then serializes anyway. Where you need more data per second,
widen the page size or the time window rather than the thread count.

`RateLimiter` is the library's rate limiter and is likewise per-worker; give
each thread its own.

### Collection constants

Use typed constants instead of bare strings to avoid typos and get IDE autocomplete:

```python
from ciaops.collections_meta import (
    TICollections, DRPCollections, ASMCollections, IMCCollections
)

# TI examples
TICollections.APT_THREAT          # "apt/threat"
TICollections.COMPROMISED_ACCOUNT_GROUP  # "compromised/account_group"
TICollections.MALWARE_CNC         # "malware/cnc"

# DRP
DRPCollections.VIOLATION          # "violation"
DRPCollections.COMPROMISED_DARKWEB  # "compromised/darkweb"

# ASM
ASMCollections.ASSETS_UPDATED     # "assets/updated"
ASMCollections.ISSUES_UPDATED     # "issues/updated"

# IMC — the two logical collections you set keys for
IMCCollections.INCIDENTS          # "imc/incidents"
IMCCollections.MATCHES            # "imc/matches"

# IMC — the ten match collections, in every spelling the API uses
IMCCollections.DARK_WEB                          # "darkWeb"      (byCollection key)
IMCCollections.get_match_segment('darkWeb')      # "dark-web"     (endpoint segment)
IMCCollections.get_collection_enum('darkWeb')    # "COLLECTION_DARK_WEB"  (filter enum)
IMCCollections.get_resolution_route('darkWeb')   # "darkweb/forums"       (TI by-id route)
IMCCollections.normalize_collection('dark-web')  # "darkWeb"      (accepts those three)
```

Look up the recommended TTL for any TI collection:

```python
TICollections.get_ttl(TICollections.MALWARE_CNC)          # 90  (days)
TICollections.get_ttl(TICollections.COMPROMISED_MESSENGER) # None  (no expiry)
```

Use constants anywhere a collection name string is expected:

```python
poller.set_keys(TICollections.APT_THREAT, keys)
generator = poller.create_update_generator(TICollections.MALWARE_CNC, sequpdate=...)
```

### Collections mapping

Method `set_keys()` sets **keys** to search in the selected **collection**. It should be python dict `mapping_keys = {key: value}` where \
**key** - result name \
**value** - dot-notation string with searchable keys

```python
mapping_keys = {"result_name": "searchable_key_1.searchable_key_2"}
```

Parser finds keys recursively in the API response, using dot-notation in **value**.
If you want to add your own data to the results start the **value** with star `*`.

```python
mapping_keys = {
	"network": "indicators.params.ip",
	"result_name": "*My_Value"
}
```

For `set_keys()` or `set_iocs_keys()` methods you can make a full template to get nested data in the way you want.

```python
mapping_keys = {
	'network': {
		'ips': 'indicators.params.ip'
	},
	'url': 'indicators.params.url',
	'type': '*network'
}
poller.set_keys(collection_name="apt/threat", keys=mapping_keys)
poller.set_iocs_keys(collection_name="apt/threat", keys={"ips": "indicators.params.ip"})
```

### Portions generator

Use `create_update_generator()` to create a generator that returns portions of feeds. \
**Update generator** - goes through the feeds in ascending order. Feed iteration is based on the `seqUpdate` field.

**Note:** `compromised/breached` is a by-id route without an `/updated` feed, so
[sequence update logic](#sequence-update-logic) does not apply to it — its feed
counterpart is `compromised/breacheddb`.

```python
from ciaops import TIRequestParams

generator = poller.create_update_generator(
    collection_name='compromised/account_group',
    request_params=TIRequestParams(
        date_from='2021-01-30',
        date_to='2021-02-03',
        query='8.8.8.8',
        limit=200,
    ),
    sequpdate=20000000,
)
```

The `compromised/spd` (Suspicious Payment Details) collection additionally
accepts a `type` filter that narrows results to a single payment-detail type.
Allowed values: `Cryptocurrency Wallet`, `Banking Account Number`,
`Mobile number`, `IBAN`, `Bank card`, `Unknown`, `QR payment string`.

```python
generator = poller.create_update_generator(
    collection_name='compromised/spd',
    request_params=TIRequestParams(type='IBAN'),
    sequpdate=20000000,
)
```

**Rate limiting:** the TI update generator throttles itself to **at most one
request every 10 seconds** by default (elapsed-aware — it sleeps only the time
remaining since the previous request), which avoids `429 Too Many Requests`.
Override it with `rate_limit_delay` (seconds), or pass `0` to disable:

```python
generator = poller.create_update_generator(
    collection_name='compromised/account_group',
    request_params=TIRequestParams(date_from='2021-01-30'),
    rate_limit_delay=0,  # disable throttling
)
```

Each portion (iterable object) presented as `Parser` class object.
You can get **raw data** (in json format) or **parsed portion** (python dictionary format),
using its methods and attributes.

```python
for portion in generator:
    parsed_json = portion.parse_portion(as_json=False)
    iocs = portion.get_iocs(as_json=False)
    sequpdate = portion.sequpdate
    count = portion.count
    raw_json = portion.raw_json
    raw_dict = portion.raw_dict
    new_parsed_json = portion.bulk_parse_portion(keys_list=[{"ips": "indicators.params.ip"}, {"url": 'indicators.params.url'}], as_json=False)
```

Attribute `sequpdate` of the generator iterable object, gives you the last **sequence update number** (`seqUpdate`)
of the feed, which you can save locally.

```python
sequpdate = portion.sequpdate
```

Attribute `count` of the generator iterable object shows you the number of feeds left in the queue.

```python
count = portion.count
```

Methods `parse_portion()` and `get_iocs()` of generator iterable objects, use your
mapping keys (IoCs keys) to return parsed data.
You can override mapping keys using `keys` parameter in these functions.

```python
parsed_json = portion.parse_portion(as_json=False)
iocs = portion.get_iocs(as_json=False, keys=mapping_override_keys)
```

Also, you can use `bulk_parse_portion()` method to get multiple parsed dicts from every feed.

```python
new_parsed_json = portion.bulk_parse_portion(keys_list=[{"ips": "indicators.params.ip"}, {"url": 'indicators.params.url'}], as_json=False)
```

### TIPoller extra methods

#### Available collections

Call `get_available_collections()` to discover which collections your API key can access before iterating.

```python
collection_list = ti.get_available_collections()
seq_update_dict = ti.get_seq_update_dict(date='2020-12-12')
compromised_account_sequpdate = seq_update_dict.get('compromised/account_group')

# Check which collections have active hunting rules applied
hunting_collections = ti.get_hunting_rules_collections()
```

#### Find feed by ID

Returns a `Parser` object for a single feed by its ID.

```python
feed = ti.search_feed_by_id(collection_name='apt/threat', feed_id='abc123')
parsed = feed.parse_portion()
```

#### Download file

Download a binary file embedded inside a threat report.

```python
binary = ti.search_file_in_threats(
    collection_name='hi/threat',
    feed_id='feed_id',
    file_id='file_id_inside_feed',
)
```

#### Download PDF reports

```python
# Download a PDF for an HI or APT threat
pdf_bytes = ti.download_threat_pdf(threat_id='abc123')

# Download an HI analytic report PDF (use file.name field from hi/analytic record)
pdf_bytes = ti.download_analytic_report_pdf(file_name='/23ae4ab7.../file/450ffbd4...')

with open('report.pdf', 'wb') as f:
    f.write(pdf_bytes)
```

#### IP scoring

Score one or more IPs against the TI database via the `scoring/ip` endpoint.

```python
# Single IP
result = ti.get_ip_scoring('8.8.8.8')
# → {"items": {"8.8.8.8": {"score": 7.5, ...}}}

# Multiple IPs
result = ti.get_ip_scoring(['8.8.8.8', '1.1.1.1'])
```

#### Domain scoring

Score one or more domains against the TI database via the `scoring/domain`
endpoint.

```python
# Single domain
result = ti.get_domain_scoring('example.com')
# → {"items": {"example.com": {"score": 7.5, ...}}}

# Multiple domains
result = ti.get_domain_scoring(['example.com', 'test.com'])
```

#### MITRE ATT&CK

Fetch the full MITRE ATT&CK technique vocabulary or a ready-to-use ID→name map.

```python
# Raw vocabulary (includes all AttackPattern details)
vocab = ti.get_mitre_techniques()

# Convenient ID → name dict
mitre_map = ti.get_mitre_attack_pattern_map()
# → {"T1059": "Command and Scripting Interpreter", "T1078": "Valid Accounts", ...}

technique_name = mitre_map.get("T1059")
```

#### Global search

Search across all TI collections by query string.

```python
results = ti.global_search('8.8.8.8')
# → [{"apiPath": "suspicious_ip/scanner", "count": 14, ...}, ...]
```

### DRPPoller methods

```python
from ciaops import DRPPoller

drp = DRPPoller(username='...', api_key='...', api_url='DRP_URL')
```

**Update generator** — iterate violation feeds:

```python
generator = drp.create_update_generator(
    collection_name='violation',
    sequpdate=1700000000000000,
    subtypes=[6],           # 1=counterfeit 2=piracy 3=partner_policy 4=trademark 5=malware 6=phishing 7=fraud 8=no_violation
    section=[1, 2],         # 1=Web 2=Mobile 3=Marketplace 4=Social 5=Advertising 6=Messengers
    brands=['brand_id'],
    approve_states=['under_review'],
)
for portion in generator:
    data = portion.parse_portion()
```

**Find feed by ID:**

```python
feed = drp.search_feed_by_id(feed_id='violation_id')
raw = feed.raw_dict
```

**Change violation status** (only when status=`detected` and approveState=`under_review`):

```python
drp.change_status(feed_id='violation_id', status='approve')  # or 'reject'
```

**Brands and subscriptions:**

```python
brands = drp.get_brands()
# → [{"name": "Brand A", "id": "id1"}, ...]

subscriptions = drp.get_subscriptions()
# → ["scam", "phishing", ...]
```

**Typo-squatting scan** (iterates from the very beginning):

```python
generator = drp.create_update_generator(
    collection_name='violation',
    use_typo_squatting=True,
)
```

**seqUpdate by date:**

```python
seq_dict = drp.get_seq_update_dict(date='2024-01-15')
# → {"violation": 1705276800000000, ...}
```

### ASMPoller methods

```python
from ciaops import ASMPoller

asm = ASMPoller(username='...', api_key='...', api_url='ASM_URL')
```

**List companies:**

```python
companies = asm.get_companies()          # all companies
active = asm.get_companies(status='active')
# → [{"id": "uuid", "name": "Acme Corp"}, ...]
```

**Update generator** — uses POST requests with automatic rate limiting.
Build the request body with `ASMRequestParams` and pass it as
`request_params`:

```python
from ciaops import ASMRequestParams

generator = asm.create_update_generator(
    collection_name='assets/updated',   # or 'leaks/updated', 'issues/updated'
    request_params=ASMRequestParams(
        company_id='company-uuid',      # or list of UUIDs
        date_from='2024-01-01',
        date_to='2024-06-01',           # optional
        count=500,                      # max 5000
        status=['new', 'confirmed'],    # optional filter
        type=['domain', 'ip'],          # optional filter (assets only)
    ),
)
for portion in generator:
    data = portion.parse_portion()
```

**How ASM pages.** Iteration is driven by `seqUpdate`, an integer the API
returns on every response page; each page's value is sent on the next request
and the walk ends when a page comes back empty. `dateFrom`/`dateTo` are
**filters, not a starting offset** — they scope the result set, so they are
sent on *every* request alongside the cursor, not only on the first. Pass
`sequpdate=` to resume from a checkpoint you stored earlier; it is sent on the
first request together with the date window.

Read the checkpoint for the next run off the last portion:

```python
sequpdate = None
for portion in generator:
    data = portion.parse_portion()
    sequpdate = portion.sequpdate   # store after the loop
```

If the server returns records but no new cursor — it should not, since
`seqUpdate` is always present — the walk yields that page, logs a warning
naming the stalled cursor, and stops rather than re-requesting the same page
forever.

**Dashboard scores:**

```python
scores = asm.get_dashboard_scores(company_id='company-uuid')

print(scores.current_score)          # 7.4
print(scores.score_trend)            # "improving" | "declining" | "stable"
print(scores.total_critical)         # 3
print(scores.severity_summary)       # {"critical": 3, "high": 12, ...}
print(scores.counters_summary)       # {"new_assets": 5, "new_issues": 2, ...}
print(scores.lowest_scoring_category)  # {"name": "Network Security", ...}

summary = scores.get_dashboard_summary()  # full dict for periodic updates
raw = asm.get_dashboard_scores(company_id='uuid', as_raw=True)  # plain dict
```

**Issue management:**

```python
evidence = asm.get_issue_evidence(issue_id='issue-uuid')

asm.add_issue_comment(
    company_id='company-uuid',
    issue_id='issue-uuid',
    body='Investigating...',
)

asm.change_issue_status(
    issues_id=['issue-uuid-1', 'issue-uuid-2'],
    status='Under review',   # Detected | Under review | Solved | Ignored | False positive
)
```

**Asset management:**

```python
asm.add_assets(
    company_id='company-uuid',
    confirmed_domain=['example.corp'],
    confirmed_ip=['1.2.3.4'],
)

asm.remove_assets(
    company_id='company-uuid',
    excluded_domain=['old.group-ib.com'],
)

asm.change_asset_status(
    assets_ids=['asset-uuid'],
    status='confirmed',   # new | false | confirmed
)
```

### IMCPoller methods

**IMC (Incident Management Center)** is the incident layer on top of the TI
collections. Server-side **incident rules** match records across TI
collections; matches inside a time window are rolled up into an **incident**
with a status, priority, scoring, per-collection match counts and an optional
AI summary. Each match carries only a **pointer** into the source collection,
which you resolve to get the record itself.

IMC lives on the **TI gateway and uses the TI credentials** — one API key
covers both the incident feed and the by-id lookups that resolve its matches.

`IMCPoller` covers the whole public v2 surface: both `IncidentService`
operations and every `RuleMatchService` endpoint, and the implementation is
checked against the generated OpenAPI contract, so a change in the API
surfaces as a failing check rather than a silent gap.

The API is **read-only**: no method on `IMCPoller` changes incident state.

```python
from ciaops import IMCPoller

imc = IMCPoller(username='example@example.corp', api_key='API_KEY', api_url='TI_API_URL')
```

Requests are held to **one per second**. That rate is the TI gateway's,
inherited because IMC shares it; IMC's own limit is not published, and the live
runs at this rate never drew a `429`. Anything in this library that
makes more than one request paces itself to it — both walks and
`iter_resolved_matches` — and `rate_limit_delay=0` turns that off when an outer
scheduler already paces the calls.

The single-request methods (`search_updated_incidents`, `fetch_incident`,
`resolve_match`) make exactly one call and do **not** sleep: pacing one request
would only slow a connector down. Calling them in a loop is where the limit
becomes yours to respect — `RateLimiter` is exported for that:

```python
from ciaops import RateLimiter

rate_limiter = RateLimiter(1.0)
for incident_id in ids:
    rate_limiter.wait()
    portion = imc.fetch_incident(incident_id)
```

#### Consuming incidents

The incident feed is paged and delta in one mechanism, on an opaque `syncToken`
cursor: keep calling with the returned token until a page comes back with no
incidents — that empty page means you are caught up. Store the last token; pass
it back later and you get only what changed. It is the IMC equivalent of
`seqUpdate`.

The sync axis is the incident's **`updatedAt`**, so a status or assignee change
brings an old incident back into the feed. Ordering and page size are fixed by
the server (there is no `limit`), and **deletions are not tracked** — the feed
carries no tombstones.

Two consequences of syncing on `updatedAt` that a connector has to plan for,
both stated by the API's own documentation:

- **An incident shared with you after you synced past it never arrives.**
  Visibility is company-scoped; if an incident becomes visible later and its
  `updatedAt` did not move, no delta poll will return it. A periodic full
  re-sync (no `sync_token`, bounded by `updated_at_from`) is the only way to
  pick those up.
- **Editing a rule changes its incidents without moving their `updatedAt`.**
  See the rule-drift note below: name, priority, scoring and tags are served
  from the rule's current state, so a delta sync will not see the change.

> **A first poll returns the account's *oldest* incidents first.** The ordering
> is ascending by `updatedAt`, so the head of the first page is the oldest
> slice, not the newest. Which means:
>
> - To look at *recent* activity, filter by time rather than reading page one:
>   `IMCIncidentFilter(updated_at_from=...)`. Draining the whole feed also gets
>   you there, just later.
> - A **filtered** query is ascending as well, so a wide window still returns
>   the oldest incidents *of that window*. Reaching "now" means a **narrow**
>   window — measured on a live account, `updated_at_from = now - 1 day` put
>   today's incidents on the first page, while `now - 365 days` put
>   84-day-old ones there. Widen only until you get results, never past that.
> - `number` is a per-rule sequence, so it says nothing about age. Compare
>   `updatedAt`, not numbers.
>
> This matters in practice: **an old incident can report a large `matchCounts`
> whose match endpoint answers `200` with an empty page.** Measured on a live
> account: incidents a few **hours** old returned their matches in full (and
> resolved to their TI records), while incidents 84+ **days** old reported
> thousands of matches and returned none. Nothing in the contract explains it
> and no threshold is published, so treat evidence on an old incident as
> something that may be gone, and drill into recent ones. The library warns
> whenever a drained walk disagrees with the reported count, so this never
> passes as "no evidence".

One page per poll (what a scheduled connector usually wants):

```python
portion = imc.search_updated_incidents(sync_token=load_checkpoint())

for incident in portion.as_incidents():
    print(incident.number, incident.status, incident.rule_name)

save_checkpoint(portion.next_sync_token)
```

Or drain the whole backlog in one run. `create_incidents_walk` returns the
**walk object**, so the checkpoint is still reachable when the loop ends: the
last `nextSyncToken` arrives on the terminal empty page, which is never
yielded.

```python
walk = imc.create_incidents_walk(sync_token=load_checkpoint())
for portion in walk:
    for incident in portion.as_incidents():
        ...
save_checkpoint(walk.sync_token)     # the caught-up checkpoint
```

Checkpointing `portion.next_sync_token` as you go is equally correct — it just
costs one extra page on the next poll. A walk is single-use: iterating one
twice raises `ConsumedWalkError`, rather than silently re-requesting and
double-counting.

**When a stored checkpoint goes bad** — malformed, no longer decodable, or from
another context — the API answers `400` and the library raises
`InvalidCursorException` (a `ConnectionException` subclass), carrying the
rejected cursor and the server's own explanation. The recovery is specific, so
it is worth handling explicitly:

```python
from ciaops import InvalidCursorException

try:
    walk = imc.create_incidents_walk(sync_token=load_checkpoint())
    for portion in walk:
        ...
    save_checkpoint(walk.sync_token)
except InvalidCursorException as exc:
    logger.warning("IMC rejected %s: %s", exc.cursor_param, exc.server_message)
    drop_checkpoint()          # resync from scratch, optionally bounded by
                               # IMCIncidentFilter(updated_at_from=...)
```

Any other `400` keeps its type and carries the server's message, so a
rejected filter value says which one.

A corrupt **`pageToken`** behaves differently: the API answers `500` there
rather than `400`, so it is retried by the shared adapter and then surfaces as
a plain `ConnectionException` (about a minute of backoff). Only the sync cursor
— the one a connector actually persists — gets the clean `400` classification.

Omit `sync_token` for the first full pull, and use
`IMCIncidentFilter(updated_at_from=...)` to choose where that first sync starts.

Every page is also a `Parser`, so `set_keys()` mapping works exactly as for the
TI feeds:

```python
imc.set_keys(IMCCollections.INCIDENTS, {
    'number': 'number',
    'status': 'status',
    'rule':   'incidentRuleRevision.incidentRule.name',
    'score':  'incidentRuleRevision.incidentRule.scoring',
    'source': '*IMC',
})
portion.parse_portion()
# → [{"number": 42, "status": "INCIDENT_STATUS_TO_DO", "rule": "...", "score": 90, "source": "IMC"}]
```

#### Filtering incidents

Build the filter with `IMCIncidentFilter`. Repeated fields accept a single
value or a list (the API ORs a field's values). Enum fields take the value the
contract declares — use the constants:

```python
from ciaops import IMCIncidentFilter, IMCIncidentStatus

filters = IMCIncidentFilter(
    statuses=[IMCIncidentStatus.TO_DO, IMCIncidentStatus.PROCESSING],
    collections='darkWeb',                  # or IMCCollection.DARK_WEB, or 'dark-web'
    priorities=[1, 2],
    tags='vip',
    incident_rule_ids='22222222-2222-7222-8222-222200001234',
    assignee_user_ids=4242,
    scoring_from=70, scoring_to=100,        # 0-100
    updated_at_from='2026-06-26',           # promoted to 2026-06-26T00:00:00Z
)

portion = imc.search_updated_incidents(filters)
```

`IMCIncidentFilter` is **immutable**: frozen, with repeated fields held as
tuples. The API applies the filter on every request of a sync walk, cursor
requests included, so one that could change half-way through would silently
skip incidents that stopped matching. Build a new filter to change criteria.

Different fields combine with **AND**, values inside one field with **OR**, and
unset fields are not applied. The filter is flat even though the response
nests: `tags`, `priorities`, `incident_rule_ids`, `author_user_ids` and the
scoring bounds all match against the incident's **rule**.

The filter handles the two silent traps for you: filter parameters are sent
with the required `filter.` prefix (an unprefixed one is **ignored** and
returns the full set), and enums are sent by name (the integer form is a
`400`).

Enum values are matched **exactly** against what the contract declares:
use the `IMCIncidentStatus.*` / `IMCCollection.*` constants. A wrong filter
value is the one mistake the API does not report — it ignores the filter and
answers with every incident, so a value this library invented would read as
"no matches" or "everything matches" depending on the field, never as an
error. `collections` accepts all three spellings the contract itself declares
(`darkWeb`, `dark-web`, `COLLECTION_DARK_WEB`) because all three are the API's
own.

Values are validated on construction against the contract's own constraints —
pass `ignore_validation=True` to send them through as-is:

| Field | Constraint |
|---|---|
| `statuses`, `collections` | exactly a declared value — `IMCIncidentStatus.*`, `IMCCollection.*` (`collections` also takes `darkWeb` / `dark-web`); the `*_UNSPECIFIED` zero value is refused — the API answers `400` for it (omit the filter to match everything) |
| `priorities` | integers **1–10** inclusive — **P1 is the highest** priority, P10 the lowest |
| `assignee_user_ids`, `author_user_ids` | **positive** integers |
| `incident_rule_ids` | canonical **UUID**s |
| `scoring_from` / `scoring_to` | **0–100**, closed range `[from, to]`, `from ≤ to` |
| `updated_at_from` / `updated_at_to` | RFC 3339, **half-open** range `[from, to)` |

Because `updated_at_*` is half-open, adjacent windows never double-count — and
equal bounds select nothing.

> **One walk, one set of criteria.** The filter is applied on every request of
> a walk, the cursor ones included, and being frozen it cannot change under a
> walk in progress. What you must not
> do is *resume a stored cursor with different criteria*: the cursor only ever
> moves forward, so anything the old filter excluded stays excluded. New
> criteria mean a new sync without a cursor, bounded by `updated_at_from` if
> you do not want the whole backlog.

#### Reading an incident

```python
parser = imc.fetch_incident('11111111-....')     # the body *is* the incident
```

The id must be a canonical UUID — the contract declares it that way, so a
malformed id is rejected before the request instead of coming back as a `400`.

Wrap any incident — from a page or from `fetch_incident` — in `IMCIncident`
for the fields an integration maps to a notable or a case:

```python
from ciaops import IMCIncident

incident = IMCIncident.from_parser(parser)       # or IMCIncident(raw_response=raw)

incident.id                        # UUID — the only global identifier
incident.number                    # 42     (sequence *within the rule*, from 1)
incident.status                    # "INCIDENT_STATUS_TO_DO"
incident.is_open                   # True   (to do / processing / on hold)
incident.window_start              # start of the aggregation window
incident.window_end                # end of the aggregation window

incident.total_matches             # 250
incident.match_counts              # {"vulnerability": 250, "darkWeb": 0, ...}
incident.collections_with_matches  # ["vulnerability"]  ← drives the drill-down

incident.rule_id                   # the rule behind the incident
incident.rule_name                 # "Monitored vendor CVEs"
incident.rule_version              # revision the incident matched
incident.rule_is_current           # is that still the rule's live revision?
incident.rule_revision_created_at  # when the revision was created
incident.rule_created_at           # when the rule was created
incident.rule_updated_at           # when the rule last changed
incident.collection_queries        # {"vulnerability": "cvss_score: >7"}  (Lucene)
incident.priority                  # 2      (1-10; 1 is the highest priority)
incident.scoring                   # 90     (0-100, set on the rule by its author)
incident.tags                      # ["vip"]

incident.author                    # {"id": "4243", "email": "..."} — rule author
incident.author_email
incident.author_id
incident.rule_assignee_email       # the rule's *default* assignee
incident.rule_assignee_id
incident.assignee                  # {} when unassigned
incident.assignee_email            # this incident's assignee
incident.assignee_id

incident.has_ai_summary            # False on accounts without the AI summary
incident.ai_summary_level          # "LEVEL_HIGH" | ... | None
incident.portal_link               # the incident's page in the portal
incident.as_summary_dict()         # flat, alert-shaped payload, link included
```

`aiSummary` is present only when its generation succeeded, so it is commonly
absent. `has_ai_summary` tells you which case you are in — report it as
unavailable rather than substituting a summary of your own. It carries three
fields — `title`, `description` and `level`.

> **Rule fields drift.** `rule_name`, `priority`, `scoring` and `tags` come
> from the rule's *current* state, not from a snapshot taken when the incident
> opened. Editing a rule changes them for all of its incidents **without
> moving those incidents' `updatedAt`**, so a delta sync will not see it. Only
> the incident's own fields (status, assignee) move the cursor.
> `rule_is_current` tells you whether the incident still points at the rule's
> live revision.

#### Drilling into the evidence

`iter_incident_matches` walks every collection that actually fired, driven by
`matchCounts.byCollection` — a collection with a zero count is skipped instead
of costing a request that returns an empty page:

```python
for collection, page in imc.iter_incident_matches(incident):
    print(collection, page.portion_size, page.next_page_token)
    for info in page.match_infos:
        print(info['idInIncident'], info['collectionItemId'],
              info['status'], info['matchedAt'])
```

`page.match_ids` gives the `idInIncident` values (a match's identity inside the
incident) and `page.collection_item_ids` the pointers into the source
collection.

For one collection at a time — the walk object exposes the resume point and the
duplicate counter after iterating:

```python
walk = imc.create_matches_walk(
    incident_id=incident.id,
    collection='vulnerability',      # or 'vulnerabilities', or 'COLLECTION_VULNERABILITY'
)
for page in walk:
    ...
walk.page_token          # None once the last page has been read; a cursor to resume otherwise
walk.duplicates_dropped  # matches skipped because they had already been seen
```

`max_pages_without_new_matches` is a hang guard: after that many pages in a
row that brought nothing new, the walk warns and stops instead of following
the cursor further. `walk.stopped_early` records the stop. The default is 2;
`1` stops at the first one. With `dedupe` on it never lets a duplicate
through; what it can do is stop ahead of a cursor that kept yielding nothing
new, which no measured server does. `iter_incident_matches` passes it
through and reports such a stop.

Three collections the contract reserves — `ddosAttack`, `openThreat`,
`telegram` — have no endpoint in public v2. Asking for one gives you that
answer explicitly rather than "unknown collection".

`telegram` is not hypothetical: the portal offers Threats → Telegram as a rule
trigger, so a rule can raise incidents whose matches live in a collection this
API version cannot walk. Such a count is reported rather than dropped — see
`incident.unknown_collections` below.

`matchCounts.byCollection` and the match endpoints can disagree: an incident
can report matches whose endpoint answers `200` with an **empty page**. That is
observed on the live API, so `iter_incident_matches` compares a drained walk
with the reported count and warns rather than letting the empty result read as
"no evidence". Never infer that a walk is multi-page from a
count either — decide from `page.next_page_token`, because the page size is
server-controlled and not part of the contract.

`gitLeak` (Git leaks) is the eleventh collection: measured on the live API on
2026-09-16 — its match endpoint answers, its filter value is accepted, and its
key is present in `matchCounts.byCollection` — and walkable like the other ten
(`git-leaks`, `COLLECTION_GIT_LEAK`, resolved through `osi/git_repository`).
The vendored contract this build checks itself against predates it, which
`IMCCollections.SERVED_AHEAD_OF_CONTRACT` records until the contract is
refreshed.

A zero count is not evidence either way, so nothing warns and nothing changes;
a **non-zero** count in a collection this version cannot walk is reported
instead of dropped — the feed logs a warning once per walk,
`iter_incident_matches` logs one per incident, and
`incident.unknown_collections` exposes it as data (`{"telegram": 5}`). The
same goes for an incident reporting a non-zero `total` with no per-collection
breakdown to walk it by — the drill-down says so instead of returning an empty
result that looks like "no evidence".

> **Always page matches with `SORT_ORDER_DESC`** (the default).
> `SORT_ORDER_ASC` pagination is broken server-side — page two repeats page
> one and drops the cursor — so an ascending walk stalls at about one page.
> The generator stops on a cursor that does not advance instead of looping,
> and keeps following one that does even when a page held nothing new, up to
> `max_pages_without_new_matches` such pages in a row. Duplicates it drops
> are always logged.

#### Resolving matches to TI records

A match tells you *which* record matched, not what is in it. Resolve the
pointer with its collection's **by-id** route:

```python
record = imc.resolve_match('threat', 'fake11111111111111111111111111110001')
# GET /api/v2/hi/threat/fake1111...  → Parser (or None if the record is gone)
```

Or resolve a whole page, paced to the API limit:

```python
for collection, page in imc.iter_incident_matches(incident):
    for info, record in imc.iter_resolved_matches(collection, page.matches):
        if record is None:
            continue                       # the record aged out of its collection
        print(info['matchedAt'], record.raw_dict.get('title'))
```

**Resolution is the expensive step, and the cost is linear.** There is no batch
lookup and the limit is one request per second per account, so a full page of
~100 matches is about a minute and a 3,000-match incident is closer to an hour;
concurrency does not help, because the limit is server-side. Resolve what you
will act on — filter the matches first, or pass `limit`:

```python
for info, record in imc.iter_resolved_matches(collection, page.matches, limit=10):
    ...
```

The walk states the projected cost before spending it, and reports what a
`limit` left untouched — the count is never silently truncated.

`resolve_match` returns `None` for a record that no longer exists; pass
`missing_ok=False` to get a `NotFoundException` instead. The pointer is
percent-encoded before the request — it is server data going into a URL path —
and a page entry that is not a match is reported and yielded as
`({}, None)` rather than dropped, so results stay aligned with the input. Keys
set for the target TI collection apply to the resolved record:

```python
imc.set_keys('hi/threat', {'title': 'title', 'countries': 'countries'})
record.parse_portion()   # → [{"title": "...", "countries": ["US"]}]
```

Resolution routes per collection:

| Portal sub-section | IMC collection | Match path segment | Resolved with |
|---|---|---|---|
| Malware → Vulnerabilities | `vulnerability` | `vulnerabilities` | `osi/vulnerability/{id}` |
| Compromises → Accounts | `compromisedAccount` | `compromised-accounts` | `compromised/account_group/{id}` |
| Compromises → Bank cards | `compromisedCard` | `compromised-cards` | `compromised/bank_card_group/{id}` |
| Compromises → Masked bank cards | `compromisedMaskedCard` | `compromised-masked-cards` | `compromised/masked_card/{id}` |
| Compromises → Shops | `compromisedShop` | `compromised-shops` | `compromised/access/{id}` |
| Threats → Darkweb | `darkWeb` | `dark-web` | `darkweb/forums/{id}` |
| Compromises → Breached DB | `breachedDb` | `breached-db` | `compromised/breached/{id}` (singular) |
| Compromises → Suspicious payment details | `suspiciousPayment` | `suspicious-payments` | `compromised/spd/{id}` |
| Compromises → Public leaks | `publicLeak` | `public-leaks` | `osi/public_leak/{id}` |
| Threats → Last Threats | `threat` | `threats` | `hi/threat/{id}` |
| Git leaks | `gitLeak` | `git-leaks` | `osi/git_repository/{id}` |

The first column is what a rule author picks in the portal; the rest is what
the API calls the same thing. The pairing comes from the portal documentation's
sub-section names and the by-id routes above — the contract itself does not
carry the portal's labels.

`breachedDb` is the one collection whose feed and by-id route are spelled
differently: the feed is consumed as `compromised/breacheddb`, the newer
endpoint, which supports `/updated` and iterates on `seqUpdate`, while a single
record is read by id from singular `compromised/breached`. A resolved record is
parsed with the mapping set for the route it came from, so `set_keys` for one
of these does not apply to the other.

Resolved records inherit their collection's sensitivity — **mask passwords,
card numbers and leak payloads** before storing or displaying them.

### Close session

Always close the session in a `try…finally` block, or use the context manager:

```python
from ciaops import TIPoller
from ciaops.exception import InputException, ConnectionException

poller = TIPoller(username='example@group-ib.com', api_key='API_KEY', api_url='API_URL')
try:
    feed = poller.search_feed_by_id('apt/threat', 'abc123')
except InputException as e:
    logger.error("Wrong input: %s", e)
except ConnectionException as e:
    logger.error("Connection error: %s", e)
finally:
    poller.close_session()
```

<br>

## Parsing

Common example of API response from Collection (received feeds):

```python
api_response = [
    {
        'iocs': {
            'network': [
                {
                    'ip': [1, 2],
                    'url': 'url.com'
                },
                {
                    'ip': [3],
                    'url': ''
                }
            ]
        }
    },
    {
        'iocs': {
            'network': [
                {
                    'ip': [4, 5],
                    'url': 'new_url.com'
                }
            ]
        }
    }
]
```

### Parse portion method

Your mapping dict for `parse_portion()` or `bulk_parse_portion()` methods:

```python
mapping_keys = {
    'network': {'ips': 'iocs.network.ip'},
    'url': 'iocs.network.url',
    'type': '*custom_network'
}
```

Result of `parse_portion()` output:

```python
parsing_result = [
    {
        'network': {'ips': [[1, 2], [3]]},
        'url': ['url.com', ''],
        'type': 'custom_network'
    },
    {
        'network': {'ips': [[4, 5]]},
        'url': ['new_url.com'],
        'type': 'custom_network'
    }
]
```

Result of `bulk_parse_portion()` output:

```python
parsing_result = [
    [
        {
            'network': {'ips': [[1, 2], [3]]},
            'url': ['url.com', ''],
            'type': 'custom_network'}
    ],
    [
        {
            'network': {'ips': [[4, 5]]},
            'url': ['new_url.com'],
            'type': 'custom_network'}
    ]
]
```

### Get IoCs method

Your mapping dict for `get_iocs()` method:

```python
mapping_keys = {
    'ips': 'iocs.network.ip',
    'url': 'iocs.network.url'
}
```

Result of `get_iocs()` output:

```python
parsing_result = {
    'ips': [1, 2, 3, 4, 5],
    'url': ['url.com', 'new_url.com']
}
```

<br>

## Utilities

`ParserHelper` and `Validator` are standalone utilities available for use outside the generator flow — for example, when post-processing raw API responses or building custom pipelines on top of the library.

```python
from ciaops.utils import ParserHelper, Validator
```

### ParserHelper

#### find_by_template

Parse a single feed `dict` against a key-mapping template. Returns a `dict` with the resolved values.

```python
feed = {
    "id": "abc123",
    "evaluation": {"severity": "high"},
    "indicators": [{"params": {"ip": "1.2.3.4"}}, {"params": {"ip": "5.6.7.8"}}]
}

keys = {
    "feed_id":  "id",
    "severity": "evaluation.severity",
    "ips":      "indicators.params.ip",
    "source":   "*Group-IB",
}

result = ParserHelper.find_by_template(feed, keys)
# {
#     "feed_id":  "abc123",
#     "severity": "high",
#     "ips":      ["1.2.3.4", "5.6.7.8"],
#     "source":   "Group-IB",
# }
```

Supported value directives:

| Directive                                                                    | Example                           | Result                                                                     |
| ---------------------------------------------------------------------------- | --------------------------------- | -------------------------------------------------------------------------- |
| Dot-path string                                                              | `"evaluation.severity"`           | Value at that path                                                         |
| `"*literal"` (star prefix)                                                   | `"*Group-IB"`                     | The literal string `"Group-IB"`                                            |
| `"#field[N]"` (hash prefix)                                                  | `"#items[0]"`                     | Element at index N of the list found at `field`                            |
| Nested dict                                                                  | `{"ips": "indicators.params.ip"}` | Recursive template application                                             |
| `{"__nested_dot_path_to_list": "path", ...}`                                 | —                                 | Maps the inner template over each item in the list at `path`               |
| `{"__concatenate": {"static": "https://portal/?id=", "dynamic": "id"}}`      | —                                 | Concatenates a static prefix with a dynamic field value                    |
| `{"__concatenate": {"collection": "apt/threat", "dynamic": "id"}}`           | —                                 | Prefix auto-resolved from portal links for the given collection            |
| `{"__concatenate": {"parts": ["*https://portal/", "category", "*-", "id"]}}` | —                                 | Multi-part concatenation: `*` marks literals, bare strings are field paths |

Optional kwargs:

- `use_join_to_end_list=True` — joins list values into a single comma-separated string.
- `except_keys=["field"]` — excludes specific keys from the joining above.

#### find_element_by_key

Traverse any `dict` or `list` using a dot-notation path. Safe for nested lists and missing keys.

```python
from ciaops.utils import find_element_by_key

find_element_by_key({"a": {"b": 1}}, "a.b")
# → 1

find_element_by_key({"items": [{"ip": "1.2.3.4"}, {"ip": "5.6.7.8"}]}, "items.ip")
# → ["1.2.3.4", "5.6.7.8"]

find_element_by_key({"a": None}, "a.b")
# → None
```

#### unpack_iocs

Recursively flattens a nested list of IoC values into a single deduplicated list. Filters out noise values (`""`, `None`, `"0.0.0.0"`, `"255.255.255.255"`). Deduplication is set-based, so the order of the result is not guaranteed.

```python
raw = [["1.2.3.4", "5.6.7.8"], ["1.2.3.4", None, "0.0.0.0"]]
ParserHelper.unpack_iocs(raw)
# → ["1.2.3.4", "5.6.7.8"]
```

### Validator

`Validator` guards against invalid inputs before they reach the API.

```python
from ciaops.utils import Validator
```

**`validate_collection_name(collection_name, method=None)`** — raises `InputException` for unknown or removed collection names. A removed collection with a replacement names it; `method` is accepted for backward compatibility and has no effect.

```python
Validator.validate_collection_name("apt/threat")        # OK
Validator.validate_collection_name("compromised/reaper")
# raises InputException: has been removed. Use 'darkweb/forums' instead.
Validator.validate_collection_name("bp/phishing")
# raises InputException: Invalid collection name 'bp/phishing'.
```

**`validate_date_format(date, formats)`** — raises `InputException` if the date string does not match any of the provided format strings.

```python
Validator.validate_date_format("2024-01-15", ("%Y-%m-%d",))   # OK
Validator.validate_date_format("15/01/2024", ("%Y-%m-%d",))   # raises InputException
```

**`validate_ips_argument(ips)`** — normalizes and validates the `ips` argument for the `scoring/ip` endpoint. Accepts a single IP string or a list of IP strings; raises `InvalidIpsParameter` on invalid input. Returns a normalized list.

```python
Validator.validate_ips_argument("8.8.8.8")             # → ["8.8.8.8"]
Validator.validate_ips_argument(["8.8.8.8", "1.1.1.1"]) # → ["8.8.8.8", "1.1.1.1"]
Validator.validate_ips_argument("8.8.8.8,1.1.1.1")     # raises InvalidIpsParameter
```

**`validate_domains_argument(domains)`** — normalizes and validates the `domains` argument for the `scoring/domain` endpoint. Accepts a single domain string or a list of domain strings; raises `InvalidDomainsParameter` on invalid input. Returns a normalized list.

```python
Validator.validate_domains_argument("example.com")               # → ["example.com"]
Validator.validate_domains_argument(["example.com", "test.com"]) # → ["example.com", "test.com"]
Validator.validate_domains_argument("example.com,test.com")      # raises InvalidDomainsParameter
```

<br>

## Portal Links

`TICollections.PORTAL_LINKS` and `generate_portal_link` map collection records
to their Group-IB Portal URLs.

```python
from ciaops import TICollections, generate_portal_link
```

### Simple collections (single ID in URL)

Most collections use a plain prefix + record ID pattern:

```python
link = generate_portal_link('apt/threat', record_id='abc123')
# → "https://tap.group-ib.com/ta/last-threats?threat=abc123"

link = generate_portal_link('malware/config', record_id='def456')
# → "https://tap.group-ib.com/malware/configs?id=def456"

link = generate_portal_link('compromised/account_group', record_id='ghi789')
# → "https://tap.group-ib.com/cd/accounts?id=ghi789"

link = generate_portal_link('imc/incidents', record_id='11111111-1111-7111-8111-111100000701')
# → "https://tap.group-ib.com/imc/incidents?id=11111111-1111-7111-8111-111100000701"
```

IMC incidents have a portal page too; the prefix is in
`IMCCollections.PORTAL_LINKS`, and `generate_portal_link` reads every
product's table. `IMCIncident.portal_link` builds the link through
`IMCCollections.incident_portal_link`, which percent-encodes the id. For a
portal on another host, pass `url_prefix` to `generate_portal_link`.

Returns `None` when `record_id` is empty or the collection has no portal mapping.

### Multi-part URL templates

Some collections require multiple fields from the feed record (e.g. `compromised/messenger`, `compromised/discord`). Pass all required field values via the `fields` dict:

```python
link = generate_portal_link(
    'compromised/messenger',
    fields={'chatStat.id': '1234', 'id': '5678'},
)
# → "https://tap.group-ib.com/ta/im?chatId=1234&msg=5678"

link = generate_portal_link(
    'compromised/discord',
    fields={'channel.id': 'ch99', 'id': 'msg42'},
)
# → "https://tap.group-ib.com/ta/im?collection=discord&chatId=ch99&msg=msg42"
```

Returns `None` if any required field is missing or empty.

### Custom prefix override

```python
link = generate_portal_link('my/collection', record_id='001', url_prefix='https://tap.group-ib.com/feed?id=')
# → "https://tap.group-ib.com/feed?id=001"
```

### Embedding portal links in parsed output

Use the `__concatenate` directive in your mapping template so that `ParserHelper` resolves the URL automatically during parsing:

```python
keys = {
    'id':         'id',
    'title':      'title',
    'portal_url': {'__concatenate': {'collection': 'apt/threat', 'dynamic': 'id'}},
}
result = ParserHelper.find_by_template(feed, keys)
# result['portal_url'] → "https://tap.group-ib.com/ta/last-threats?threat=<id>"
```

### Inspecting the full URL map

```python
from ciaops import TICollections

for collection, template in TICollections.PORTAL_LINKS.items():
    print(collection, '->', template)
```

<br>

## Adapter utilities

`ConfigParser` and `FileHandler` are used by file-config based adapters such as the MISP adapter. They are not required for standard TI/DRP/ASM polling.

`FileHandler`'s YAML methods need the optional `pyaml` dependency — install it with `pip install ciaops[misp]`. Without it, only the YAML methods raise `ImportError`; the JSON methods and `ConfigParser` work regardless. The `save_data_to_*` methods update an **existing** file and raise `FileNotFoundError` rather than creating one.

```python
from ciaops.utils import ConfigParser, FileHandler
```

> **Note:** they live in `ciaops.utils`. They are not re-exported from the
> top-level `ciaops` package, and there is no `ciaops.adapters.misp_utils`
> module — the adapters package holds `misp_adapter` and `opencti_adapter`.

### ConfigParser

Parses YAML and JSON config files used by MISP-style adapters.

```python
cp = ConfigParser()

# Extract credentials from a YAML config dict as a dynamic Enum
creds = cp.get_creds(yaml_config)          # reads yaml_config["creds"]
creds = cp.get_creds(yaml_config, key="auth")  # custom key

creds.USERNAME.value   # "user@example.corp"
creds.API_KEY.value    # "abc123"
creds.API_URL.value    # "https://..."

# Get only enabled / disabled collections from YAML config
enabled  = ConfigParser.get_enabled_collections(yaml_config)   # ["apt/threat", ...]
disabled = ConfigParser.get_disabled_collections(yaml_config)

# Read a single collection's default_date
date = ConfigParser.get_collection_default_date(yaml_config, "apt/threat")
```

### FileHandler

A Borg-singleton file handler for reading and writing YAML and JSON config files. All instances share the same internal state, and reads and writes are serialized through a shared lock.

```python
fh = FileHandler()

# Check file existence / emptiness
fh.is_exist("/path/to/config.yml")   # True / False
fh.is_empty("/path/to/config.yml")   # True / False

# Read configs
yaml_config = fh.read_yaml_config("/path/to/config.yml")
json_config = fh.read_json_config("/path/to/mapping.json")

# Persist updated collection state back to YAML
fh.save_collection_info(
    config="/path/to/config.yml",
    collection="apt/threat",
    seqUpdate=16172928022293,
    default_date="2024-01-15",
)

# Overwrite an entire config file
fh.save_data_to_yaml_config(data, "/path/to/config.yml")
fh.save_data_to_json_config(data, "/path/to/mapping.json")
```

<br>

## Examples

### Full version of program

```python
import logging
from ciaops import TIPoller
from ciaops.exception import InputException, ConnectionException, ParserException

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
...

poller = TIPoller(username=username, api_key=api_key, api_url=api_url)
try:
    poller.set_proxies(proxy_protocol=PROXY_PROTOCOL,
                       proxy_port=PROXY_PORT,
                       proxy_ip=PROXY_ADDRESS,
                       proxy_password=PROXY_PASSWORD,
                       proxy_username=PROXY_USERNAME)
    poller.set_verify(True)

    for collection, keys in keys_config.items():
        poller.set_keys(collection, keys)

    for collection, state in update_generator_config.items():
        if state.get("sequpdate"):
            generator = poller.create_update_generator(
                collection_name=collection,
                sequpdate=state.get("sequpdate"),
            )
        elif state.get("date_from"):
            sequpdate = poller.get_seq_update_dict(
                date=state.get("date_from"), collection_name=collection
            ).get(collection)
            generator = poller.create_update_generator(
                collection_name=collection, sequpdate=sequpdate
            )
        else:
            continue

        for portion in generator:
            save_portion(portion.parse_portion())
            update_generator_config[collection]["sequpdate"] = portion.sequpdate

except InputException as e:
    logging.exception("Wrong input: {0}".format(e))
except ConnectionException as e:
    logging.exception("Something wrong with connection: {0}".format(e))
except ParserException as e:
    logging.exception("Exception occured during parsing: {0}".format(e))
finally:
    poller.close_session()
```

<br>

## API logic

To iterate over received portions from API response, you should follow one of the next iteration logic:

- **Result ID iteration** - based on `resultId` parameter, which was retrieved from previous response.
  Uses common collection name endpoint (`apt/threat`) which is added to the base URL >>> `/api/v2/apt/threat`.
- **Sequence update iteration** - based on `seqUpdate` parameter, which was retrieved from previous response.
  Uses updated endpoint (`/updated`) after collection name (`/apt/threat`) >>> `/api/v2/apt/threat/updated`.

To search IPs, domains, hashes, emails, etc., you should follow the next logic:

- **Search logic** -
  First you should reach `/api/v2/search` endpoint with any `q` parameter >>> `/api/v2/search?q=8.8.8.8`.
  In the output response you will receive collections, which contains the search result (`8.8.8.8`).
  Use _Sequence update iteration_ as a next step to retrieve all events.

To get the latest updates on each collection events you should follow the next logic:

- **Sequence update logic** -
  first you should reach `/api/v2/sequence_list` endpoint with `date` and `collection` parameters (optional) >>> `/api/v2/sequence_list?date=2022-01-01&collection=apt/threat`.
  In the output response you will receive `seqUpdate` number, which you should use in the next request to collection `/updated` endpoint.
  Use _Sequence update iteration_ as a next step to retrieve all events.

<br>

### Sequence update logic

Most of the collections at the Threat Intelligence portal has `/updated` endpoint.
And this endpoint uses updated logic based on `seqUpdate` key field, which comes from API JSON response.

The `seqUpdate` key – is a time from Epoch converted to a big number (microseconds), using the next formula:

```text
UTC timestamp * 1000 * 1000.
```

_Note:_ Don't rely on this formula. Because of the rising amount of data it could be changed.
For that purpose `/api/v2/sequence_list` endpoint was created.
Use this endpoint to get required `seqUpdate` number.

#### API response

Each row in our database has its own unique sequence update number. So, we can get all the events one by one.
To check it you can explore JSON output and then explore each item in the `"items"` field.
So, each item contains a `seqUpdate` field. And the last element’s `seqUpdate` is put to the top level of JSON output.
You can use it to get the next portion of feeds.
Each collection has its own updated route like `/api/v2/apt/threat/updated`, so we can use the next output as an example.

```json
{
    "count": 1761,
    "items": [
        {"id": "fake286ca753feed3476649438e4e4488"...},
        {"id": "fake51d29357b22b80564a1d2f9fc8751"...},
        {
            "author": null,
            "companyId": [],
            "id": "fake4f16300296d20ef9b909dc0d354fb",
            ......,
            "indicators": [
                {
                    "dateFirstSeen": null,
                    "dateLastSeen": null,
                    "deleted": false,
                    "description": null,
                    "domain": "example.corp",
                    "id": "fakebe483bb82759fbee7038235e0f52d0",
                    .....
                }
            ],
            "indicatorsIds": [
                "fakebe483bb82759fbee7038235e0f52d0"
            ],
            "isPublished": true,
            "isTailored": false,
            "labels": [],
            "langs": [
                "en"
            ],
            "malwareList": [],
            ......,
            "seqUpdate": 16172928022293
        },
    ],
    "seqUpdate": 16172928022293
}
```

#### Iteration steps

To iterate over `/api/v2/apt/threat/updated` endpoint data, you need to collect this
field number (`"seqUpdate": 16172928022293`) right at the top level of the JSON response,
received from previous request or from `/sequence_list` endpoint.

```console
curl -X 'GET' 'https://<base URL>/api/v2/sequence_list'
```

Add gathered `seqUpdate` in the next request, using endpoint params.

```console
curl -X 'GET' 'https://<base URL>/api/v2/apt/threat/updated?seqUpdate=16172928022293'
```

In the received JSON output check the `"count": 1751`. -> \
Gather `seqUpdate` from last feed or at top level -> \
Put it in next request ->

```console
curl -X 'GET' 'https://<base URL>/api/v2/apt/threat/updated?seqUpdate=16172928536227'
```

In the received JSON output, check the `"count": 1741` -> \
Gather `seqUpdate` from last feed or at top level -> \
Repeat till the end.

#### Stop the iteration

The "stop word" in that logic is items `"count"` or `"items"` list length.
For the collection `apt/threat` in above example, the `limit` is set to 10 by default,
the other collections usually have 100 `limit`. The limit depends on the amount of data to not overload the JSON output.
For example, usually you receive a portion of 100 feeds (not 10) for the first iteration. ->
Then could be a portion of 23 feeds -> Then a portion of 0 feeds -> The end.

<br>

### Search logic

Search logic is used to find attribution to the search value in Threat Intelligence database.

#### Global search

To find events related to IP, domain, hash, email, etc., you should send request to the `/api/v2/search` endpoint
with any `q` parameter (`/api/v2/search?q=8.8.8.8`).
It will return a list of collections, which contains this searchable parameter.
As a next step we need to use _Sequence update iteration_ over all items in each collection.
You can specify the searchable type keyword to avoid side results by setting `q` parameter like `/api/v2/search?q=ip:8.8.8.8`.
The same can be done for domain, email, hash, etc (`/api/v2/search?q=domain:example.corp`, `/api/v2/search?q=email:example@example.corp`).

```json
[
    {
        "apiPath": "suspicious_ip/open_proxy",
        "label": "Suspicious IP :: Open Proxy",
        "link": "https://<base-url>/api/v2/suspicious_ip/open_proxy?q=ip:8.8.8.8",
        "count": 14,
        "time": 0.304644684,
        "detailedLinks": null
    },
    {
        "apiPath": "attacks/ddos",
        "label": "Attack :: DDoS",
        "link": "https://<base-url>/api/v2/attacks/ddos?q=ip:8.8.8.8",
        "count": 1490,
        "time": 0.389418291,
        "detailedLinks": null
    },
    {"apiPath": "attacks/deface"...},
    {"apiPath": "malware/config"...},
    {"apiPath": "suspicious_ip/scanner"...}
]

```

#### Iteration steps

On the first search step we receive information that collection `attacks/ddos` contains 1490 items (`"count": 1490`).
Let's extract all of them. First we need to send request to this collection with the `q` parameter (`?q=ip:8.8.8.8`).
Then we retrieve `"seqUpdate"` field right at the top level of the JSON response and use it in the next request (`"seqUpdate": 1673373011294`).

```json
{
  "count": 1490,
  "items": [
    {
      "body": null,
      "cnc": {"cnc": "http://example.corp/drv/"...},
      "company": null,
      "companyId": null,
      "dateBegin": null,
      "dateEnd": null,
      "dateReg": "2017-08-16T00:00:00+00:00",
      "evaluation": {},
      "favouriteForCompanies": [],
      "headers": [],
      "hideForCompanies": [],
      "id": "examplec58903baddc84b8c51eaef1f904374025d",
      "isFavourite": false,
      ...
    }
  ],
  ...,
  "seqUpdate": 1673373011294
}
```

So the next request should look like this `/api/v2/attacks/ddos/updated?q=ip:8.8.8.8&seqUpdate=1673373011294`.
We can also set the `limit` parameter in the requests, like `limit=500`.
Explore the example below.

```console
curl -X 'GET' 'https://<base URL>/api/v2/search?q=ip:8.8.8.8'
```

Add gathered `seqUpdate` in the next request, using endpoint params.

```console
curl -X 'GET' 'https://<base URL>/api/v2/attacks/ddos/updated?q=ip:8.8.8.8&seqUpdate=1673373011294'
```

In the received JSON output check the `"count": 1390`. -> \
Gather `seqUpdate` from last feed or at top level -> \
Put it in next request ->

```console
curl -X 'GET' 'https://<base URL>/api/v2/attacks/ddos/updated?q=ip:8.8.8.8&seqUpdate=1673375930599'
```

In the received JSON output, check the `"count": 1290` -> \
Gather `seqUpdate` from last feed or at top level -> \
Repeat till the end.

#### Stop the iteration

The "stop word" in that logic is items `"count"` or `"items"` list length.
For the collection `attacks/ddos` in above example, the `limit` is set to 100 by default,
for other collections it may differ. The limit depends on the amount of data, to not overload the JSON output.
For example, usually you receive a portion of 100 feeds for the first iteration. ->
Then could be a portion of 23 feeds -> Then a portion of 0 feeds -> The end.

<br>

## Records limits

Default limit is 100 records per request. Due to different size of feeds there are different limits for getting data.

To change record limit in response add param `limit=500` to the request.
All limits for different collections can be found at Portal documentation.

```console
curl -X 'GET' 'https://<base URL>/api/v2/apt/threat/updated?limit=500&seqUpdate=16172928022293'
```

<br>

## Recommended TTL

TTL (Time To Live) is the maximum length of time an indicator or dataset (package) can exist. Calculated in days — during this period the platform guarantees that the data represents a valid, active IoC. Once the TTL expires the record should be considered stale and removed or re-evaluated. `None` means no expiry: the data does not have a defined lifetime and should be retained indefinitely.

| Endpoint                              | Recommended TTL (days) |
| ------------------------------------- | ---------------------- |
| **Threat Intelligence**               |                        |
| `apt/threat_actor/updated`            | 360                    |
| `darkweb/forums/updated`              | 90                     |
| `hi/threat_actor/updated`             | 360                    |
| `apt/threat/updated`                  | 360                    |
| `hi/threat/updated`                   | 360                    |
| `hi/open_threats/updated`             | None                   |
| `hi/analytic/updated`                 | None                   |
| **Malware**                           |                        |
| `malware/config/updated`              | 30                     |
| `malware/malware/updated`             | None                   |
| `malware/signature/updated`           | None                   |
| `malware/yara/updated`                | None                   |
| `malware/cnc/updated`                 | 90                     |
| **Attacks**                           |                        |
| `attacks/phishing_kit/updated`        | 30                     |
| `attacks/phishing_group/updated`      | 30                     |
| `attacks/ddos/updated`                | 30                     |
| `attacks/deface/updated`              | 30                     |
| **Vulnerabilities**                   |                        |
| `osi/vulnerability/updated`           | 30                     |
| **Compromised**                       |                        |
| `compromised/messenger/updated`       | None                   |
| `compromised/discord/updated`         | None                   |
| `compromised/access/updated`          | 90                     |
| `compromised/account_group/updated`   | 90                     |
| `compromised/breacheddb/updated`      | 90                     |
| `compromised/bank_card_group/updated` | 90                     |
| `compromised/masked_card/updated`     | 90                     |
| `compromised/spd/updated`             | 90                     |
| **OSI**                               |                        |
| `osi/public_leak/updated`             | 30                     |
| `osi/git_repository/updated`          | 30                     |
| **Suspicious IP**                     |                        |
| `suspicious_ip/tor_node/updated`      | 30                     |
| `suspicious_ip/open_proxy/updated`    | 15                     |
| `suspicious_ip/socks_proxy/updated`   | 2                      |
| `suspicious_ip/vpn/updated`           | 30                     |
| `suspicious_ip/scanner/updated`       | 15                     |
| **IoC**                               |                        |
| `ioc/common/updated`                  | 90                     |
| `ioc/primary/updated`                 | 90                     |

<br>

## Troubleshooting

### 401 response code

This code is returned when no credentials were sent. Make sure that you send the Authorization header and that you use Basic auth.

### 403 response code

There are several possible reasons for it:

- IP limitation. Make sure that you request from an allowed IP address; the allowlist is managed on the Portal profile page, Security and Access.
- API KEY issue. Make sure that your API KEY is active and valid, or regenerate it on the Portal profile page.
- No access to the feed. Make sure that you have access to the requested feed. You can find the available feeds on the Profile page -> Security and Access.

### 504 response code or timeout

Try setting a smaller limit when requesting the API.

## FAQ

Have a question? Ask in the SD Ticket on our Portal or integration@group-ib.com
