diff --git a/README.md b/README.md index 5b92c77e2..58836728f 100644 --- a/README.md +++ b/README.md @@ -202,6 +202,7 @@ For further Information see the [license file](https://misp.github.io/misp-modul * [Nexthink NXQL Export](https://misp.github.io/misp-modules/export_mod/#nexthink-nxql-export) - Nexthink NXQL query export module * [OSQuery Export](https://misp.github.io/misp-modules/export_mod/#osquery-export) - OSQuery export of a MISP event. * [Event to PDF Export](https://misp.github.io/misp-modules/export_mod/#event-to-pdf-export) - Simple export of a MISP event to PDF. +* [QRadar Reference Set Export](https://misp.github.io/misp-modules/export_mod/#qradar-reference-set-export) - Export deduplicated, typed MISP indicators as QRadar reference-set CSV or JSON values. * [Splunk Lookup Export](https://misp.github.io/misp-modules/export_mod/#splunk-lookup-export) - Export event and object indicators as a Splunk CSV lookup with CIM field mapping hints. * [Test Export](https://misp.github.io/misp-modules/export_mod/#test-export) - Skeleton export module. * [ThreatStream Export](https://misp.github.io/misp-modules/export_mod/#threatstream-export) - Module to export a structured CSV file for uploading to threatStream. @@ -235,6 +236,7 @@ For further Information see the [license file](https://misp.github.io/misp-modul * [Export to Sentinel or Defender](https://misp.github.io/misp-modules/action_mod/#export-to-sentinel-or-defender) - Export indicators to Microsoft Sentinel or Microsoft Defender. Requires an existing installation of MISP2Sentinel or MISP2Defender. * [Mattermost](https://misp.github.io/misp-modules/action_mod/#mattermost) - Simplistic module to send message to a Mattermost channel. * [Nextcloud talk](https://misp.github.io/misp-modules/action_mod/#nextcloud-talk) - Simplistic module to send a message to a Nextcloud talk conversation. +* [QRadar Reference Set Publish](https://misp.github.io/misp-modules/action_mod/#qradar-reference-set-publish) - Publish typed MISP indicators to an existing QRadar reference set and explicit domain. * [SentinelOne IOC Publish](https://misp.github.io/misp-modules/action_mod/#sentinelone-ioc-publish) - Publish MISP event and object indicators to SentinelOne Threat Intelligence with explicit scope. * [Slack](https://misp.github.io/misp-modules/action_mod/#slack) - Simplistic module to send messages to a Slack channel. * [Test action](https://misp.github.io/misp-modules/action_mod/#test-action) - This module is merely a test, always returning true. Triggers on event publishing. diff --git a/documentation/README.md b/documentation/README.md index 366ee18bf..bfaad23cb 100644 --- a/documentation/README.md +++ b/documentation/README.md @@ -3749,6 +3749,30 @@ Simple export of a MISP event to PDF. ----- +#### [QRadar Reference Set Export](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/export_mod/qradar_reference_set_export.py) + +Export deduplicated, typed MISP indicators as QRadar reference-set CSV or JSON values. +[[source code](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/export_mod/qradar_reference_set_export.py)] + +- **features**: +>Exports one selected indicator type from supplied event and object attributes, with validation, normalization and deduplication. Produces a headerless single-column CSV for Reference Set Management, or a JSON value array. Defaults to IDS-marked IPv4 addresses; IPv6 uses a separate text set. No QRadar connection is required. See the [usage guide](https://github.com/MISP/misp-modules/blob/main/documentation/export_mod/qradar_reference_set_export.md). + +- **config**: +> - indicator_type +> - format +> - include_non_ids + +- **input**: +>MISP events, including object attributes, in standard, bare or legacy format. + +- **output**: +>Base64-encoded UTF-8 CSV or JSON; counts and provenance are returned separately from the file. + +- **references**: +>https://www.ibm.com/docs/en/qradar-on-cloud?topic=overview-importing-iocs-reference-set + +----- + #### [Splunk Lookup Export](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/export_mod/splunk_export.py) Export event and object indicators as a Splunk CSV lookup with CIM field mapping hints. @@ -4352,6 +4376,33 @@ Simplistic module to send a message to a Nextcloud talk conversation. ----- +#### [QRadar Reference Set Publish](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/action_mod/qradar_reference_set_publish.py) + +Publish typed MISP indicators to an existing QRadar reference set and explicit domain. +[[source code](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/action_mod/qradar_reference_set_publish.py)] + +- **features**: +>Adds or refreshes IDS-marked event and object indicators through reference_data_collections/set_entries. Requires an existing reference-set ID and explicit domain ID, validates the target entry type and checks each write acknowledgement. Supports credential-free dry runs and reports partial failures. Set expiry remains controlled by QRadar. Credentials come from the module server environment. See the [usage guide](https://github.com/MISP/misp-modules/blob/main/documentation/action_mod/qradar_reference_set_publish.md). + +- **config**: +>{'params': {'reference_set_id': {'type': 'string', 'description': 'Required: existing QRadar reference-set numeric ID'}, 'domain_id': {'type': 'string', 'description': 'Required: numeric domain ID (0 is Default Domain), or SHARED'}, 'indicator_type': {'type': 'string', 'description': 'ipv4, ipv6, domain, url, md5, sha1, sha256 or sha512', 'value': 'ipv4'}, 'source': {'type': 'string', 'description': 'QRadar entry source label', 'value': 'misp-modules'}, 'dry_run': {'type': 'string', 'description': 'true previews entry payloads without network access', 'value': 'false'}}, 'blocking': True, 'support_filters': False, 'expect_misp_core_format': True} + +- **input**: +>Selected MISP events and object attributes in workflow data. + +- **output**: +>Counts and per-value created, updated, planned or failed outcomes with MISP provenance. + +- **references**: +> - https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-set_entries-POST.html +> - https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-sets-id-GET.html + +- **requirements**: +> - requests +> - QRadar API 20.0 and an authorized service token with reference-data access + +----- + #### [SentinelOne IOC Publish](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/action_mod/sentinelone_ioc_publish.py) Publish MISP event and object indicators to SentinelOne Threat Intelligence with explicit scope. diff --git a/documentation/action_mod/qradar_reference_set_publish.md b/documentation/action_mod/qradar_reference_set_publish.md new file mode 100644 index 000000000..325b06bc5 --- /dev/null +++ b/documentation/action_mod/qradar_reference_set_publish.md @@ -0,0 +1,170 @@ +# QRadar Reference Set Publish + +`qradar_reference_set_publish` is a blocking workflow action that adds or refreshes +selected MISP indicators in an existing QRadar reference set and an explicit domain. +It uses the current **API 20.0** `reference_data_collections` endpoints. It shares +validation and normalization with [QRadar Reference Set Export](../export_mod/qradar_reference_set_export.md), +which produces offline CSV or JSON files. Different names allow both modules to +appear in the module server's registry. + +## Setup + +1. Create a dedicated reference set in QRadar with a compatible entry type, and + configure its expiry policy. Find its numeric ID through QRadar's API explorer + at `/api_doc`, using `GET /reference_data_collections/sets`. +2. Create an authorized service token whose role and security profile permit + reading the selected set and adding/updating entries in the intended domain. + Confirm that your QRadar installation supports API 20.0. +3. Configure `QRADAR_URL` as an HTTPS console origin, for example + `https://qradar.example`, and `QRADAR_API_TOKEN` in the environment of the + **misp-modules service**. Restart that service. The URL must not contain + credentials, an API path, a query or a fragment. +4. For a private certificate authority, optionally set `QRADAR_CA_BUNDLE` to a PEM + CA bundle accessible to the service. TLS verification remains enabled; the + module provides no option to disable it. +5. Enable **QRadar Reference Set Publish** and add it to a workflow supplying the + intended events and object attributes. Preview with `dry_run=true`, then set it + to `false` to publish. Dry runs need no credentials and cannot verify set types, + domain permissions or existing values. + +Credentials, console URL and CA trust are server settings. Workflow parameters +cannot change them. The token is sent in the `SEC` header and the API version is +explicitly pinned with `Version: 20.0`. + +| Parameter | Default | Meaning | +| --- | --- | --- | +| `reference_set_id` | Required | Positive numeric ID of an existing reference set. Names and arbitrary paths are not accepted. | +| `domain_id` | Required | Explicit non-negative domain ID, or `SHARED` to publish shared reference data. `0` is QRadar's Default Domain, distinct from shared data. `SHARED` is sent as JSON `null`; neither scope is chosen implicitly. | +| `indicator_type` | `ipv4` | `ipv4`, `ipv6`, `domain`, `url`, `md5`, `sha1`, `sha256` or `sha512`. | +| `source` | `misp-modules` | Entry source label, 1–64 ASCII letters, digits, spaces or `._:/-`, starting with a letter or digit. | +| `dry_run` | `false` | Preview per-entry JSON payloads without network access. | + +Boolean values accept JSON booleans, integers 0/1 and strings `true`, `false`, `1`, +`0`. Empty optional settings use defaults. IDs accept integers or digit strings. +MISP displays node defaults as placeholders; the module also applies them when +values are omitted. Shared reference data can be visible across tenants; select +`SHARED` only when that is the intended scope. + +## Input and set compatibility + +The action accepts standard, bare or legacy event envelopes, singly or in a list, +and traverses their object attributes. Only non-deleted `to_ids=true` attributes +are eligible. Deleted events and objects are excluded. No non-IDS override is +available for publication. Only supplied attributes are considered; missing data +is not fetched from MISP. + +The selected type, normalization and supported composites are described in the +[export guide](../export_mod/qradar_reference_set_export.md#supported-indicators-and-target-types). +Before writing anything, the action fetches `GET /reference_data_collections/sets/{id}` +and confirms the returned set ID and `entry_type`: + +| Selected type | Required set entry type | +| --- | --- | +| `ipv4` | `IP` | +| `ipv6`, `url` | `ALN` | +| `domain`, `md5`, `sha1`, `sha256`, `sha512` | `ALN` or `ALNIC` | + +This prevents IPv6 from being sent to an IPv4 set and URLs from being stored in a +case-insensitive set. A text set cannot identify its intended semantic use, so use +dedicated sets and select the intended type. The caller must have access to the +chosen collection and domain; QRadar validates that authorization on each write. + +Unsupported, malformed and other-type attributes are skipped with counts. +Canonical values are deduplicated within an invocation. All contributing MISP +event, attribute and object references remain in the returned `sources` lists. +Only value, collection ID, domain ID and the configured source label are sent to +QRadar. Comments, tags, event titles, provenance notes and sample data are not sent. +Select inputs according to MISP sharing and handling policy. + +The action advertises `support_filters=false`. Payloads containing `filteredItems` +or `matchingData` fail rather than silently publishing unfiltered events. Select +the intended data upstream. + +## Publication and lifecycle + +Each value is submitted using `POST /reference_data_collections/set_entries`: + +```json +{"collection_id": 42, "domain_id": 7, "source": "misp-modules", "value": "192.0.2.1"} +``` + +QRadar returns HTTP **201** for a created entry and **200** for an updated entry. +The action counts success only when the returned numeric entry ID, collection ID, +domain ID, value and source confirm the submitted payload. Missing or mismatched +acknowledgements fail with an unknown-outcome message. A 202 asynchronous response +is not treated as success; the action uses synchronous per-entry requests. + +This is an add/update operation on the explicitly chosen set. An existing matching +value can be updated even if another publisher supplied it, including its source +metadata. The source label is not an ownership guard. Use a dedicated set/domain +when other publishers must remain independent. Repeated publication follows +QRadar's entry update behavior and the set's existing `FIRST_SEEN`, `LAST_SEEN` or +`NO_EXPIRY` policy. The module does not set timestamps or promise a per-IOC lifetime. + +The action does not create sets, change their type or TTL, purge values, or delete +entries. Removing an attribute from MISP or marking it non-IDS does not withdraw +it from QRadar. Use the set's configured expiry or separate administration for +withdrawal. Reference-set membership takes effect in the rules configured to use +that set; publication does not create QRadar rules or automatically block traffic. + +## Workflow/API example + +POST to the misp-modules `/query` endpoint for a preview: + +```json +{ + "module": "qradar_reference_set_publish", + "params": { + "reference_set_id": "42", + "domain_id": "7", + "indicator_type": "ipv4", + "source": "MISP production", + "dry_run": "true" + }, + "data": { + "Event": { + "uuid": "f1ef0409-c80a-4012-b401-b7d1050f4ed4", + "Attribute": [{ + "uuid": "aa09819b-f2d8-4661-83ea-d4897cafebc8", + "type": "ip-dst", + "value": "192.0.2.1", + "to_ids": true + }] + } + } +} +``` + +Responses contain `result_key=qradar_reference_set_publish` and a `data` summary +with `success`, selected target/type/source, `dry_run`, `created`, `updated`, +`planned`, `failed`, `skipped` and `items`. Items identify each value, its sources +and status; live successes include the remote entry `id`, and dry-run items include +the planned payload. Successful summaries are available under +`_env.misp_module_results.qradar_reference_set_publish`. + +Preflight target failures return a top-level `error` with `success=false`, before +any value is attempted. Per-entry failures also return `error`, preserve completed +outcomes and set `success=false`. **Publication may be partial.** The blocking +action stops dependent workflow execution on failure. MISP may retain only the +top-level error on failure; standalone API callers receive the full summary. +HTTP errors, redirects, invalid JSON and malformed acknowledgements are reported +without returning provider error bodies or exception text. + +## Limits and validation + +At most **100 unique selected values** are accepted per action invocation; larger +inputs fail before network access. Split larger selections. Requests use a +15-second timeout; a 240-second processing budget is checked after the target +lookup and before each write. TLS is verified and redirects are disabled. Rate +limits and uncertain responses fail without automatic retries or rollback. +Check QRadar before retrying an uncertain write. + +Tests cover normalization, real `requests` serialization through mocked HTTP +transport, discovery/query endpoints, target validation, acknowledgements and +partial failures. No live QRadar tenant was used; verify permissions, set expiry +and deployment-specific behavior before enabling publication. Existing MISP +`file` and `domain-ip` objects suffice; no new templates are needed. + +References: [API 20.0 entry creation/update](https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-set_entries-POST.html), +[set metadata](https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-sets-id-GET.html), +and [authorized service authentication](https://ibmsecuritydocs.github.io/qradar_api_overview/c_rest_api_getting_started.html). diff --git a/documentation/export_mod/qradar_reference_set_export.md b/documentation/export_mod/qradar_reference_set_export.md new file mode 100644 index 000000000..58e3df706 --- /dev/null +++ b/documentation/export_mod/qradar_reference_set_export.md @@ -0,0 +1,106 @@ +# QRadar Reference Set Export + +`qradar_reference_set_export` produces a deduplicated list of one selected indicator +type for a QRadar reference set. It exports values from supplied MISP events and +objects, without contacting QRadar or requiring credentials. For direct workflow +publication, use [QRadar Reference Set Publish](../action_mod/qradar_reference_set_publish.md). +The two modules have distinct names because misp-modules registers module names +across all module families. + +## Configuration and formats + +Enable **QRadar Reference Set Export** in MISP's export-module settings. Set the +module configuration there, or supply `config` when calling `/query` directly: + +| Configuration | Default | Meaning | +| --- | --- | --- | +| `indicator_type` | `ipv4` | `ipv4`, `ipv6`, `domain`, `url`, `md5`, `sha1`, `sha256` or `sha512`. Export one type per run to keep reference sets separate. | +| `format` | `csv` | `csv` produces a headerless single-column UTF-8 file, one value per line, with standard CSV quoting. `json` produces a JSON array of strings. | +| `include_non_ids` | `false` | Include non-IDS attributes when explicitly enabled. Deleted data remains excluded. | + +Boolean settings accept JSON booleans, integers 0/1 and strings `true`, `false`, +`1`, `0`. Empty settings use the documented defaults. + +For CSV, create or select a compatible reference set in **Admin → Reference Set +Management**, view its contents, select **Import**, choose the file and the intended +QRadar domain, and import it. There is no header to remove. Do not add provenance +columns to the import file. Select the correct domain; shared reference data has +broader visibility than domain-specific data. + +JSON is for tooling that consumes a value array. It is not a full request body for +the current `/reference_data_collections/set_entries` endpoint: that API expects +an object per entry, including collection and domain IDs. The companion action +constructs those objects. MISP's export extension/MIME metadata is static, so this +module advertises `.csv`/`text/csv` for both modes. Rename a JSON download to `.json` +when required; its contents are still JSON. + +## Supported indicators and target types + +| Selected type | MISP attributes | Normalization | Compatible QRadar entry type | +| --- | --- | --- | --- | +| `ipv4` | `ip-src`, `ip-dst`, IP component of `domain\|ip` | Validate and canonicalize IPv4. | `IP` | +| `ipv6` | Same IP attributes | Validate and canonicalize IPv6. Keep separate from IPv4 because QRadar's `IP` type stores IPv4. | `ALN` | +| `domain` | `domain`, `hostname`, domain component of `domain\|ip` | IDNA encode, lowercase, remove a trailing DNS dot, validate labels and the 253-character normalized name limit. | `ALN` or `ALNIC` | +| `url` | `url` | Require an absolute URL with a host, valid port and no whitespace; preserve case and the exact value. | `ALN` | +| `md5`, `sha1`, `sha256`, `sha512` | Corresponding hash types, `filename\|HASH`, and the MD5 component of `malware-sample` | Validate hexadecimal length and lowercase. Split filename composites at the last pipe. | `ALN` or `ALNIC` | + +`ALN` is case-sensitive text; `ALNIC` ignores case. URLs use `ALN` to preserve +case-sensitive paths. IPv6 in a text set requires rules/properties that compare +compatible canonical text; it does not acquire the semantics of QRadar's IPv4 +set type. Separate sets are recommended for each indicator type even when several +types share the same QRadar entry type. + +Only supported values of the selected type are exported. Other types, malformed +values and unsupported attributes have separate skipped counts. CIDRs, scoped +IPv6, email addresses, filenames alone and port-qualified composites are excluded. +Port constraints are not discarded to create broader indicators. `domain|ip` is +validated as a whole before either component is selected. Control characters in +indicator values are rejected. No binary sample data is exported. + +Standard (`{"Event": {...}}`), bare and legacy event envelopes are supported, +including lists of events. Only `to_ids=true` attributes are included by default. +Deleted events, objects and attributes are always excluded. Duplicate canonical +values are emitted once and sorted for deterministic files. Existing `file` and +`domain-ip` objects are supported; no additional object templates are needed. + +## API example and response + +POST to the misp-modules `/query` endpoint: + +```json +{ + "module": "qradar_reference_set_export", + "config": {"indicator_type": "domain", "format": "csv"}, + "data": [{ + "Event": { + "uuid": "f1ef0409-c80a-4012-b401-b7d1050f4ed4", + "Attribute": [{ + "uuid": "aa09819b-f2d8-4661-83ea-d4897cafebc8", + "type": "domain", + "value": "EXAMPLE.org.", + "to_ids": true + }] + } + }] +} +``` + +The response has `response: []` and base64-encoded file contents in `data`. Here, +decoding `data` yields `example.org` followed by a newline. Empty selections produce +an empty CSV file or `[]` in JSON mode. + +Additional API fields include `summary` with selected type, format, count and +skipped counts, and `sources`, mapping each exported value to its contributing +event/attribute/object references. Warnings describe skipped attributes. These +fields are not embedded in the import file. MISP's normal file download may not +display them; inspect the `/query` response when you need the audit details. +Malformed requests or configuration return `error` instead of a file. + +Apply MISP sharing and handling policies before export. + +The export has no network calls or action-specific 100-value limit. QRadar's own +value-length and import limits remain applicable. Tests validate serialization and +mocked service endpoints; no live QRadar import was performed. + +References: [IBM IOC import instructions](https://www.ibm.com/docs/en/qradar-on-cloud?topic=overview-importing-iocs-reference-set) +and [QRadar reference-set types](https://www.ibm.com/docs/en/qradar-on-cloud?topic=overview-adding-editing-deleting-reference-sets). diff --git a/documentation/mkdocs/action_mod.md b/documentation/mkdocs/action_mod.md index 1c5762327..1c41ee3a1 100644 --- a/documentation/mkdocs/action_mod.md +++ b/documentation/mkdocs/action_mod.md @@ -53,6 +53,33 @@ Simplistic module to send a message to a Nextcloud talk conversation. ----- +#### [QRadar Reference Set Publish](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/action_mod/qradar_reference_set_publish.py) + +Publish typed MISP indicators to an existing QRadar reference set and explicit domain. +[[source code](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/action_mod/qradar_reference_set_publish.py)] + +- **features**: +>Adds or refreshes IDS-marked event and object indicators through reference_data_collections/set_entries. Requires an existing reference-set ID and explicit domain ID, validates the target entry type and checks each write acknowledgement. Supports credential-free dry runs and reports partial failures. Set expiry remains controlled by QRadar. Credentials come from the module server environment. See the [usage guide](https://github.com/MISP/misp-modules/blob/main/documentation/action_mod/qradar_reference_set_publish.md). + +- **config**: +>{'params': {'reference_set_id': {'type': 'string', 'description': 'Required: existing QRadar reference-set numeric ID'}, 'domain_id': {'type': 'string', 'description': 'Required: numeric domain ID (0 is Default Domain), or SHARED'}, 'indicator_type': {'type': 'string', 'description': 'ipv4, ipv6, domain, url, md5, sha1, sha256 or sha512', 'value': 'ipv4'}, 'source': {'type': 'string', 'description': 'QRadar entry source label', 'value': 'misp-modules'}, 'dry_run': {'type': 'string', 'description': 'true previews entry payloads without network access', 'value': 'false'}}, 'blocking': True, 'support_filters': False, 'expect_misp_core_format': True} + +- **input**: +>Selected MISP events and object attributes in workflow data. + +- **output**: +>Counts and per-value created, updated, planned or failed outcomes with MISP provenance. + +- **references**: +> - https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-set_entries-POST.html +> - https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-sets-id-GET.html + +- **requirements**: +> - requests +> - QRadar API 20.0 and an authorized service token with reference-data access + +----- + #### [SentinelOne IOC Publish](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/action_mod/sentinelone_ioc_publish.py) Publish MISP event and object indicators to SentinelOne Threat Intelligence with explicit scope. diff --git a/documentation/mkdocs/export_mod.md b/documentation/mkdocs/export_mod.md index e449b617d..2681ccb99 100644 --- a/documentation/mkdocs/export_mod.md +++ b/documentation/mkdocs/export_mod.md @@ -274,6 +274,30 @@ Simple export of a MISP event to PDF. ----- +#### [QRadar Reference Set Export](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/export_mod/qradar_reference_set_export.py) + +Export deduplicated, typed MISP indicators as QRadar reference-set CSV or JSON values. +[[source code](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/export_mod/qradar_reference_set_export.py)] + +- **features**: +>Exports one selected indicator type from supplied event and object attributes, with validation, normalization and deduplication. Produces a headerless single-column CSV for Reference Set Management, or a JSON value array. Defaults to IDS-marked IPv4 addresses; IPv6 uses a separate text set. No QRadar connection is required. See the [usage guide](https://github.com/MISP/misp-modules/blob/main/documentation/export_mod/qradar_reference_set_export.md). + +- **config**: +> - indicator_type +> - format +> - include_non_ids + +- **input**: +>MISP events, including object attributes, in standard, bare or legacy format. + +- **output**: +>Base64-encoded UTF-8 CSV or JSON; counts and provenance are returned separately from the file. + +- **references**: +>https://www.ibm.com/docs/en/qradar-on-cloud?topic=overview-importing-iocs-reference-set + +----- + #### [Splunk Lookup Export](https://github.com/MISP/misp-modules/tree/main/misp_modules/modules/export_mod/splunk_export.py) Export event and object indicators as a Splunk CSV lookup with CIM field mapping hints. diff --git a/documentation/mkdocs/index.md b/documentation/mkdocs/index.md index 5838fd7df..6523d8fab 100644 --- a/documentation/mkdocs/index.md +++ b/documentation/mkdocs/index.md @@ -156,6 +156,7 @@ For more information: [Extending MISP with Python modules](https://www.misp-proj * [Nexthink NXQL Export](https://misp.github.io/misp-modules/export_mod/#nexthink-nxql-export) - Nexthink NXQL query export module * [OSQuery Export](https://misp.github.io/misp-modules/export_mod/#osquery-export) - OSQuery export of a MISP event. * [Event to PDF Export](https://misp.github.io/misp-modules/export_mod/#event-to-pdf-export) - Simple export of a MISP event to PDF. +* [QRadar Reference Set Export](https://misp.github.io/misp-modules/export_mod/#qradar-reference-set-export) - Export deduplicated, typed MISP indicators as QRadar reference-set CSV or JSON values. * [Splunk Lookup Export](https://misp.github.io/misp-modules/export_mod/#splunk-lookup-export) - Export event and object indicators as a Splunk CSV lookup with CIM field mapping hints. * [Test Export](https://misp.github.io/misp-modules/export_mod/#test-export) - Skeleton export module. * [ThreatStream Export](https://misp.github.io/misp-modules/export_mod/#threatstream-export) - Module to export a structured CSV file for uploading to threatStream. @@ -188,6 +189,7 @@ For more information: [Extending MISP with Python modules](https://www.misp-proj * [CrowdStrike IOC Publish](https://misp.github.io/misp-modules/action_mod/#crowdstrike-ioc-publish) - Create and renew module-owned Falcon custom IOCs from MISP event and object attributes. * [Mattermost](https://misp.github.io/misp-modules/action_mod/#mattermost) - Simplistic module to send message to a Mattermost channel. * [Nextcloud talk](https://misp.github.io/misp-modules/action_mod/#nextcloud-talk) - Simplistic module to send a message to a Nextcloud talk conversation. +* [QRadar Reference Set Publish](https://misp.github.io/misp-modules/action_mod/#qradar-reference-set-publish) - Publish typed MISP indicators to an existing QRadar reference set and explicit domain. * [SentinelOne IOC Publish](https://misp.github.io/misp-modules/action_mod/#sentinelone-ioc-publish) - Publish MISP event and object indicators to SentinelOne Threat Intelligence with explicit scope. * [Slack](https://misp.github.io/misp-modules/action_mod/#slack) - Simplistic module to send messages to a Slack channel. * [Test action](https://misp.github.io/misp-modules/action_mod/#test-action) - This module is merely a test, always returning true. Triggers on event publishing. diff --git a/misp_modules/lib/qradar_reference_set.py b/misp_modules/lib/qradar_reference_set.py new file mode 100644 index 000000000..77d1140d4 --- /dev/null +++ b/misp_modules/lib/qradar_reference_set.py @@ -0,0 +1,136 @@ +"""Shared, network-free conversion of MISP attributes to typed QRadar set values.""" + +import ipaddress +import re +from collections import Counter +from urllib.parse import urlsplit + +HASH_LENGTHS = {"md5": 32, "sha1": 40, "sha256": 64, "sha512": 128} +ENTRY_TYPES = { + "ipv4": {"IP"}, + "ipv6": {"ALN"}, + "domain": {"ALN", "ALNIC"}, + "url": {"ALN"}, + **{kind: {"ALN", "ALNIC"} for kind in HASH_LENGTHS}, +} +SUPPORTED_TYPES = ( + {"ip-src", "ip-dst", "domain", "hostname", "url", "domain|ip", "malware-sample"} + | HASH_LENGTHS.keys() + | {f"filename|{kind}" for kind in HASH_LENGTHS} +) + + +def boolean(value, name): + if value is None or value == "": + return False + if isinstance(value, (bool, int)) and value in (0, 1): + return bool(value) + if isinstance(value, str) and value.lower() in ("true", "false", "1", "0"): + return value.lower() in ("true", "1") + raise ValueError(f"{name} must be true, false, 1 or 0.") + + +def indicator_type(value): + if value is None or value == "": + return "ipv4" + if not isinstance(value, str) or value not in ENTRY_TYPES: + raise ValueError("indicator_type must be ipv4, ipv6, domain, url, md5, sha1, sha256 or sha512.") + return value + + +def _records(container, key): + records = container.get(key, []) + if not isinstance(records, list) or any(not isinstance(item, dict) for item in records): + raise ValueError(f"{key} must be a list of objects.") + return records + + +def _atomic(kind, value): + if any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("Control characters are not supported.") + if kind in HASH_LENGTHS: + if not re.fullmatch(rf"[0-9a-fA-F]{{{HASH_LENGTHS[kind]}}}", value): + raise ValueError("Invalid hash.") + return kind, value.lower() + if kind in ("ip-src", "ip-dst"): + if "%" in value: + raise ValueError("Scoped IPv6 is unsupported.") + address = ipaddress.ip_address(value) + return f"ipv{address.version}", str(address) + if kind == "url": + parsed = urlsplit(value) + if not parsed.scheme or not parsed.hostname or any(char.isspace() for char in value): + raise ValueError("URL must be absolute with a host and no whitespace.") + parsed.port # Reject invalid ports while preserving the exact URL value. + return "url", value + domain = value.removesuffix(".").encode("idna").decode("ascii").lower() + if len(domain) > 253 or any( + not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label) for label in domain.split(".") + ): + raise ValueError("Invalid DNS name.") + return "domain", domain + + +def _indicators(kind, value): + if "|" not in kind and kind != "malware-sample": + return [_atomic(kind, value)] + first, separator, second = value.rpartition("|") + if not separator or not first.strip() or not second.strip(): + raise ValueError("Invalid composite.") + if kind == "domain|ip": + return [_atomic("domain", first), _atomic("ip-dst", second)] + return [_atomic("md5" if kind == "malware-sample" else kind.split("|")[1], second)] + + +def collect(data, selected_type, include_non_ids=False, limit=None): + """Return unique values with their provenance, plus skipped-attribute counts.""" + if isinstance(data, dict): + data = [data] + if not isinstance(data, list) or any(not isinstance(envelope, dict) for envelope in data): + raise ValueError("data must contain an event or list of events.") + values, skipped = {}, Counter() + for envelope in data: + event = envelope.get("Event", envelope) + if not isinstance(event, dict): + raise ValueError("Event must be an object.") + if boolean(event.get("deleted"), "deleted"): + continue + collections = dict(event, **envelope) + attributes = [(attribute, {}) for attribute in _records(collections, "Attribute")] + for obj in _records(collections, "Object"): + if not boolean(obj.get("deleted"), "deleted"): + attributes.extend((attribute, obj) for attribute in _records(obj, "Attribute")) + for attribute, obj in attributes: + if boolean(attribute.get("deleted"), "deleted") or ( + not boolean(attribute.get("to_ids"), "to_ids") and not include_non_ids + ): + skipped["deleted or non-IDS"] += 1 + continue + kind, value = attribute.get("type"), attribute.get("value") + if not isinstance(kind, str) or kind not in SUPPORTED_TYPES: + skipped["unsupported"] += 1 + continue + try: + if not isinstance(value, str) or not value.strip(): + raise ValueError("Empty value.") + atoms = [normalized for atom_type, normalized in _indicators(kind, value) if atom_type == selected_type] + except ValueError: + skipped["malformed"] += 1 + continue + if not atoms: + skipped["other indicator type"] += 1 + continue + ref = { + "event": str(event.get("uuid") or event.get("id") or "unknown"), + "attribute": str(attribute.get("uuid") or attribute.get("id") or "unknown"), + "misp_type": kind, + } + if obj: + ref["object"] = str(obj.get("uuid") or obj.get("id") or obj.get("name") or "unknown") + for atom in atoms: + refs = values.setdefault(atom, []) + if ref not in refs: + refs.append(ref) + if limit is not None and len(values) > limit: + raise ValueError(f"At most {limit} unique values are allowed; split the input.") + return values, dict(skipped) diff --git a/misp_modules/modules/action_mod/qradar_reference_set_publish.py b/misp_modules/modules/action_mod/qradar_reference_set_publish.py new file mode 100644 index 000000000..addd6d427 --- /dev/null +++ b/misp_modules/modules/action_mod/qradar_reference_set_publish.py @@ -0,0 +1,255 @@ +"""Add MISP indicators to an explicitly selected QRadar reference set and domain.""" + +import json +import os +import re +import time +from urllib.parse import urlsplit + +import requests + +from misp_modules.lib.qradar_reference_set import ENTRY_TYPES, boolean, collect, indicator_type + +moduleinfo = { + "version": "1.0", + "author": "MISP Project", + "name": "QRadar Reference Set Publish", + "description": "Publish typed MISP indicators to an existing QRadar reference set and explicit domain.", + "module-type": ["action"], + "logo": "", + "requirements": ["requests", "QRadar API 20.0 and an authorized service token with reference-data access"], + "features": ( + "Adds or refreshes IDS-marked event and object indicators through reference_data_collections/set_entries." + " Requires an existing reference-set ID and explicit domain ID, validates the target entry type and checks" + " each write acknowledgement. Supports credential-free dry runs and reports partial failures. Set expiry" + " remains controlled by QRadar. Credentials come from the module server environment. See the [usage" + " guide](https://github.com/MISP/misp-modules/blob/main/documentation/action_mod/qradar_reference_set_publish.md)." + ), + "references": [ + "https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-set_entries-POST.html", + "https://ibmsecuritydocs.github.io/qradar_api_20.0/20.0--reference_data_collections-sets-id-GET.html", + ], + "input": "Selected MISP events and object attributes in workflow data.", + "output": "Counts and per-value created, updated, planned or failed outcomes with MISP provenance.", +} +moduleconfig = { + "params": { + "reference_set_id": {"type": "string", "description": "Required: existing QRadar reference-set numeric ID"}, + "domain_id": { + "type": "string", + "description": "Required: numeric domain ID (0 is Default Domain), or SHARED", + }, + "indicator_type": { + "type": "string", + "description": "ipv4, ipv6, domain, url, md5, sha1, sha256 or sha512", + "value": "ipv4", + }, + "source": {"type": "string", "description": "QRadar entry source label", "value": "misp-modules"}, + "dry_run": { + "type": "string", + "description": "true previews entry payloads without network access", + "value": "false", + }, + }, + "blocking": True, + "support_filters": False, + "expect_misp_core_format": True, +} +returns = "data" +API_PATH = "/api/reference_data_collections" +API_VERSION = "20.0" +MAX_VALUES = 100 +DEADLINE_SECONDS = 240 + + +class PublishError(Exception): + """A sanitized error suitable for a workflow result.""" + + +def _identifier(value, name, minimum): + if isinstance(value, bool) or not isinstance(value, (int, str)) or not re.fullmatch(r"[0-9]{1,19}", str(value)): + raise ValueError(f"{name} must be an explicit integer ID.") + number = int(value) + if not minimum <= number <= 9223372036854775807: + raise ValueError(f"{name} is outside its supported range.") + return number + + +def _settings(params): + if not isinstance(params, dict): + raise ValueError("params must be an object.") + source = params.get("source") + if source is None or source == "": + source = "misp-modules" + if not isinstance(source, str) or not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9 ._:/-]{0,63}", source): + raise ValueError("source must be 1-64 ASCII letters, digits, spaces or ._:/- and start with a letter or digit.") + return { + "collection_id": _identifier(params.get("reference_set_id"), "reference_set_id", 1), + "domain_id": ( + None if params.get("domain_id") == "SHARED" else _identifier(params.get("domain_id"), "domain_id", 0) + ), + "indicator_type": indicator_type(params.get("indicator_type")), + "source": source, + "dry_run": boolean(params.get("dry_run"), "dry_run"), + } + + +def _connection(): + origin = os.getenv("QRADAR_URL", "") + parsed = urlsplit(origin) + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or parsed.path not in ("", "/") + or parsed.query + or parsed.fragment + or any(ord(char) <= 32 or ord(char) == 127 for char in origin) + ): + raise ValueError("Invalid QRADAR_URL.") + parsed.port + token = os.getenv("QRADAR_API_TOKEN", "") + if not token or any(not 33 <= ord(char) <= 126 for char in token): + raise ValueError("Invalid QRADAR_API_TOKEN.") + ca_bundle = os.getenv("QRADAR_CA_BUNDLE") or True + return origin.rstrip("/") + API_PATH, token, ca_bundle + + +def _request(session, method, endpoint, verify, **kwargs): + try: + response = session.request(method, endpoint, timeout=15, verify=verify, allow_redirects=False, **kwargs) + except (requests.RequestException, OSError): + raise PublishError(f"{method} failed; remote outcome may be unknown. Check QRadar before retrying.") from None + expected = (200,) if method == "GET" else (200, 201) + if response.status_code not in expected: + raise PublishError(f"{method}: QRadar HTTP {response.status_code}; remote outcome may be unknown.") + try: + body = response.json() + except ValueError: + raise PublishError(f"{method}: invalid JSON acknowledgement; remote outcome may be unknown.") from None + if not isinstance(body, dict) or body.get("errors") or body.get("warnings") or body.get("http_response"): + raise PublishError(f"{method}: invalid response or API error; remote outcome may be unknown.") + return response.status_code, body + + +def _check_target(session, endpoint, verify, settings): + _, target = _request(session, "GET", f"{endpoint}/sets/{settings['collection_id']}", verify) + if type(target.get("id")) is not int or target["id"] != settings["collection_id"]: + raise PublishError("Reference-set identity could not be confirmed; no values submitted.") + if ( + not isinstance(target.get("entry_type"), str) + or target["entry_type"] not in ENTRY_TYPES[settings["indicator_type"]] + ): + allowed = ", ".join(sorted(ENTRY_TYPES[settings["indicator_type"]])) + raise PublishError(f"Reference-set entry_type must be {allowed} for this indicator type; no values submitted.") + return target + + +def _publish(session, endpoint, verify, payload): + status, record = _request(session, "POST", f"{endpoint}/set_entries", verify, json=payload) + if ( + type(record.get("id")) is not int + or record["id"] <= 0 + or type(record.get("collection_id")) is not int + or (payload["domain_id"] is not None and type(record.get("domain_id")) is not int) + or any(key not in record or record[key] != value for key, value in payload.items()) + or record.get("delete_entry") not in (None, False) + ): + raise PublishError("POST: no matching set/domain/value acknowledgement; remote outcome may be unknown.") + return "created" if status == 201 else "updated", record["id"] + + +def handler(q=False): + if not q: + return False + try: + request = json.loads(q) + if not isinstance(request, dict): + raise ValueError("Request must be an object.") + if "filteredItems" in request or "matchingData" in request: + raise ValueError("Workflow filters/extracted data are unsupported; supply selected events in data.") + settings = _settings(request.get("params", {})) + values, skipped = collect(request.get("data"), settings["indicator_type"], limit=MAX_VALUES) + except (ValueError, TypeError) as exc: + return {"error": f"QRadar reference set publish: {exc}"} + summary = { + "success": True, + **settings, + "created": 0, + "updated": 0, + "planned": 0, + "failed": 0, + "skipped": skipped, + "items": [], + } + result = {"result_key": "qradar_reference_set_publish", "data": summary} + if not values: + return result + deadline = time.monotonic() + DEADLINE_SECONDS + session = None + try: + if not settings["dry_run"]: + try: + endpoint, token, verify = _connection() + except ValueError: + return { + "error": ( + "Configure QRADAR_URL as an HTTPS console origin and QRADAR_API_TOKEN on the module server." + ) + } + session = requests.Session() + session.headers.update({"SEC": token, "Version": API_VERSION, "Accept": "application/json"}) + try: + target = _check_target(session, endpoint, verify, settings) + except PublishError as exc: + summary["success"] = False + result["error"] = f"QRadar reference set publish: {exc}" + return result + summary["reference_set_name"] = target.get("name") + for value, refs in values.items(): + payload = { + "collection_id": settings["collection_id"], + "domain_id": settings["domain_id"], + "source": settings["source"], + "value": value, + } + outcome = {"value": value, "sources": refs} + if settings["dry_run"]: + outcome.update(status="planned", payload=payload) + summary["planned"] += 1 + else: + try: + if time.monotonic() >= deadline: + raise PublishError("Publication time budget exceeded; value not attempted.") + status, identifier = _publish(session, endpoint, verify, payload) + outcome.update(status=status, id=identifier) + summary[status] += 1 + except PublishError as exc: + outcome.update(status="failed", error=str(exc)) + summary["failed"] += 1 + except Exception: + outcome.update( + status="failed", + error="QRadar request failed; remote outcome may be unknown. Check QRadar before retrying.", + ) + summary["failed"] += 1 + summary["items"].append(outcome) + finally: + if session is not None: + session.close() + if summary["failed"]: + summary["success"] = False + result["error"] = ( + f"QRadar reference set publish: {summary['failed']} failed; {summary['created']} created and" + f" {summary['updated']} updated. Publication may be partial." + ) + return result + + +def introspection(): + return {"config": moduleconfig} + + +def version(): + return dict(moduleinfo, config=moduleconfig) diff --git a/misp_modules/modules/export_mod/qradar_reference_set_export.py b/misp_modules/modules/export_mod/qradar_reference_set_export.py new file mode 100644 index 000000000..68d7ba3d4 --- /dev/null +++ b/misp_modules/modules/export_mod/qradar_reference_set_export.py @@ -0,0 +1,87 @@ +"""Export one typed list of MISP indicators for a QRadar reference set.""" + +import base64 +import csv +import io +import json + +from misp_modules.lib.qradar_reference_set import SUPPORTED_TYPES, boolean, collect, indicator_type + +moduleinfo = { + "version": "1.0", + "author": "MISP Project", + "name": "QRadar Reference Set Export", + "description": "Export deduplicated, typed MISP indicators as QRadar reference-set CSV or JSON values.", + "module-type": ["export"], + "logo": "", + "requirements": [], + "features": ( + "Exports one selected indicator type from supplied event and object attributes, with validation, normalization" + " and deduplication. Produces a headerless single-column CSV for Reference Set Management, or a JSON value" + " array. Defaults to IDS-marked IPv4 addresses; IPv6 uses a separate text set. No QRadar connection is" + " required." + " See the [usage" + " guide](https://github.com/MISP/misp-modules/blob/main/documentation/export_mod/qradar_reference_set_export.md)." + ), + "references": ["https://www.ibm.com/docs/en/qradar-on-cloud?topic=overview-importing-iocs-reference-set"], + "input": "MISP events, including object attributes, in standard, bare or legacy format.", + "output": "Base64-encoded UTF-8 CSV or JSON; counts and provenance are returned separately from the file.", +} +moduleconfig = ["indicator_type", "format", "include_non_ids"] + + +def handler(q=False): + if not q: + return False + try: + request = json.loads(q) + if not isinstance(request, dict): + raise ValueError("Request must be an object.") + config = request.get("config") + if config is None: + config = {} + if not isinstance(config, dict): + raise ValueError("config must be an object.") + selected = indicator_type(config.get("indicator_type")) + output_format = config.get("format") + if output_format is None or output_format == "": + output_format = "csv" + if output_format not in ("csv", "json"): + raise ValueError("format must be csv or json.") + values, skipped = collect( + request.get("data"), selected, boolean(config.get("include_non_ids"), "include_non_ids") + ) + ordered = sorted(values) + if output_format == "json": + output = json.dumps(ordered, ensure_ascii=False) + "\n" + else: + buffer = io.StringIO(newline="") + writer = csv.writer(buffer, lineterminator="\n") + writer.writerows([value] for value in ordered) + output = buffer.getvalue() + except (ValueError, TypeError) as exc: + return {"error": f"QRadar reference set export: {exc}"} + result = { + "response": [], + "data": base64.b64encode(output.encode("utf-8")).decode("ascii"), + "summary": {"indicator_type": selected, "format": output_format, "count": len(values), "skipped": skipped}, + "sources": values, + } + if skipped: + result["warnings"] = [f"Skipped {count} {reason} attribute(s)." for reason, count in sorted(skipped.items())] + return result + + +def introspection(): + # MISP's download extension/MIME are static; JSON downloads also retain the .csv extension. + return { + "input": sorted(SUPPORTED_TYPES), + "inputSource": ["event", "event-collection"], + "outputFileExtension": "csv", + "responseType": "text/csv", + "userConfig": {}, + } + + +def version(): + return dict(moduleinfo, config=moduleconfig) diff --git a/tests/test_qradar_reference_set.py b/tests/test_qradar_reference_set.py new file mode 100644 index 000000000..16f943046 --- /dev/null +++ b/tests/test_qradar_reference_set.py @@ -0,0 +1,588 @@ +import base64 +import copy +import csv +import io +import json +from unittest.mock import MagicMock + +import pytest +import requests +from pymisp import MISPObject +from tornado.testing import AsyncHTTPTestCase +from tornado.web import Application + +from misp_modules import __main__ as service +from misp_modules.lib import qradar_reference_set as conversion +from misp_modules.modules.action_mod import qradar_reference_set_publish as publish +from misp_modules.modules.export_mod import qradar_reference_set_export as export + + +def attribute(kind="ip-dst", value="192.0.2.1", **metadata): + return {"type": kind, "value": value, "uuid": "attribute-1", "to_ids": True, **metadata} + + +def event(attributes=None, objects=None, **metadata): + return {"uuid": "event-1", "Attribute": attributes or [], "Object": objects or [], **metadata} + + +def export_values(data=None, **config): + result = export.handler(json.dumps({"data": event([attribute()]) if data is None else data, "config": config})) + assert "error" not in result, result + text = base64.b64decode(result["data"]).decode("utf-8") + values = json.loads(text) if config.get("format") == "json" else [row[0] for row in csv.reader(io.StringIO(text))] + return values, result + + +def run(data=None, **params): + return publish.handler( + json.dumps( + { + "data": event([attribute()]) if data is None else data, + "params": {"reference_set_id": "42", "domain_id": "7", **params}, + } + ) + ) + + +def response(body, status=200): + result = MagicMock(spec=requests.Response) + result.status_code = status + result.json.return_value = body + return result + + +@pytest.fixture +def client(monkeypatch): + session = MagicMock(spec=requests.Session) + session.headers = {} + + def request(method, url, **kwargs): + if method == "GET": + return response( + { + "id": 42, + "name": "MISP IPv4", + "entry_type": "IP", + "expiry_type": "LAST_SEEN", + "time_to_live": 86400000, + } + ) + return response({**kwargs["json"], "id": 901, "delete_entry": False}, 201) + + session.request.side_effect = request + session.constructor = MagicMock(return_value=session) + monkeypatch.setattr(publish.requests, "Session", session.constructor) + monkeypatch.setenv("QRADAR_URL", "https://qradar.example") + monkeypatch.setenv("QRADAR_API_TOKEN", "test-token") + monkeypatch.delenv("QRADAR_CA_BUNDLE", raising=False) + return session + + +@pytest.mark.parametrize( + "kind,value,selected,expected", + [ + ("ip-src", "192.0.2.1", "ipv4", "192.0.2.1"), + ("ip-dst", "2001:0DB8:0000::1", "ipv6", "2001:db8::1"), + ("domain", "EXAMPLE.org.", "domain", "example.org"), + ("hostname", "bücher.example", "domain", "xn--bcher-kva.example"), + ("url", 'https://example.org:8443/A?q="x",Y=1', "url", 'https://example.org:8443/A?q="x",Y=1'), + *[(kind, "A" * length, kind, "a" * length) for kind, length in conversion.HASH_LENGTHS.items()], + *[ + (f"filename|{kind}", "a|b.exe|" + "A" * length, kind, "a" * length) + for kind, length in conversion.HASH_LENGTHS.items() + ], + ("malware-sample", "a.exe|" + "A" * 32, "md5", "a" * 32), + ("domain|ip", "EXAMPLE.org|192.0.2.1", "domain", "example.org"), + ("domain|ip", "example.org|192.0.2.1", "ipv4", "192.0.2.1"), + ("domain|ip", "example.org|2001:db8::1", "ipv6", "2001:db8::1"), + ], +) +@pytest.mark.parametrize("output_format", ["csv", "json"]) +def test_export_normalizes_supported_types_and_serializes_plain_values(kind, value, selected, expected, output_format): + values, result = export_values(event([attribute(kind, value)]), indicator_type=selected, format=output_format) + assert values == [expected] + assert result["summary"]["count"] == 1 + assert result["sources"][expected][0]["misp_type"] == kind + + +def test_standard_legacy_bare_object_deduplication_and_nonmutation(): + obj = {"uuid": "object-1", "name": "domain-ip", "Attribute": [attribute(uuid="object-attribute")]} + standard = event([attribute()], [obj]) + legacy = {"Event": {"uuid": "event-2"}, "Attribute": [attribute(uuid="legacy-attribute")]} + data = [{"Event": standard}, standard, legacy] + original = copy.deepcopy(data) + values, result = export_values(data) + assert values == ["192.0.2.1"] + refs = result["sources"][values[0]] + assert len(refs) == 3 and refs[1]["object"] == "object-1" + assert data == original + + +@pytest.mark.parametrize( + "template,relation,value,selected", + [ + ("file", "sha256", "a" * 64, "sha256"), + ("domain-ip", "domain", "example.org", "domain"), + ], +) +def test_existing_object_templates_are_accepted(template, relation, value, selected): + obj = MISPObject(template, strict=True) + obj.add_attribute(relation, value=value, to_ids=True) + obj._validate() + values, result = export_values(event(objects=[json.loads(obj.to_json())]), indicator_type=selected) + assert values == [value] and result["sources"][value][0]["object"] == obj.uuid + + +def test_csv_has_no_header_and_is_deterministic(): + data = event([attribute(value="192.0.2.2"), attribute(), attribute()]) + values, result = export_values(data) + assert values == ["192.0.2.1", "192.0.2.2"] + assert base64.b64decode(result["data"]) == b"192.0.2.1\n192.0.2.2\n" + + +def test_empty_selection_exports_empty_csv_or_json_array(): + assert export_values([])[0] == [] + assert base64.b64decode(export_values([])[1]["data"]) == b"" + assert export_values([], format="json")[0] == [] + + +def test_only_selected_type_and_ids_are_included_and_deleted_data_is_never_included(): + data = [ + event( + [ + attribute(to_ids="false"), + attribute(deleted="1"), + attribute("domain", "example.org"), + attribute("ip-dst|port", "192.0.2.1|443"), + attribute("email-src", "test@example.org"), + ], + [{"deleted": True, "Attribute": [attribute()]}], + ), + event([attribute()], deleted=True), + ] + values, result = export_values(data) + assert values == [] + assert result["summary"]["skipped"] == {"deleted or non-IDS": 2, "other indicator type": 1, "unsupported": 2} + assert len(result["warnings"]) == 3 + assert export_values(data, include_non_ids="true")[0] == ["192.0.2.1"] + + +@pytest.mark.parametrize( + "kind,value", + [ + ("ip-dst", "192.0.2.0/24"), + ("ip-dst", "999.2.3.4"), + ("ip-src", "fe80::1%eth0"), + ("domain", "bad'name.example"), + ("domain", "a..example"), + ("domain", "a" * 64 + ".example"), + ("domain", ".".join(["a" * 63] * 4)), + ("domain", "example.org.."), + ("domain", "bad\ud800.example"), + ("domain", "example.org\n"), + ("url", "https://example.org:bad/"), + ("url", "https://example.org:65536/"), + ("url", "example.org/path"), + ("url", "https://example.org/a b"), + ("url", "https://example.org/a\x00b"), + ("md5", "bad"), + ("sha256", "g" * 64), + ("filename|md5", "a.exe|bad"), + ("domain|ip", "example.org|not-ip"), + ("domain|ip", "bad..example|192.0.2.1"), + ("domain", None), + ("ip-dst", ""), + ], +) +def test_invalid_supported_values_are_skipped_before_network(client, kind, value): + result = run(event([attribute(kind, value)])) + assert result["data"]["skipped"] == {"malformed": 1} + client.constructor.assert_not_called() + + +@pytest.mark.parametrize( + "value,expected", + [ + (True, True), + (1, True), + ("TRUE", True), + ("1", True), + (False, False), + ("0", False), + ("false", False), + (None, False), + ("", False), + ], +) +def test_misp_boolean_encodings(value, expected): + values, _ = export_values(event([attribute(to_ids=value)])) + assert bool(values) is expected + + +@pytest.mark.parametrize( + "config", + [ + [], + {"indicator_type": "all"}, + {"indicator_type": []}, + {"format": "xml"}, + {"format": False}, + {"include_non_ids": "maybe"}, + {"include_non_ids": 2}, + ], +) +def test_export_rejects_bad_configuration(config): + assert "error" in export.handler(json.dumps({"data": [], "config": config})) + + +@pytest.mark.parametrize( + "data", + [None, "bad", [1], {"Event": "bad"}, event(attributes=[1]), {"Object": {}}, event([attribute(to_ids="maybe")])], +) +def test_both_modules_reject_malformed_input(data): + assert "error" in export.handler(json.dumps({"data": data})) + assert "error" in publish.handler(json.dumps({"data": data, "params": {"reference_set_id": 42, "domain_id": 7}})) + + +@pytest.mark.parametrize("module", [export, publish]) +def test_request_contract(module): + assert module.handler() is False + for payload in ["invalid JSON", "[]", "null"]: + assert "error" in module.handler(payload) + + +def test_publish_checks_set_and_sends_current_api_contract(client): + result = run() + assert result["data"]["created"] == 1 and result["data"]["success"] is True + assert result["result_key"] == "qradar_reference_set_publish" + lookup, create = client.request.call_args_list + assert lookup.args == ("GET", "https://qradar.example/api/reference_data_collections/sets/42") + assert create.args == ("POST", "https://qradar.example/api/reference_data_collections/set_entries") + assert create.kwargs["json"] == { + "collection_id": 42, + "domain_id": 7, + "source": "misp-modules", + "value": "192.0.2.1", + } + assert client.headers == {"SEC": "test-token", "Version": "20.0", "Accept": "application/json"} + assert ( + create.kwargs["verify"] is True and create.kwargs["allow_redirects"] is False and create.kwargs["timeout"] == 15 + ) + assert not any( + field in create.kwargs["json"] for field in ["id", "delete_entry", "first_seen", "last_seen", "notes"] + ) + client.close.assert_called_once() + + +def test_existing_values_are_reported_as_updated_by_http_status(client): + client.request.side_effect = [ + response({"id": 42, "entry_type": "IP"}), + response({"id": 901, "collection_id": 42, "domain_id": 7, "source": "misp-modules", "value": "192.0.2.1"}), + ] + result = run() + assert result["data"]["updated"] == 1 and result["data"]["created"] == 0 + + +@pytest.mark.parametrize( + "selected,entry_type,value", + [ + ("domain", "ALN", "example.org"), + ("domain", "ALNIC", "example.org"), + ("url", "ALN", "https://example.org/CaseSensitive"), + ("ipv6", "ALN", "2001:db8::1"), + *[ + (kind, entry_type, "a" * length) + for kind, length in conversion.HASH_LENGTHS.items() + for entry_type in ["ALN", "ALNIC"] + ], + ], +) +def test_compatible_set_types_publish(client, selected, entry_type, value): + kind = "ip-src" if selected == "ipv6" else selected + payload = {"collection_id": 42, "domain_id": 7, "source": "misp-modules", "value": value} + client.request.side_effect = [response({"id": 42, "entry_type": entry_type}), response({**payload, "id": 901}, 201)] + assert run(event([attribute(kind, value)]), indicator_type=selected)["data"]["created"] == 1 + + +@pytest.mark.parametrize( + "target,selected,kind,value", + [ + ({"id": 43, "entry_type": "IP"}, "ipv4", "ip-dst", "192.0.2.1"), + ({"id": True, "entry_type": "IP"}, "ipv4", "ip-dst", "192.0.2.1"), + ({"id": "42", "entry_type": "IP"}, "ipv4", "ip-dst", "192.0.2.1"), + ({"id": 42, "entry_type": "ALN"}, "ipv4", "ip-dst", "192.0.2.1"), + ({"id": 42, "entry_type": []}, "ipv4", "ip-dst", "192.0.2.1"), + ({"id": 42, "entry_type": "ALNIC"}, "url", "url", "https://example.org/Case"), + ({"id": 42, "entry_type": "IP"}, "ipv6", "ip-src", "2001:db8::1"), + ], +) +def test_wrong_identity_or_type_prevents_all_writes(client, target, selected, kind, value): + client.request.side_effect = None + client.request.return_value = response(target) + result = run(event([attribute(kind, value)]), indicator_type=selected) + assert "error" in result and result["data"]["success"] is False + assert client.request.call_count == 1 + client.close.assert_called_once() + + +def test_dry_run_needs_no_credentials_and_matches_export_values(client, monkeypatch): + monkeypatch.delenv("QRADAR_URL") + monkeypatch.delenv("QRADAR_API_TOKEN") + data = event([attribute(), attribute(uuid="other-source")]) + result = run(data, dry_run="true", domain_id="0") + assert result["data"]["planned"] == 1 + item = result["data"]["items"][0] + assert item["payload"]["domain_id"] == 0 + assert item["payload"]["value"] == export_values(data)[0][0] + assert len(item["sources"]) == 2 + client.constructor.assert_not_called() + + +def test_shared_domain_is_explicit_and_serialized_as_null(client): + result = run(domain_id="SHARED") + assert result["data"]["created"] == 1 + assert client.request.call_args.kwargs["json"]["domain_id"] is None + + +def test_shared_acknowledgement_requires_an_explicit_null_domain_field(client): + record = {"id": 901, "collection_id": 42, "source": "misp-modules", "value": "192.0.2.1"} + client.request.side_effect = [response({"id": 42, "entry_type": "IP"}), response(record, 201)] + assert run(domain_id="SHARED")["data"]["failed"] == 1 + + +@pytest.mark.parametrize( + "params", + [ + {"reference_set_id": None}, + {"reference_set_id": 0}, + {"reference_set_id": "42/evil"}, + {"reference_set_id": True}, + {"reference_set_id": 1.5}, + {"reference_set_id": "9" * 20}, + {"domain_id": None}, + {"domain_id": -1}, + {"domain_id": False}, + {"domain_id": "7,8"}, + {"indicator_type": "all"}, + {"source": "bad\nlabel"}, + {"source": False}, + {"dry_run": "maybe"}, + ], +) +def test_invalid_action_parameters_fail_before_network(client, params): + assert "error" in run(**params) + client.constructor.assert_not_called() + + +@pytest.mark.parametrize("field", ["filteredItems", "matchingData"]) +def test_action_rejects_unapplied_workflow_filters(client, field): + result = publish.handler( + json.dumps({"params": {"reference_set_id": 42, "domain_id": 7}, "data": event([attribute()]), field: []}) + ) + assert "error" in result + client.constructor.assert_not_called() + + +@pytest.mark.parametrize( + "origin", + [ + "http://qradar.example", + "https://user:password@qradar.example", + "https://@qradar.example", + "https://:password@qradar.example", + "https://qradar.example/api", + "https://qradar.example/?q=x", + "https://qradar.example/#x", + "https://qradar.example:bad", + "https://qradar.example\n", + "", + ], +) +def test_invalid_console_origins_cannot_receive_tokens(client, monkeypatch, origin): + monkeypatch.setenv("QRADAR_URL", origin) + result = run() + assert "error" in result and "password" not in json.dumps(result) + client.constructor.assert_not_called() + + +@pytest.mark.parametrize("token", ["", "secret\r\nHeader: bad", "secret token", "tökén"]) +def test_invalid_token_is_not_exposed(client, monkeypatch, token): + monkeypatch.setenv("QRADAR_API_TOKEN", token) + assert "error" in run() + client.constructor.assert_not_called() + + +def test_server_ca_bundle_is_used_without_disabling_verification(client, monkeypatch): + monkeypatch.setenv("QRADAR_CA_BUNDLE", "/etc/certs/qradar-ca.pem") + assert run()["data"]["created"] == 1 + assert all(call.kwargs["verify"] == "/etc/certs/qradar-ca.pem" for call in client.request.call_args_list) + + +def test_unreadable_ca_bundle_fails_preflight_without_exposing_exception_details(client): + client.request.side_effect = OSError("test-token") + result = run() + assert "error" in result and "test-token" not in json.dumps(result) + assert client.request.call_count == 1 + client.close.assert_called_once() + + +def test_action_limit_is_checked_before_any_network(client): + data = event([attribute(value=f"192.0.2.{index}") for index in range(101)]) + assert "error" in run(data) + client.constructor.assert_not_called() + assert export_values(data)[1]["summary"]["count"] == 101 + + +@pytest.mark.parametrize("status", [202, 204, 301, 302, 400, 401, 403, 404, 422, 429, 500]) +def test_bad_lookup_http_status_does_not_write_or_expose_response(client, status): + client.request.side_effect = None + client.request.return_value = response({"message": "test-token"}, status) + result = run() + assert "error" in result and "test-token" not in json.dumps(result) + assert client.request.call_count == 1 + + +@pytest.mark.parametrize( + "body", + [ + [], + None, + {}, + {"id": 42, "entry_type": "IP", "errors": [{}]}, + {"http_response": {"code": 500}}, + {"id": 42, "entry_type": "IP", "warnings": [{}]}, + ], +) +def test_malformed_lookup_and_api_errors_prevent_writes(client, body): + client.request.side_effect = None + client.request.return_value = response(body) + assert "error" in run() + assert client.request.call_count == 1 + + +@pytest.mark.parametrize( + "change", + [ + {"id": None}, + {"id": True}, + {"collection_id": 43}, + {"collection_id": True}, + {"domain_id": 8}, + {"domain_id": True}, + {"value": "192.0.2.2"}, + {"source": "foreign"}, + {"delete_entry": True}, + ], +) +def test_write_acknowledgement_must_confirm_identity_domain_value_and_source(client, change): + record = {"id": 901, "collection_id": 42, "domain_id": 7, "value": "192.0.2.1", "source": "misp-modules", **change} + client.request.side_effect = [response({"id": 42, "entry_type": "IP"}), response(record, 201)] + result = run() + assert result["data"]["failed"] == 1 and result["data"]["created"] == 0 + assert "unknown" in result["data"]["items"][0]["error"] + assert client.request.call_count == 2 + + +@pytest.mark.parametrize( + "failure", + [requests.Timeout("test-token"), requests.ConnectionError("test-token"), response({"message": "test-token"}, 429)], +) +def test_partial_publication_preserves_success_and_reports_uncertain_outcomes_without_retry(client, failure): + good = {"id": 901, "collection_id": 42, "domain_id": 7, "value": "192.0.2.1", "source": "misp-modules"} + client.request.side_effect = [response({"id": 42, "entry_type": "IP"}), response(good, 201), failure] + result = run(event([attribute(), attribute(value="192.0.2.2")])) + assert result["data"]["created"] == 1 and result["data"]["failed"] == 1 + assert result["data"]["success"] is False and "partial" in result["error"] + assert "test-token" not in json.dumps(result) and client.request.call_count == 3 + client.close.assert_called_once() + + +def test_invalid_json_response_is_sanitized(client): + bad = response({}) + bad.json.side_effect = ValueError("test-token") + client.request.side_effect = [response({"id": 42, "entry_type": "IP"}), bad] + result = run() + assert result["data"]["failed"] == 1 and "test-token" not in json.dumps(result) + + +def test_processing_budget_is_checked_after_target_lookup(client, monkeypatch): + clock = iter([0, publish.DEADLINE_SECONDS + 1]) + monkeypatch.setattr(publish.time, "monotonic", lambda: next(clock)) + result = run() + assert result["data"]["failed"] == 1 and client.request.call_count == 1 + + +def test_requests_serializes_actual_wire_contract(monkeypatch): + calls = [] + + def send(self, request, **kwargs): + calls.append((request, kwargs)) + if request.method == "GET": + assert request.path_url == "/api/reference_data_collections/sets/42" + body, status = {"id": 42, "entry_type": "IP"}, 200 + else: + assert request.path_url == "/api/reference_data_collections/set_entries" + payload = json.loads(request.body) + assert payload == {"collection_id": 42, "domain_id": 0, "source": "misp-modules", "value": "192.0.2.1"} + assert request.headers["Content-Type"] == "application/json" + body, status = {**payload, "id": 901}, 201 + result = requests.Response() + result.status_code, result._content = status, json.dumps(body).encode() + result.request, result.url = request, request.url + return result + + monkeypatch.setenv("QRADAR_URL", "https://qradar.example/") + monkeypatch.setenv("QRADAR_API_TOKEN", "test-token") + monkeypatch.delenv("QRADAR_CA_BUNDLE", raising=False) + monkeypatch.setattr(requests.adapters.HTTPAdapter, "send", send) + assert run(domain_id=0)["data"]["created"] == 1 + assert [request.method for request, _ in calls] == ["GET", "POST"] + assert all(request.headers["SEC"] == "test-token" and request.headers["Version"] == "20.0" for request, _ in calls) + + +class TestQRadarHTTP(AsyncHTTPTestCase): + def get_app(self): + self.previous, self.cache = service.MODULES_HANDLERS.copy(), service.ListModules.CACHE + service.MODULES_HANDLERS.clear() + service.MODULES_HANDLERS.update( + { + "qradar_reference_set_export": export, + "type:qradar_reference_set_export": "export_mod", + "qradar_reference_set_publish": publish, + "type:qradar_reference_set_publish": "action_mod", + } + ) + service.ListModules.CACHE = None + return Application([(r"/modules", service.ListModules), (r"/query", service.QueryModule)]) + + def tearDown(self): + service.MODULES_HANDLERS.clear() + service.MODULES_HANDLERS.update(self.previous) + service.ListModules.CACHE = self.cache + super().tearDown() + + def test_discovery_export_and_action_have_distinct_names_and_contracts(self): + descriptors = {descriptor["name"]: descriptor for descriptor in json.loads(self.fetch("/modules").body)} + assert set(descriptors) == {"qradar_reference_set_export", "qradar_reference_set_publish"} + assert descriptors["qradar_reference_set_export"]["type"] == "export_mod" + assert descriptors["qradar_reference_set_publish"]["meta"]["config"]["blocking"] is True + for name in descriptors: + result = self.fetch( + "/query", + method="POST", + headers={"Content-Type": "application/json"}, + body=json.dumps( + { + "module": name, + "data": [event([attribute()])], + "params": {"reference_set_id": 42, "domain_id": 7, "dry_run": True}, + } + ), + ) + assert result.code == 200 + body = json.loads(result.body) + assert "error" not in body + if name.endswith("_export"): + assert base64.b64decode(body["data"]) == b"192.0.2.1\n" + else: + assert body["data"]["planned"] == 1