Metadata-Version: 2.4
Name: ciaops
Version: 3.3.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**. DRP is covered on both of its APIs: the client API v2 (`DRPV2Poller`) and the legacy client API (`DRPPoller`).

## **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)
    - [DRPV2Poller methods](#drpv2poller-methods)
      - [Syncing violations](#syncing-violations)
      - [Filtering violations](#filtering-violations)
      - [Reading a violation](#reading-a-violation)
      - [Counting violations](#counting-violations)
      - [Downloading captures](#downloading-captures)
      - [Companies, brands and tags](#companies-brands-and-tags)
      - [Writing](#writing)
      - [Errors](#errors)
      - [Moving from DRPPoller](#moving-from-drppoller)
    - [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, DRPV2Poller, ASMPoller, IMCPoller

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

# Digital Risk Protection — client API v2 (https://drp.group-ib.com/api/v2/)
drp = DRPV2Poller(username='example@example.corp', api_key='API_KEY', api_url='DRP_V2_API_URL')

# Digital Risk Protection — legacy client API (https://drp.group-ib.com/client_api/)
drp_legacy = 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 and the DRP client API v2) 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). The DRP client API v2
publishes no limit; its walks are paced to 1 request/second. 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, DRPV2Collections, ASMCollections,
    IMCCollections, DRPV2Status, DRPV2ViolationType,
)

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

# DRP, legacy client API
DRPCollections.VIOLATION          # "violation"
DRPCollections.COMPROMISED_DARKWEB  # "compromised/darkweb"

# DRP client API v2 — the logical collection you set keys for, and the
# vocabularies its filters and records share
DRPV2Collections.VIOLATIONS       # "drp/v2/violations"
DRPV2Status.LIVE                  # {"detected", "inResponse"}
DRPV2ViolationType.UPLOADABLE     # every type but "noViolation"

# 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

`DRPPoller` speaks the **legacy** DRP client API (`/client_api`). New
integrations should use the client API v2 and `DRPV2Poller` (next section);
the same credentials work for both.

```python
from ciaops import DRPPoller

drp = DRPPoller(username='...', api_key='...', api_url='https://drp.group-ib.com/client_api/')
```

**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, ...}
```

**A captured file by its hash**, such as a screenshot hash from a
violation's `images`:

```python
content = drp.download_file(sha256='<64 hex characters>')   # bytes
```

Store the bytes; never render a captured page.

### DRPV2Poller methods

**The DRP client API v2** (`/api/v2`) is the current Digital Risk Protection
API: one flat violation object, a cursor feed that delivers a violation
again when it changes, a `revision` per violation, counts in one call, bulk
writes, and one error envelope. It takes **the same credentials** as the
legacy client API — the portal email and the API key from the DRP profile —
against its own base URL. `DRPPoller` stays the client for the legacy
`/client_api`; the two APIs share no paths and no vocabulary, so each has its
own poller.

`DRPV2Poller` covers every operation of the contract — the feed, a single
violation, stats, both captures, companies and tags, upload, approve and
reject, comments, tag attach and detach — and is checked against the
vendor's OpenAPI document, so a change in the API surfaces as a failing
check rather than a silent gap.

```python
from ciaops import DRPV2Poller

drp = DRPV2Poller(
    username='example@example.corp',
    api_key='API_KEY',
    api_url='https://drp.group-ib.com/api/v2/',
)
```

Walks are paced to **one request per second**, through one limiter per
poller; `rate_limit_delay=0` turns that off when an outer scheduler already
paces the calls. Single-request methods do not sleep.

#### Syncing violations

The feed pages on an opaque cursor, `nextCursor`, present on **every**
response — the empty one that ends the feed included. Keep reading until a
page comes back empty; store that last cursor; next time pass it back **with
the same filter** and you receive only the violations that changed since. A
short page is not the end; only an empty one is.

```python
from ciaops import DRPV2ViolationFilter, DRPV2Status

criteria = DRPV2ViolationFilter(statuses=DRPV2Status.LIVE)   # detected, inResponse

walk = drp.create_violations_walk(criteria, cursor=load_checkpoint())
for page in walk:
    for violation in page.as_violations():
        upsert(violation.violation_id, violation)
    save_checkpoint(page.next_cursor)        # optional: checkpoint as you go
save_checkpoint(walk.cursor)                 # the caught-up checkpoint
```

`walk.cursor` moves past a page only when the loop asks for the next one.
While your code handles a page it still points at that page's start, so a
checkpoint saved from an `except` or `finally` block, or after a `break`,
delivers the page again next time instead of skipping what was not handled.

Or one page per poll, the shape a scheduled connector usually wants:

```python
page = drp.fetch_violations_page(criteria, cursor=load_checkpoint(), limit=500)
handle(page.as_violations())
save_checkpoint(page.next_cursor)            # also on an empty page
```

What to plan for:

- **The whole history by default.** When the API receives no `updatedFrom`
  it applies about the last month — now minus one calendar month — on every
  page, pages read from a stored cursor included, and the sync still ends on
  a normal empty page. `DRPV2Poller` therefore always sends a lower bound: the
  filter's `updated_from`, or `1970-01-01T00:00:00Z` when it is unset. Set
  `updated_from` to start a first sync later.
- **The filter goes out with every request.** The server applies it to each
  page, so a stored cursor must be read with the filter it was taken with.
  `DRPV2ViolationFilter` is frozen for that reason.
- **Upsert on `violation_id`.** A violation that changes is delivered again,
  later in the same walk too; keep the last delivery. `revision` is the
  number of history events, not a version of the content — diff the fields
  you use.
- **A newly matched signature does not move a violation in the feed.** Read
  the violation by id when its current state matters.
- **A violation that stops matching the filter does not come back** to say
  so. To track exits (a takedown that ended, say), compare snapshots by
  `violation_id`.
- `limit` (1-500, default 500) only changes how many requests a walk takes,
  never which violations come back.
- A page without its `items` list or its `nextCursor` raises
  `ParserException` and leaves the cursor where it was.

**Store the cursor as text, exactly as received.** The server compares
cursors as text and accepts any spelling of a UUID, so a copy that was
upper-cased, stripped of dashes or wrapped in braces — by a database column
type, for instance — resumes at another position without an error. Before
sending, a walk raises `InputException` for any string holding a UUID in a
form other than the issued one, whatever `ignore_validation` says. A cursor
the server cannot read at all raises `InvalidCursorException`:

```python
from ciaops import InvalidCursorException

try:
    walk = drp.create_violations_walk(criteria, cursor=load_checkpoint())
    for page in walk:
        ...
    save_checkpoint(walk.cursor)
except InvalidCursorException as exc:
    logger.warning("DRP rejected the stored cursor: %s", exc.server_message)
    drop_checkpoint()             # resync, bounded by criteria.updated_from
```

An empty first page from a stored cursor means caught up — or a cursor that
sorts after every position, such as a random UUID. To tell them apart,
compare what you hold with `get_violation_stats()` over the whole history,
with a `DRPV2StatsFilter` carrying the walk's `brand_ids`, `company_ids` and
`typosquatting`: `total`, or the `by_status` counts for the statuses you
filter on (stats ignores the status filter).

Every page is a `Parser`, so `set_keys()` mapping works as for the other
feeds. A walk takes the mapping set when it is created:

```python
from ciaops import DRPV2Collections

drp.set_keys(DRPV2Collections.VIOLATIONS, {
    'id':     'violationId',
    'url':    'uri',
    'status': 'status',
    'brand':  'brand.brandId',
})
for page in drp.create_violations_walk(criteria, cursor=load_checkpoint()):
    rows = page.parse_portion()
```

#### Filtering violations

| Field | Parameter | Values |
|---|---|---|
| `updated_from`, `updated_to` | `updatedFrom`, `updatedTo` | `datetime` (an aware one is converted to UTC), `date`, `'YYYY-MM-DD'`, RFC 3339 with `Z` or an offset |
| `statuses` | `status[]` | `DRPV2Status`: `found`, `detected`, `inResponse`, `false`, `legal`, `onTracking`, `noContent`, `onParking`, `resolved` |
| `types` | `type[]` | `DRPV2ViolationType`: `phishing`, `counterfeit`, `copyright`, `partnerPolicyCompliance`, `trademark`, `malware`, `scam`, `noViolation` |
| `approve_states` | `approveState[]` | `DRPV2ApproveState`: `notRequired`, `underReview`, `approved`, `rejected` |
| `section` | `section` | `DRPV2Section`: `web`, `mobileApp`, `marketplace`, `socialNetwork`, `advertising`, `messenger` |
| `typosquatting` | `typosquatting` | `DRPV2Typosquatting`: `exclude` (the server's default), `include`, `only` |
| `brand_ids`, `company_ids`, `tag_ids` | `brandIds[]`, `companyIds[]`, `tagIds[]` | ids from `get_companies()` and the records' `tag_ids` |

Different fields combine with AND, values of one field with OR. A repeated
field takes one value or any iterable; an empty one is refused rather than
read as "no filter", which would select everything. Values are checked
against the vocabulary before anything is sent — the server rejects an
unknown value with a `400` that does not say which — and a legacy client API
spelling is refused with the v2 token it maps to (`in_response` →
`inResponse`, `under_review` → `underReview`, `piracy` → `copyright`, the
boolean typosquatting flag `True` → `include`).

**How the bounds select.** The server reads a bound's date in UTC and
selects that whole day in the platform's time zone, which need not start at
midnight UTC; a time of day is dropped. By default (`exact_window=True`) the
filter treats the bounds as UTC moments instead — a date is its UTC midnight,
and an upper date includes its whole UTC day — and the walk asks the server
for a day more on each side: `updatedFrom` goes out as the day before the
lower bound's date (not before 1970-01-01, where the whole history starts)
and `updatedTo` as the day after the upper bound's. The
walk then returns every violation of the window, whatever time zone the
platform's days follow, and up to a day more on each side;
`DRPV2ViolationFilter.matches()` cuts a violation to the window, to the
second. It applies every other criterion to the violation's own fields too:

```python
criteria = DRPV2ViolationFilter(
    updated_from='2026-09-01T06:00:00Z', updated_to='2026-09-30'
)
in_window = [
    violation
    for page in drp.create_violations_walk(criteria)
    for violation in page.as_violations()
    if criteria.matches(violation)
]
```

Consecutive windows filtered this way — one UTC day each, say — deliver
every violation exactly once. With `exact_window=False` the bounds go out as
given and select the platform's own days, the ones `get_violation_stats()`
counts; consecutive date windows then tile without `matches()`, which
applies only a bound with a time of day. A bound with a time of day past the
edge of the platform's day does not bring back the violations beyond that
edge in that mode.

A violation can be present in several channels. `section` selects every
violation present in the channel named; each record lists its channels in
`sections`, and its own `section` is the primary one, which can be another.

A violation closed as `false` is never served, so filtering on that status
returns an empty page.

**An approval queue** is the `detected` violations whose `approve_state` is
`underReview`:

```python
from ciaops import DRPV2ApproveState

criteria = DRPV2ViolationFilter(
    statuses=DRPV2Status.DETECTED,
    approve_states=DRPV2ApproveState.UNDER_REVIEW,
    typosquatting='include',
)
queue = [
    violation
    for page in drp.create_violations_walk(criteria)
    for violation in page.as_violations()
]
```

**Scope ids come from `get_companies()`.** A brand outside the credential's
scope is a `403`; a company outside it returns an empty result, which reads
exactly like "no data".

#### Reading a violation

`DRPV2Violation` is a read view over the raw record (`raw_response`), with an
accessor for every field of the contract:

| Group | Accessors |
|---|---|
| identity | `violation_id`, `revision`, `created_at`, `updated_at`, `portal_link` |
| scope | `brand` (`brand_id`, `brand_name`), `company` (`company_id`, `company_name`), `brand_id`, `company_id` |
| verdict | `uri`, `type`, `section`, `sections`, `source`, `is_typosquatting`, `status`, `approve` (`state`, `is_required`), `approve_state` |
| detail | `scores`, `score('risk')`, `hunting_rules`, `whois` (`domain`, `ip`), `title`, `description`, `tags`, `tag_ids`, `images`, `comments` |
| captures | `screenshot_url`, `html_url`, `has_screenshot`, `has_html`, `capture_hashes(kind)` |
| history | `violation_history` (as served), `timeline` (a stable order), `latest_status_change` |
| shortcuts | `is_live` (`detected` or `inResponse`), `is_under_review`, `is_uploaded` |

Every nested block reads as `None` or empty when the server sends `null`,
and a name served as `""` (tag and hunting-rule names usually are) reads as
`None`. Resolve tag names by id from stats over the whole history in the
`include` view — `get_violation_stats(DRPV2StatsFilter(typosquatting='include')).tag_names`
— since `get_tags()` lists only some of them. A transition is in a history
event's `change` (`field`, `from_`, `to`); `details` is the action's payload.

`brand` and `company` are the customer's own identity; `title`,
`description`, comments, the whois registrant and tag names (a `vip` tag
names a monitored person) can name people; hunting-rule names carry the
customer's own wording. None of them is an indicator, and an integration
may want to mask them before they leave the customer's systems.

One violation by id, with its whole history:

```python
violation = drp.fetch_violation('9f2c…').violation      # NotFoundException if unknown
```

The parser returned by `fetch_violation()` honors the `set_keys()` mapping
too.

#### Counting violations

`get_violation_stats()` returns the total and its breakdowns — by status,
type, approve state, brand, company and tag — in one call. It honors five
filters — `updated_from`, `updated_to`, `brand_ids`, `company_ids` and
`typosquatting` — so it takes its own `DRPV2StatsFilter`:

```python
from ciaops import DRPV2StatsFilter

stats = drp.get_violation_stats(DRPV2StatsFilter(typosquatting='include'))
stats.total
stats.by_status['inResponse']      # every known status is a key, 0 if absent
stats.tag_names                    # tag_id -> name, for every tag attached
```

- It counts the violations the feed returns for the same window, brands,
  companies and view: its window selects on `updatedAt` too, and each
  breakdown matches the records' own fields. Each bound selects a whole
  day of the platform's calendar, as on the feed with `exact_window=False`;
  a time of day is ignored.
- To check what you hold against a count, walk the same bounds with
  `exact_window=False` to the empty page and call stats right after;
  violations that change in between move the two apart.
- `total` is the only grand total. `by_tag` counts attachments, so its rows
  overlap and do not sum to `total`.
- `typosquatting` is always 0 in the default `exclude` view.
- Like the feed, an unset `updated_from` counts the whole history.

#### Downloading captures

Most violations carry no capture, and the record says so: download only
when `has_screenshot` / `has_html` is true, and soon after first seeing the
violation, since a capture can disappear.

```python
if violation.has_screenshot:
    capture = drp.download_screenshot(violation.violation_id, missing_ok=True)
    if capture is not None:
        capture.media_type         # from the bytes: 'image/jpeg', 'image/png', …
        save(f"{capture.sha256}.{capture.extension}.bin", capture.content)

if violation.has_html:
    page = drp.download_html(violation.violation_id, missing_ok=True)   # DRPV2Capture
```

A `200` that is not a capture — not `application/octet-stream`, empty, or
named by another hash than its own, such as a sign-in page from a proxy —
raises `ParserException`, and a redirect raises `ConnectionException`.
`capture.matches(violation)` tells whether the capture belongs to that
violation and its history names the capture. For the capture's own
violation, `False` is no sign of a wrong capture: a history can record a
capture without naming its file.

A captured page is the abusive page verbatim: store the bytes and **never
render it**, re-serve it or inline it into a ticket, and defang any URL
taken from it. `screenshot_url` and `html_url` work only with the API
credential, so they are no link for a ticket either. The media type comes
from the bytes, because the server sends `application/octet-stream` and its
file names can lack an extension.

These routes serve the **current** capture. The hashes an `enrich` event of
the history names (`DRPV2HistoryEvent.files`) identify captures; they are
not download addresses.

#### Companies, brands and tags

```python
for company in drp.get_companies():
    company.company_id, company.brand_ids
    for license_ in company.licenses:
        license_.type, license_.status, license_.expires_at
    tags = drp.get_tags(company_id=company.company_id)
```

`licenses` reports what a company subscribes to and when it expires; it is
not a complete list of what the company may see, so never switch a feature
off because a type is missing from it. The tag catalog lists a subset of
the tags attached in the requested scope and changes over time: read it again
before attaching tags, and resolve tag names from stats' `tag_names` in the
`include` view. Leave the system tags out of a tag picker: `Added by
customer` marks an uploaded violation, and `Reported by customer` one whose
URL was uploaded again. A
`403` from `get_tags()` means a company or brand id outside the credential's
scope, or mistyped — not a problem with the key.

#### Writing

Every write is **sent once** — never retried by the library, because a
second comment or upload is a second write. Any failure after a write was
sent — a timeout, a `5xx`, a dropped connection, an answer that cannot be
read — leaves its outcome unknown. A write that carried an
`idempotency_key` can be sent again under the same key, which the API
applies once; read the violations back before resending one without a key.
Each write returns one outcome per item, read from the response
body: `200`, `207` and — for an upload in which nothing succeeded — `400`
all come back as a result object, not as an exception.

A write takes seconds to minutes to show on `fetch_violation()`, which can
meanwhile return the previous state or, for a new upload, `404`.
`wait_for_violations()` re-reads with backoff (up to ten minutes) until a
condition holds for every id, and `wait_for_violation()` does it for one:

```python
from ciaops import DRPV2ApproveState, DRPV2Status

fresh = [drp.fetch_violation(vid).violation for vid in selected]   # just before
pending = [
    v.violation_id
    for v in fresh
    if v.status == DRPV2Status.DETECTED and v.is_under_review
]
if pending:
    result = drp.approve_violations(pending)
    confirmed = drp.wait_for_violations(          # id -> violation, or None
        result.succeeded_ids or pending,
        lambda v: v.approve_state == DRPV2ApproveState.APPROVED,
    )
```

`upload_violations()` registers 1-100 URLs or domains, each for one of
your brands. There is no delete: an upload is permanent. A new violation
starts as `found`, awaiting review, and becomes `detected` once confirmed.
A URL already known under the brand creates nothing: the violation keeps
its status, type and approval, and gains the system tag `Reported by
customer` once.

```python
from ciaops import DRPV2UploadItem

result = drp.upload_violations([
    DRPV2UploadItem(url='https://brand-lookalike.example.com', type='phishing', brand_id=brand_id),
])
result.violation_ids          # new and already known
result.errors                 # (url, brand_id) -> forbidden, invalidItem, …
for row in result.succeeded:
    row.violation_id, row.existing, row.status    # status: a known one's
```

`type` is any violation type but `noViolation`. Two items naming the identical
URL and brand with different types are refused before sending. URLs are
compared as written: the server de-duplicates on a normalized URL, so two
spellings of one address can still name one violation.

A row's `existing` tells a new violation from a known one. For a known one,
`type` is the type sent, not the violation's, and `status` is its current
status when the API reports it. A row's `url` is the server's echo (a bare
domain comes back as `//domain`), not a key; to tie an id to an item for
certain, upload that item alone. The same items sent again under the same
`idempotency_key` within 24 hours return the first answer and create
nothing.

`approve_violations(ids)` and `reject_violations(ids)` take 1-100 ids. Both
act only on a violation whose `status` is `detected`; any other status
fails with `notUnderReview`. On a `detected` violation, by its
`approve_state`:

| `approve_state` | `approve_violations` | `reject_violations` |
|---|---|---|
| `underReview` | approved | rejected |
| `notRequired` | fails with `notUnderReview` | **rejected** |
| `rejected` | **reversed** to approved | no change |
| `approved` | no change | **reversed** to rejected |

A stale selection can reverse a decision: read the violations again just
before the call and send only the ones still awaiting a decision. Approving
does not change `status`.

`comment_violations(ids, comment=text)` appends the same text to each
violation. A comment cannot be edited or deleted, and a comment posted
through the API is `public`: every portal user of the tenant can read it.
Pass an `idempotency_key` — one fresh value per comment, the same value on
every retry of it: a repeat under the key posts nothing and reports the
first comment.

```python
import uuid

key = str(uuid.uuid4())                 # keep it with the pending comment
drp.comment_violations([violation_id], comment='Takedown requested', idempotency_key=key)
```

`attach_tags(ids, tag_ids)` refuses the whole request (`400 badRequest`,
nothing applied) for a tag id the server does not know at all; it does not
enforce the `get_tags()` list, so check against a fresh one yourself.
Attaching a tag a violation already carries, or detaching one it lacks,
changes nothing.
`detach_tags(ids, tag_ids)` takes the ids from the violations' own `tag_ids`,
since `get_tags()` omits tags that violations carry.

The violation records no author of a write, so log your own audit trail.
The library logs, at `INFO` under the `ciaops.pollers.drp_v2` logger
hierarchy, the ids each write
succeeded for (for an upload, the type and brand of a new violation and the
status of a known one), and at `WARNING` what
failed — for an upload the counts per code, the failed URLs only at `DEBUG`,
since a reported URL can carry a victim's address.

#### Errors

| Response | Raised |
|---|---|
| `400 invalidCursor` | `InvalidCursorException` (`cursor`, `server_message`) |
| `404 notFound` | `NotFoundException` — an unknown id, a violation the credential cannot see, or a missing capture |
| an answer of the wrong shape | `ParserException` — a page without `items` or `nextCursor`, stats without `total`, a write answer without its lists, a `200` that is not a capture |
| any other failure | `ConnectionException` |

`ParserException` is not a `ConnectionException`; after a write, treat both
as "outcome unknown". Every `ConnectionException` carries `status_code`
(`None` when no response arrived) and, when the API sent one, `error_code` —
the stable code from the error envelope (`DRPV2ErrorCode`: `badRequest`,
`unauthenticated`, `forbidden`, `notFound`, `rateLimited`, `internal`).
Branch on `error_code`, never on the message text. A `403 forbidden` on a
request naming a brand or company id means that id is outside the
credential's scope or mistyped. A `404` without a code means the route does
not exist — usually an `api_url` that is not the `/api/v2/` base — and is a
plain `ConnectionException`, so `missing_ok=True` does not hide it.

#### Moving from DRPPoller

| `DRPPoller` (legacy `/client_api`) | `DRPV2Poller` (`/api/v2`) |
|---|---|
| `create_update_generator('violation', ...)` on `seqUpdate` | `create_violations_walk(criteria, cursor=...)` on `nextCursor` |
| `get_seq_update_dict(date)` | `DRPV2ViolationFilter(updated_from=date)` for the first run |
| `search_feed_by_id(id)` | `fetch_violation(id)` |
| `change_status(id, 'approve' / 'reject')` | `approve_violations([id])` / `reject_violations([id])` |
| `add_violations([{url, violationSubtype, brandId}])` | `upload_violations([DRPV2UploadItem(url, type, brand_id)])` |
| `get_brands()`, `get_subscriptions()` | `get_companies()` — brands and licenses with expiry |
| — | `get_violation_stats()`, `get_tags()`, `comment_violations()`, `attach_tags()` / `detach_tags()`, `download_screenshot()` / `download_html()` |

The vocabulary is camelCase and the same in filters and records:
`approve_states=['under_review']` becomes `approve_states=['underReview']`,
the numeric `subtypes` and `section` codes become `types` and `section`
tokens (`6` → `'phishing'`, `1` → `'web'`), and `use_typo_squatting=True`
becomes `typosquatting='include'`. The legacy `violationSubtype` is the v2
`type` (`piracy` reads as `copyright`); the legacy `violationType` has no v2
counterpart. A v2 cursor and a legacy `seqUpdate` are not interchangeable:
start the v2 sync with `updated_from`.

### 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,                      # 1..5000; refused outside that range
        status=['new', 'confirmed'],    # optional filter
        type=['domain', 'ip'],          # optional filter (assets only)
        discover=['auto', 'cloud'],     # assets only: auto|enrich|manual|cloud
        has_issues=['crit', 'high'],    # assets only: crit|high|medium|low|clean
    ),
)
for portion in generator:
    data = portion.parse_portion()
```

Fixed-vocabulary filters (`severity`, `asset_discovered`, `asset_status`,
`discover`, `has_issues`) and `count` are validated on construction and raise
`ValueError`; pass `ignore_validation=True` to send a value the library does
not know. Note the two discovery filters have different vocabularies: the
issues feed's `asset_discovered` accepts `integration`, the assets feed's
`discover` does not.

**What a portion holds.** The three feeds answer in one of two envelope
shapes: the flat `items: [...]` the API documents, or the recorded one where
`items` is an object keyed by company id (`items[companyId].items[type][status][]`
for assets, `items[companyId].items[status][]` for issues,
`items[companyId].items[]` for leaks). Each portion is an `ASMParser`, a
`Parser` that accepts both and flattens the nested form, so `parse_portion()`,
`get_iocs()` and `portion_size` see records, not company ids:

```python
for portion in generator:
    records = portion.records         # flat list of record dicts, server order
    for company_id, block in portion.by_company.items():
        # the company-keyed object as the server sent it ({} for the flat
        # form); the only place a leak record's company is available
        print(company_id, block.get('companyName'))
    if portion.skipped:
        pass  # entries that were not records were dropped; already logged
```

`portion.count` is the server's own top-level count and is not comparable
with `portion_size`: on the live API it is a cumulative string such as
`"10 of 46"` (records served so far of the total), and a recorded nested page
reports 30 against one record. A page
that yields no records while the server reports a positive count, or that
carried entries the parser had to skip, is logged at WARNING rather than
passed off as a normal end of feed.

**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')

# Several issues in one call. Issue types the bulk endpoint does not serve
# come back with supported=false and are logged; fetch those one by one.
bulk = asm.get_issues_evidence_bulk(issue_ids=['issue-uuid-1', 'issue-uuid-2'])

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
)
```

The write direction is title case: `change_issue_status` refuses the lowercase
form the `issues/updated` `status` filter takes (`ASMCollections.ISSUE_STATUS_OPTIONS`
holds the accepted values). Pass `ignore_validation=True` to send a status the
library does not know.

**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  ("false" is the string)
)

seed = asm.get_seed_assets(company_id='company-uuid')
# → {"name": ..., "confirmedDomain": [...], "excludedDomain": [...],
#    "confirmedIP": [...], "excludedIP": [...]} — the curated lists that
#    add_assets/remove_assets edit, not the discovered assets
```

`change_asset_status` sends the body key the API takes, `assetsId`; releases
up to 3.2.0 sent `assetsIds`.

**Clients, companies, partners, logs, reports** — read-only, each returning
the response object as the API sends it:

```python
asm.get_clients()                      # {count, items: [{id, name, industry, country, status, managerEmail}]}
asm.get_client(client_id='...')        # one client object
asm.get_partners()                     # {count, items: [{id, name, subscribeStart, ...}]}
asm.get_companies(client_id='...')     # the companies list, narrowed to one client
asm.get_company(company_id='...')      # settings: name, website, branches, licenseType, activeScan, ...
asm.get_company_logs(company_id='...', date_from='2025-01-01', date_to='2025-01-31')
asm.get_user_logs(date_from='2025-01-01')   # the authenticating user's own log
asm.get_reports(company_id='...')      # {count, items: [{id, score, timestamp}]}
asm.get_last_report(company_id='...')  # the latest full report
asm.get_report(report_id='...')        # one full report, by *report* id
```

Report objects are camelCase at the top level (`companyId`, `scanDate`,
`rootIndicators`, `detailedList`, `totalChecks`) with snake_case keys inside
(`actual_score`, `first_seen`, `ssl_count`); the vendor's written docs spell
the top level in snake_case, the API does not. In `get_company_logs` the
`items` value is an object keyed by company id, not a list.

Every id that goes into a URL path is validated and percent-encoded, and every
method raises `ParserException` if the body is not a JSON object, so a
misrouted or malformed response is reported against the endpoint that
produced it rather than as an `AttributeError` later on.

### 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
