diff --git a/external-import/rst-threat-library/.dockerignore b/external-import/rst-threat-library/.dockerignore new file mode 100644 index 0000000000..c5d360e969 --- /dev/null +++ b/external-import/rst-threat-library/.dockerignore @@ -0,0 +1,9 @@ +__metadata__ +**/__pycache__ +**/__docs__ +**/.venv +**/venv +**/logs +**/config.yml +**/*.egg-info +**/*.gql diff --git a/external-import/rst-threat-library/Dockerfile b/external-import/rst-threat-library/Dockerfile new file mode 100644 index 0000000000..4e6bbedf32 --- /dev/null +++ b/external-import/rst-threat-library/Dockerfile @@ -0,0 +1,28 @@ +FROM python:3.12-alpine +ENV CONNECTOR_TYPE=EXTERNAL_IMPORT + +WORKDIR /opt/opencti-connector-rst-threat-library + +COPY src /opt/opencti-connector-rst-threat-library/src + +# Install pinned deps first, then connectors-sdk with --no-deps so its pycti +RUN apk --no-cache add --virtual .build-deps git build-base && \ + apk --no-cache add libmagic && \ + grep -v '^connectors-sdk' src/requirements.txt | grep -v '^#' | grep -v '^$' > /tmp/reqs.txt && \ + pip3 install --no-cache-dir -r /tmp/reqs.txt && \ + SDK_SPEC=$(grep '^connectors-sdk' src/requirements.txt | sed 's/^connectors-sdk[[:space:]]*@[[:space:]]*//') && \ + pip3 install --no-cache-dir --no-deps "$SDK_SPEC" && \ + rm -f /tmp/reqs.txt && \ + apk del .build-deps + +COPY entrypoint.sh / +RUN sed -i 's/\r$//' /entrypoint.sh && chmod +x /entrypoint.sh + +RUN addgroup -S opencti && adduser -S opencti -G opencti && \ + chown -R opencti:opencti /opt/opencti-connector-rst-threat-library /entrypoint.sh + +USER opencti:opencti + +HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \ + CMD sh -c "ps | grep -q '[p]ython3'" +ENTRYPOINT ["/bin/sh", "/entrypoint.sh"] diff --git a/external-import/rst-threat-library/README.md b/external-import/rst-threat-library/README.md new file mode 100644 index 0000000000..a9132d9792 --- /dev/null +++ b/external-import/rst-threat-library/README.md @@ -0,0 +1,80 @@ +# OpenCTI RST Threat Library Connector + +Ingests intrusion sets, malware, tools, and campaigns from the **RST Cloud Threat Library REST API** (`https://api.rstcloud.net/v1`) into OpenCTI, and keeps them in sync as upstream objects change. + +## Introduction + +The connector polls `GET /threat-objects/` for each configured type and upserts the results into OpenCTI keyed on `standard_id`. + +## Installation + +### Requirements + +- OpenCTI Platform >= 6.8.12 with a working worker +- Python >= 3.11 (for manual deployment) +- An RST Cloud Threat Library API key +- Docker / Docker Compose + +### Configuration variables + +Configuration is set either in `docker-compose.yml` (Docker) or in `config.yml` (manual deployment). See `config.yml.sample` for a full annotated example. Field descriptions and examples are also defined on the Pydantic models in `src/connector/settings.py` and mirrored in `__metadata__/connector_config_schema.json`. + +#### OpenCTI environment variables + +| Parameter | config.yml | Docker environment variable | Mandatory | Example | Description | +| ------------- | ---------- | --------------------------- | --------- | ----------------------- | ------------------------------------------------ | +| OpenCTI URL | `url` | `OPENCTI_URL` | Yes | `http://localhost:8080` | URL of the OpenCTI platform. | +| OpenCTI Token | `token` | `OPENCTI_TOKEN` | Yes | `ChangeMe` | Admin / API token used to bootstrap the connector. | + +#### Base connector environment variables + +| Parameter | config.yml | Docker environment variable | Default | Mandatory | Example | Description | +| ------------------------------ | -------------------------------------------- | -------------------------------------------------------- | ------------------- | --------- | -------------------------------------------- | --------------------------------------------------------------------------- | +| Connector ID | `id` | `CONNECTOR_ID` | / | Yes | `b7f9a6b4-6b2a-4c3f-9f7b-8b5e5b6f2d1a` | Unique UUID v4 for this connector instance. | +| Connector Type | `type` | `CONNECTOR_TYPE` | `EXTERNAL_IMPORT` | Yes | `EXTERNAL_IMPORT` | Must be `EXTERNAL_IMPORT`. | +| Connector Name | `name` | `CONNECTOR_NAME` | `RST Threat Library`| Yes | `RST Threat Library` | Display name in OpenCTI. | +| Connector Scope | `scope` | `CONNECTOR_SCOPE` | / | Yes | `intrusion-set,malware,tool,campaign` | STIX domain types emitted. | +| Log Level | `log_level` | `CONNECTOR_LOG_LEVEL` | `error` | No | `info` | `debug`, `info`, `warn`, or `error`. | +| Duration Period | `duration_period` | `CONNECTOR_DURATION_PERIOD` | `PT1H` | No | `PT1H` | ISO-8601 interval between runs. | +| Queue Threshold | `queue_threshold` | `CONNECTOR_QUEUE_THRESHOLD` | `500` | No | `500` | Max RabbitMQ queue size (MB) before pausing ingestion. | +| Update Existing Data | `update_existing_data` | `CONNECTOR_UPDATE_EXISTING_DATA` | `true` | No | `true` | Upsert existing STIX objects when `true`. | +| Auto-create Service Account | `auto_create_service_account` | `CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT` | `false` | No | `true` | Create a Connectors-group service account on first start. | +| Service Account Confidence | `auto_create_service_account_confidence_level` | `CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT_CONFIDENCE_LEVEL` | `50` | No | `50` | Max confidence for the auto-created service account. | + +#### Connector extra parameters + +| Parameter | config.yml | Docker environment variable | Default | Mandatory | Example | Description | +| --------------------------------- | ----------------------------------- | ---------------------------------------------------- | -------------------------------------------- | --------- | -------------------------------------------- | --------------------------------------------------------------------------- | +| API base URL | `baseurl` | `RST_THREAT_LIBRARY_BASEURL` | `https://api.rstcloud.net/v1` | Yes | `https://api.rstcloud.net/v1` | RST Cloud API root. | +| API key | `apikey` | `RST_THREAT_LIBRARY_APIKEY` | / | Yes | `ChangeMe` | RST Cloud Threat Library API key. | +| Auth header name | `auth_header` | `RST_THREAT_LIBRARY_AUTH_HEADER` | `x-api-key` | No | `x-api-key` | HTTP header carrying the API key. | +| HTTP proxy | `proxy` | `RST_THREAT_LIBRARY_PROXY` | *(empty)* | No | `http://proxy.example.com:8080` | Forward HTTP proxy URL. Empty = direct egress. | +| Verify TLS | `ssl_verify` | `RST_THREAT_LIBRARY_SSL_VERIFY` | `true` | No | `true` | Verify TLS certificates. | +| Connect timeout | `contimeout` | `RST_THREAT_LIBRARY_CONTIMEOUT` | `30` | No | `30` | HTTP connect timeout in seconds. | +| Read timeout | `readtimeout` | `RST_THREAT_LIBRARY_READTIMEOUT` | `120` | No | `600` | HTTP read timeout in seconds. | +| HTTP fetch retries | `retry` | `RST_THREAT_LIBRARY_RETRY` | `2` | No | `10` | Per-request retry count. | +| OpenCTI push max retries | `max_retries` | `RST_THREAT_LIBRARY_MAX_RETRIES` | `3` | No | `3` | Attempts when pushing to OpenCTI (minimum 1). | +| OpenCTI push retry delay | `retry_delay` | `RST_THREAT_LIBRARY_RETRY_DELAY` | `10` | No | `10` | Initial retry delay in seconds. | +| OpenCTI push backoff multiplier | `retry_backoff_multiplier` | `RST_THREAT_LIBRARY_RETRY_BACKOFF_MULTIPLIER` | `2.0` | No | `2.0` | Exponential backoff multiplier. | +| OpenCTI push mode | `opencti_push_mode` | `RST_THREAT_LIBRARY_OPENCTI_PUSH_MODE` | `bundle` | No | `bundle` | `bundle` (worker) or `api` (GraphQL import). | +| Object types to pull | `object_types` | `RST_THREAT_LIBRARY_OBJECT_TYPES` | `intrusion-sets,malware,tools,campaigns` | No | `intrusion-sets,malware,tools,campaigns` | Comma-separated paths under `/threat-objects/`. | +| Sort field | `order_by` | `RST_THREAT_LIBRARY_ORDER_BY` | `modified` | No | `modified` | Use `modified` for incremental sync. | +| Sort direction | `order_mode` | `RST_THREAT_LIBRARY_ORDER_MODE` | `desc` | No | `desc` | `asc` or `desc`. | +| Page size | `page_size` | `RST_THREAT_LIBRARY_PAGE_SIZE` | `100` | No | `20` | `limit` per request. | +| Intrusion-set merge/split | `merge_split` | `RST_THREAT_LIBRARY_MERGE_SPLIT` | `false` | No | `false` | Reconcile intrusion-set alias merge/split. | +| Respect local user edits | `respect_user_edits` | `RST_THREAT_LIBRARY_RESPECT_USER_EDITS` | `false` | No | `false` | Preserve higher-confidence OpenCTI edits. | +| Intrusion-set import confidence | `intrusion_set_default_confidence` | `RST_THREAT_LIBRARY_INTRUSION_SET_DEFAULT_CONFIDENCE`| *(unset)* | No | `80` | When set (0–100), replaces upstream confidence on intrusion sets. | +| Sync labels on import | `sync_labels` | `RST_THREAT_LIBRARY_SYNC_LABELS` | `RST Threat Library` | No | `RST Threat Library` | Labels merged on import; scopes merge/split. | +| Reconcile exclude labels | `reconcile_exclude_labels` | `RST_THREAT_LIBRARY_RECONCILE_EXCLUDE_LABELS` | *(empty)* | No | `MITRE,manual` | Labels excluded from merge/split fusion. | +| Reconcile createdBy allowlist | `reconcile_allow_created_by` | `RST_THREAT_LIBRARY_RECONCILE_ALLOW_CREATED_BY` | *(empty)* | No | `identity--…` | If set, fuse only entities with these `createdBy` IDs. | +| Initial backfill cutoff | `import_from_date` | `RST_THREAT_LIBRARY_IMPORT_FROM_DATE` | *(empty)* | No | `2024-01-01` | `YYYY-MM-DD`. Empty = full history on first run. | + +### Recommended values (RST API reliability) + +For deep pagination (especially `tools`), these reduce `504 Gateway Timeout` and `IncompleteRead` errors: + +```env +RST_THREAT_LIBRARY_PAGE_SIZE=20 +RST_THREAT_LIBRARY_READTIMEOUT=600 +RST_THREAT_LIBRARY_RETRY=10 +``` diff --git a/external-import/rst-threat-library/__metadata__/CONNECTOR_CONFIG_DOC.md b/external-import/rst-threat-library/__metadata__/CONNECTOR_CONFIG_DOC.md new file mode 100644 index 0000000000..72d9381603 --- /dev/null +++ b/external-import/rst-threat-library/__metadata__/CONNECTOR_CONFIG_DOC.md @@ -0,0 +1,45 @@ +# Connector Configurations + +Below is an exhaustive list of the environment variables supported by the RST +Threat Library connector. Required values must be supplied before starting the +connector. + +### Type: `object` + +| Property | Type | Required | Possible values | Default | Description | +| -------- | ---- | -------- | --------------- | ------- | ----------- | +| OPENCTI_URL | `string` | Yes | URL | | The base URL of the OpenCTI instance. | +| OPENCTI_TOKEN | `string` | Yes | string | | The API token used to connect to OpenCTI. | +| CONNECTOR_ID | `string` | Yes | UUID v4 | | A unique identifier for this connector instance. | +| CONNECTOR_TYPE | `const` | Yes | `EXTERNAL_IMPORT` | `"EXTERNAL_IMPORT"` | The OpenCTI connector type. | +| CONNECTOR_NAME | `string` | Yes | string | `"RST Threat Library"` | The connector name displayed in OpenCTI. | +| CONNECTOR_SCOPE | `string` | Yes | string | `"intrusion-set,malware,tool,campaign"` | Comma-separated STIX domain types emitted by the connector. | +| CONNECTOR_LOG_LEVEL | `string` | | `debug`, `info`, `warn`, `warning`, `error` | `"error"` | The minimum level of logs to display. | +| CONNECTOR_DURATION_PERIOD | `string` | | ISO-8601 duration | `"PT1H"` | The period between connector runs. | +| CONNECTOR_QUEUE_THRESHOLD | `number` | | Number greater than 0 | `500` | Maximum RabbitMQ queue size in MB before ingestion pauses. | +| CONNECTOR_UPDATE_EXISTING_DATA | `boolean` | | `true`, `false` | `true` | Whether existing STIX objects may be updated. | +| CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT | `boolean` | | `true`, `false` | `false` | Whether to create a dedicated Connectors-group service account on first start. | +| CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT_CONFIDENCE_LEVEL | `integer` | | `0` to `100` | `50` | Maximum confidence level assigned to the auto-created service account. | +| RST_THREAT_LIBRARY_BASEURL | `string` | Yes | URL | `"https://api.rstcloud.net/v1"` | The RST Cloud Threat Library API base URL. | +| RST_THREAT_LIBRARY_APIKEY | `string` | Yes | string | | The RST Cloud Threat Library API key. | +| RST_THREAT_LIBRARY_AUTH_HEADER | `string` | | string | `"x-api-key"` | The HTTP header used to send the API key. | +| RST_THREAT_LIBRARY_PROXY | `string` | | URL or empty string | `""` | Optional forward HTTP proxy URL. An empty value uses direct egress. | +| RST_THREAT_LIBRARY_SSL_VERIFY | `boolean` | | `true`, `false` | `true` | Whether TLS certificates are verified for API requests. | +| RST_THREAT_LIBRARY_CONTIMEOUT | `integer` | | Positive integer | `30` | HTTP connection timeout in seconds. | +| RST_THREAT_LIBRARY_READTIMEOUT | `integer` | | Positive integer | `120` | HTTP read timeout in seconds. | +| RST_THREAT_LIBRARY_RETRY | `integer` | | Non-negative integer | `2` | Number of retries for each RST API request. | +| RST_THREAT_LIBRARY_MAX_RETRIES | `integer` | | Positive integer (≥ 1) | `3` | Maximum attempts when sending data to OpenCTI (at least one). | +| RST_THREAT_LIBRARY_RETRY_DELAY | `integer` | | Non-negative integer | `10` | Initial delay in seconds before retrying an OpenCTI push. | +| RST_THREAT_LIBRARY_RETRY_BACKOFF_MULTIPLIER | `number` | | Positive number | `2.0` | Exponential backoff multiplier for OpenCTI push retries. | +| RST_THREAT_LIBRARY_OPENCTI_PUSH_MODE | `string` | | `bundle`, `api` | `"bundle"` | The OpenCTI write path: worker bundle or GraphQL API import. | +| RST_THREAT_LIBRARY_OBJECT_TYPES | `string` | | Comma-separated string | `"intrusion-sets,malware,tools,campaigns"` | Threat-object API paths to poll. | +| RST_THREAT_LIBRARY_ORDER_BY | `string` | | string | `"modified"` | Field used to order API results for incremental synchronization. | +| RST_THREAT_LIBRARY_ORDER_MODE | `string` | | `asc`, `desc` | `"desc"` | Direction used to order API results. | +| RST_THREAT_LIBRARY_PAGE_SIZE | `integer` | | Positive integer | `100` | Maximum number of threat objects requested per page. | +| RST_THREAT_LIBRARY_MERGE_SPLIT | `boolean` | | `true`, `false` | `false` | Whether intrusion-set alias merge/split reconciliation is enabled. | +| RST_THREAT_LIBRARY_RESPECT_USER_EDITS | `boolean` | | `true`, `false` | `false` | Whether higher-confidence OpenCTI edits are preserved. | +| RST_THREAT_LIBRARY_INTRUSION_SET_DEFAULT_CONFIDENCE | `integer` | | `0` to `100` | | Optional confidence value that replaces upstream confidence on imported intrusion sets. | +| RST_THREAT_LIBRARY_SYNC_LABELS | `string` | | Comma-separated string | `"RST Threat Library"` | Labels applied during import and used to scope merge/split reconciliation. | +| RST_THREAT_LIBRARY_RECONCILE_EXCLUDE_LABELS | `string` | | Comma-separated string | `""` | Labels that exclude entities from merge/split fusion. | +| RST_THREAT_LIBRARY_RECONCILE_ALLOW_CREATED_BY | `string` | | Comma-separated identity IDs | `""` | When set, only entities created by these identities may be fused. | +| RST_THREAT_LIBRARY_IMPORT_FROM_DATE | `string` | | Date in `YYYY-MM-DD` format or empty string | `""` | Initial backfill cutoff. An empty value imports the full available history. | diff --git a/external-import/rst-threat-library/__metadata__/connector_config_schema.json b/external-import/rst-threat-library/__metadata__/connector_config_schema.json new file mode 100644 index 0000000000..bb5096fdf8 --- /dev/null +++ b/external-import/rst-threat-library/__metadata__/connector_config_schema.json @@ -0,0 +1,339 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://www.filigran.io/connectors/rst-threat-library_config.schema.json", + "type": "object", + "properties": { + "OPENCTI_URL": { + "description": "The base URL of the OpenCTI instance.", + "format": "uri", + "maxLength": 2083, + "minLength": 1, + "type": "string" + }, + "OPENCTI_TOKEN": { + "description": "The API token to connect to OpenCTI.", + "type": "string" + }, + "CONNECTOR_ID": { + "description": "A unique identifier for this connector instance (UUID v4).", + "format": "uuid", + "type": "string" + }, + "CONNECTOR_NAME": { + "default": "RST Threat Library", + "description": "The name of the connector.", + "examples": [ + "RST Threat Library" + ], + "type": "string" + }, + "CONNECTOR_SCOPE": { + "description": "The scope of the connector, e.g. 'flashpoint'.", + "items": { + "type": "string" + }, + "type": "array" + }, + "CONNECTOR_LOG_LEVEL": { + "default": "error", + "description": "The minimum level of logs to display.", + "enum": [ + "debug", + "info", + "warn", + "warning", + "error" + ], + "type": "string" + }, + "CONNECTOR_TYPE": { + "const": "EXTERNAL_IMPORT", + "default": "EXTERNAL_IMPORT", + "type": "string" + }, + "CONNECTOR_DURATION_PERIOD": { + "default": "PT1H", + "description": "The period of time to await between two runs of the connector.", + "examples": [ + "PT1H", + "PT30M" + ], + "format": "duration", + "type": "string" + }, + "CONNECTOR_QUEUE_THRESHOLD": { + "default": 500.0, + "description": "Server capacity: max RabbitMQ queue size (in MB) before the connector pauses ingestion. Surfaced in the OpenCTI UI.", + "examples": [ + 500.0 + ], + "exclusiveMinimum": 0, + "type": "number" + }, + "CONNECTOR_UPDATE_EXISTING_DATA": { + "default": true, + "description": "Whether to update existing STIX objects in OpenCTI.", + "examples": [ + true, + false + ], + "type": "boolean" + }, + "CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT": { + "default": false, + "description": "Create a dedicated Connectors-group service account for this connector on first start and run subsequent API calls as that user.", + "examples": [ + true, + false + ], + "type": "boolean" + }, + "CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT_CONFIDENCE_LEVEL": { + "default": 50, + "description": "Max confidence level for the auto-created connector service account.", + "examples": [ + 50, + 80 + ], + "maximum": 100, + "minimum": 0, + "type": "integer" + }, + "RST_THREAT_LIBRARY_BASEURL": { + "default": "https://api.rstcloud.net/v1", + "description": "RST Cloud API base URL.", + "examples": [ + "https://api.rstcloud.net/v1" + ], + "format": "uri", + "maxLength": 2083, + "minLength": 1, + "type": "string" + }, + "RST_THREAT_LIBRARY_APIKEY": { + "description": "RST Cloud Threat Library API key.", + "examples": [ + "ChangeMe" + ], + "type": "string" + }, + "RST_THREAT_LIBRARY_AUTH_HEADER": { + "default": "x-api-key", + "description": "HTTP header name used to send the API key.", + "examples": [ + "x-api-key" + ], + "type": "string" + }, + "RST_THREAT_LIBRARY_CONTIMEOUT": { + "default": 30, + "description": "HTTP connect timeout in seconds.", + "examples": [ + 30 + ], + "exclusiveMinimum": 0, + "type": "integer" + }, + "RST_THREAT_LIBRARY_READTIMEOUT": { + "default": 120, + "description": "HTTP read timeout in seconds.", + "examples": [ + 120, + 600 + ], + "exclusiveMinimum": 0, + "type": "integer" + }, + "RST_THREAT_LIBRARY_RETRY": { + "default": 2, + "description": "Per-request HTTP retry count.", + "examples": [ + 2, + 10 + ], + "minimum": 0, + "type": "integer" + }, + "RST_THREAT_LIBRARY_SSL_VERIFY": { + "default": true, + "description": "Verify TLS certificates for API requests.", + "examples": [ + true, + false + ], + "type": "boolean" + }, + "RST_THREAT_LIBRARY_PAGE_SIZE": { + "default": 100, + "description": "Page size (limit) for Threat Library list requests.", + "examples": [ + 20, + 100 + ], + "exclusiveMinimum": 0, + "type": "integer" + }, + "RST_THREAT_LIBRARY_ORDER_BY": { + "default": "modified", + "description": "Sort field for incremental polling.", + "examples": [ + "modified" + ], + "type": "string" + }, + "RST_THREAT_LIBRARY_ORDER_MODE": { + "default": "desc", + "description": "Sort direction for incremental polling.", + "enum": [ + "asc", + "desc" + ], + "examples": [ + "desc", + "asc" + ], + "type": "string" + }, + "RST_THREAT_LIBRARY_PROXY": { + "default": "", + "description": "Optional forward HTTP proxy URL. Empty means direct egress.", + "examples": [ + "", + "http://proxy.example.com:8080" + ], + "type": "string" + }, + "RST_THREAT_LIBRARY_MAX_RETRIES": { + "default": 3, + "description": "Maximum attempts when pushing bundles to OpenCTI (at least one).", + "examples": [ + 3 + ], + "minimum": 1, + "type": "integer" + }, + "RST_THREAT_LIBRARY_RETRY_DELAY": { + "default": 10, + "description": "Initial retry delay in seconds for OpenCTI push failures.", + "examples": [ + 10 + ], + "minimum": 0, + "type": "integer" + }, + "RST_THREAT_LIBRARY_RETRY_BACKOFF_MULTIPLIER": { + "default": 2.0, + "description": "Exponential backoff multiplier for OpenCTI push retries.", + "examples": [ + 2.0 + ], + "exclusiveMinimum": 0, + "type": "number" + }, + "RST_THREAT_LIBRARY_OBJECT_TYPES": { + "default": "intrusion-sets,malware,tools,campaigns", + "description": "Comma-separated threat-object paths to poll.", + "examples": [ + "intrusion-sets,malware,tools,campaigns" + ], + "items": { + "type": "string" + }, + "type": "array" + }, + "RST_THREAT_LIBRARY_IMPORT_FROM_DATE": { + "default": "", + "description": "Optional initial backfill cutoff (YYYY-MM-DD).", + "examples": [ + "", + "2024-01-01" + ], + "type": "string" + }, + "RST_THREAT_LIBRARY_OPENCTI_PUSH_MODE": { + "default": "bundle", + "description": "OpenCTI write path: bundle (worker) or api (GraphQL import).", + "enum": [ + "bundle", + "api" + ], + "examples": [ + "bundle", + "api" + ], + "type": "string" + }, + "RST_THREAT_LIBRARY_SYNC_LABELS": { + "default": "RST Threat Library", + "description": "Labels merged on import; scopes merge/split.", + "examples": [ + "RST Threat Library" + ], + "items": { + "type": "string" + }, + "type": "array" + }, + "RST_THREAT_LIBRARY_RECONCILE_EXCLUDE_LABELS": { + "default": "", + "description": "Entities with these labels are excluded from merge/split fusion.", + "examples": [ + "", + "MITRE,manual" + ], + "items": { + "type": "string" + }, + "type": "array" + }, + "RST_THREAT_LIBRARY_RECONCILE_ALLOW_CREATED_BY": { + "default": "", + "description": "If set, merge/split fuses only entities with these createdBy IDs.", + "examples": [ + "", + "identity--11111111-1111-1111-1111-111111111111" + ], + "items": { + "type": "string" + }, + "type": "array" + }, + "RST_THREAT_LIBRARY_MERGE_SPLIT": { + "default": false, + "description": "Enable intrusion-set merge/split against the full catalogue.", + "examples": [ + false, + true + ], + "type": "boolean" + }, + "RST_THREAT_LIBRARY_RESPECT_USER_EDITS": { + "default": false, + "description": "Preserve OpenCTI content when its confidence exceeds upstream.", + "examples": [ + false, + true + ], + "type": "boolean" + }, + "RST_THREAT_LIBRARY_INTRUSION_SET_DEFAULT_CONFIDENCE": { + "default": null, + "description": "When set, replaces Threat Library confidence on imported intrusion sets (STIX confidence / OpenCTI source confidence).", + "examples": [ + null, + 80 + ], + "maximum": 100, + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "OPENCTI_URL", + "OPENCTI_TOKEN", + "CONNECTOR_ID", + "CONNECTOR_SCOPE", + "RST_THREAT_LIBRARY_APIKEY" + ], + "additionalProperties": true +} diff --git a/external-import/rst-threat-library/__metadata__/connector_manifest.json b/external-import/rst-threat-library/__metadata__/connector_manifest.json new file mode 100644 index 0000000000..c77bb2758b --- /dev/null +++ b/external-import/rst-threat-library/__metadata__/connector_manifest.json @@ -0,0 +1,22 @@ +{ + "title": "RST Threat Library", + "slug": "rst-threat-library", + "description": "Imports RST Cloud Threat Library intrusion sets, malware, tools and campaigns into OpenCTI.", + "short_description": "Imports RST Cloud Threat Library into OpenCTI.", + "logo": "external-import/rst-threat-library/__metadata__/logo.png", + "use_cases": [ + "Commercial Threat Intel" + ], + "verified": false, + "last_verified_date": null, + "playbook_supported": false, + "max_confidence_level": 50, + "support_version": ">=6.8.12", + "subscription_link": null, + "source_code": null, + "manager_supported": false, + "container_version": "rolling", + "container_image": "opencti/connector-rst-threat-library", + "container_type": "EXTERNAL_IMPORT" +} + diff --git a/external-import/rst-threat-library/__metadata__/logo.png b/external-import/rst-threat-library/__metadata__/logo.png new file mode 100644 index 0000000000..5b329c8291 Binary files /dev/null and b/external-import/rst-threat-library/__metadata__/logo.png differ diff --git a/external-import/rst-threat-library/config.yml.sample b/external-import/rst-threat-library/config.yml.sample new file mode 100644 index 0000000000..1ee9394494 --- /dev/null +++ b/external-import/rst-threat-library/config.yml.sample @@ -0,0 +1,39 @@ +opencti: + url: 'http://localhost:8080' # example: 'http://localhost:8080' + token: 'ChangeMe' # OpenCTI admin / API token + +connector: + type: 'EXTERNAL_IMPORT' + id: 'ChangeMe' # example UUID v4: 'b7f9a6b4-6b2a-4c3f-9f7b-8b5e5b6f2d1a' + name: 'RST Threat Library' # optional (default: 'RST Threat Library') + scope: 'intrusion-set,malware,tool,campaign' # STIX domain types emitted + log_level: 'error' # optional: 'debug', 'info', 'warn', 'error' (default: 'error') + duration_period: 'PT1H' # optional, ISO-8601 interval between runs (default: 'PT1H') + queue_threshold: 500 # optional (default: 500) + update_existing_data: true # optional (default: true) + auto_create_service_account: true # optional (default: false in settings; true in compose) + auto_create_service_account_confidence_level: 50 # optional (default: 50) + +rst_threat_library: + baseurl: 'https://api.rstcloud.net/v1' # optional (default: 'https://api.rstcloud.net/v1') + apikey: 'ChangeMe' # RST Cloud Threat Library API key + auth_header: 'x-api-key' # optional (default: 'x-api-key') + contimeout: 30 # optional, HTTP connect timeout seconds (default: 30) + readtimeout: 120 # optional, HTTP read timeout seconds (default: 120; try 600 for deep pages) + retry: 2 # optional, per-request HTTP retries (default: 2; try 10 for reliability) + ssl_verify: true # optional (default: true) + page_size: 100 # optional (default: 100; try 20 for reliability) + order_by: 'modified' # optional (default: 'modified') + order_mode: 'desc' # optional: 'asc' or 'desc' (default: 'desc') + proxy: '' # optional forward HTTP proxy URL, e.g. 'http://proxy.example.com:8080' + max_retries: 3 # optional, OpenCTI push retries (default: 3) + retry_delay: 10 # optional, OpenCTI push retry delay seconds (default: 10) + retry_backoff_multiplier: 2.0 # optional (default: 2.0) + object_types: 'intrusion-sets,malware,tools,campaigns' # optional + import_from_date: '' # optional initial backfill cutoff, e.g. '2024-01-01' + opencti_push_mode: 'bundle' # optional: 'bundle' or 'api' (default: 'bundle') + sync_labels: 'RST Threat Library' # optional; scopes merge/split + reconcile_exclude_labels: '' # optional, e.g. 'MITRE,manual' + reconcile_allow_created_by: '' # optional, comma-separated identity IDs + merge_split: false # optional (default: false) + respect_user_edits: false # optional (default: false) diff --git a/external-import/rst-threat-library/docker-compose.yml b/external-import/rst-threat-library/docker-compose.yml new file mode 100644 index 0000000000..310714f8c2 --- /dev/null +++ b/external-import/rst-threat-library/docker-compose.yml @@ -0,0 +1,42 @@ +version: "3" +services: + connector-rst-threat-library: + image: opencti/connector-rst-threat-library:latest + environment: + - OPENCTI_URL=http://localhost + - OPENCTI_TOKEN=ChangeMe + - CONNECTOR_ID=ChangeMe + - CONNECTOR_NAME=RST Threat Library + - CONNECTOR_SCOPE=intrusion-set,malware,tool,campaign + - CONNECTOR_LOG_LEVEL=info + - CONNECTOR_DURATION_PERIOD=PT1H + - CONNECTOR_QUEUE_THRESHOLD=500 + - CONNECTOR_UPDATE_EXISTING_DATA=true + - CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT=true + - CONNECTOR_AUTO_CREATE_SERVICE_ACCOUNT_CONFIDENCE_LEVEL=50 + - RST_THREAT_LIBRARY_BASEURL=https://api.rstcloud.net/v1 + - RST_THREAT_LIBRARY_APIKEY=ChangeMe + - RST_THREAT_LIBRARY_AUTH_HEADER=x-api-key + - RST_THREAT_LIBRARY_PROXY= + - RST_THREAT_LIBRARY_SSL_VERIFY=true + - RST_THREAT_LIBRARY_CONTIMEOUT=30 + - RST_THREAT_LIBRARY_READTIMEOUT=120 + - RST_THREAT_LIBRARY_RETRY=2 + - RST_THREAT_LIBRARY_MAX_RETRIES=3 + - RST_THREAT_LIBRARY_RETRY_DELAY=10 + - RST_THREAT_LIBRARY_RETRY_BACKOFF_MULTIPLIER=2.0 + - RST_THREAT_LIBRARY_OBJECT_TYPES=intrusion-sets,malware,tools,campaigns + - RST_THREAT_LIBRARY_ORDER_BY=modified + - RST_THREAT_LIBRARY_ORDER_MODE=desc + - RST_THREAT_LIBRARY_PAGE_SIZE=100 + - RST_THREAT_LIBRARY_OPENCTI_PUSH_MODE=bundle + - RST_THREAT_LIBRARY_SYNC_LABELS=RST Threat Library + - RST_THREAT_LIBRARY_RECONCILE_EXCLUDE_LABELS= + - RST_THREAT_LIBRARY_RECONCILE_ALLOW_CREATED_BY= + - RST_THREAT_LIBRARY_MERGE_SPLIT=false + - RST_THREAT_LIBRARY_RESPECT_USER_EDITS=false + - RST_THREAT_LIBRARY_INTRUSION_SET_DEFAULT_CONFIDENCE= + - RST_THREAT_LIBRARY_IMPORT_FROM_DATE= + restart: always + depends_on: + - opencti \ No newline at end of file diff --git a/external-import/rst-threat-library/entrypoint.sh b/external-import/rst-threat-library/entrypoint.sh new file mode 100644 index 0000000000..3e20eaebd1 --- /dev/null +++ b/external-import/rst-threat-library/entrypoint.sh @@ -0,0 +1,4 @@ +#!/bin/sh +set -e +cd /opt/opencti-connector-rst-threat-library +exec python3 src/main.py diff --git a/external-import/rst-threat-library/src/connector/__init__.py b/external-import/rst-threat-library/src/connector/__init__.py new file mode 100644 index 0000000000..3af3db25c8 --- /dev/null +++ b/external-import/rst-threat-library/src/connector/__init__.py @@ -0,0 +1,4 @@ +from connector.connector import RSTThreatLibrary +from connector.settings import ConnectorSettings + +__all__ = ["ConnectorSettings", "RSTThreatLibrary"] diff --git a/external-import/rst-threat-library/src/connector/confidence.py b/external-import/rst-threat-library/src/connector/confidence.py new file mode 100644 index 0000000000..65748f78c9 --- /dev/null +++ b/external-import/rst-threat-library/src/connector/confidence.py @@ -0,0 +1,51 @@ +"""Confidence-based analyst edit protection for Threat Library sync.""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + + +def _norm_int(value: Any) -> Optional[int]: + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def confidence_value(source: Dict[str, Any]) -> int: + """Return STIX confidence as an integer; missing values count as 0.""" + value = _norm_int(source.get("confidence")) + return value if value is not None else 0 + + +def upstream_confidence_from_record( + stored_record: Optional[Dict[str, Any]], +) -> Optional[int]: + if not stored_record: + return None + return _norm_int(stored_record.get("upstream_confidence")) + + +def analyst_confidence_wins( + opencti_entity: Dict[str, Any], + *, + api_item: Optional[Dict[str, Any]] = None, + stored_record: Optional[Dict[str, Any]] = None, +) -> bool: + """True when OpenCTI confidence exceeds Threat Library confidence.""" + opencti_conf = confidence_value(opencti_entity) + if api_item is not None: + api_conf = confidence_value(api_item) + else: + stored = upstream_confidence_from_record(stored_record) + api_conf = stored if stored is not None else 0 + return opencti_conf > api_conf + + +def make_sync_record(api_item: Dict[str, Any]) -> Dict[str, Any]: + """Build connector state after a successful push.""" + return { + "upstream_confidence": confidence_value(api_item), + } diff --git a/external-import/rst-threat-library/src/connector/connector.py b/external-import/rst-threat-library/src/connector/connector.py new file mode 100644 index 0000000000..efdea090b2 --- /dev/null +++ b/external-import/rst-threat-library/src/connector/connector.py @@ -0,0 +1,1338 @@ +import json +import sys +import time +import traceback +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Set, Tuple + +import requests +from pycti import OpenCTIConnectorHelper, get_config_variable + +from connector.confidence import ( + confidence_value, + make_sync_record, + upstream_confidence_from_record, +) +from connector.converter_to_stix import ConverterToStix +from connector.merge_split import ( + MergeCandidate, + SplitCandidate, + analyze_intrusion_set_merge_split, + identifiers_from_api_item, + identifiers_from_opencti, + pick_opencti_merge_survivor, +) +from connector.settings import ConnectorSettings +from connector.utils import ThreatObjectType +from rst_threat_library_client import ThreatLibraryClient + +_OPENCTI_MERGE_SOURCE_BATCH = 3 +_SPLIT_FAILURE_SKIP_THRESHOLD = 3 +_SPLIT_FAILURES_STATE_KEY = "split_failures" +_MERGE_SURVIVOR_READ_ATTEMPTS = 8 +_MERGE_SURVIVOR_READ_DELAY_S = 2.0 + + +@dataclass +class UpsertPrep: + skip: bool + api_item: Dict[str, Any] + opencti_entity: Optional[Dict[str, Any]] = None + skip_reason: str = "" + + +class RSTThreatLibrary: + def __init__( + self, *, config: ConnectorSettings, helper: OpenCTIConnectorHelper + ) -> None: + self.config = config + self.helper = helper + self.converter = ConverterToStix(helper=self.helper) + + tl = self.config.rst_threat_library + + self.object_types = [str(t).strip() for t in tl.object_types if str(t).strip()] + self.import_from_date = tl.import_from_date or "" + + self._client_config = { + "baseurl": str(tl.baseurl), + "apikey": tl.apikey, + "auth_header": tl.auth_header, + "contimeout": int(tl.contimeout), + "readtimeout": int(tl.readtimeout), + "retry": int(tl.retry), + "ssl_verify": bool(tl.ssl_verify), + "page_size": int(tl.page_size), + "order_by": tl.order_by, + "order_mode": tl.order_mode, + "proxy": tl.proxy or "", + } + + self._max_retries = int(tl.max_retries) + self._retry_delay = int(tl.retry_delay) + self._retry_backoff_multiplier = float(tl.retry_backoff_multiplier) + + _push = (tl.opencti_push_mode or "bundle").strip().lower() + self.opencti_push_mode = _push if _push in ("bundle", "api") else "bundle" + + self._sync_labels = [str(x).strip() for x in tl.sync_labels if str(x).strip()] + self._reconcile_exclude_labels = [ + str(x).strip() for x in tl.reconcile_exclude_labels if str(x).strip() + ] + self._reconcile_allow_created_by = [ + str(x).strip() for x in tl.reconcile_allow_created_by if str(x).strip() + ] + + self.update_existing_data = bool( + get_config_variable( + "CONNECTOR_UPDATE_EXISTING_DATA", + ["connector", "update_existing_data"], + self.config.to_helper_config(), + default=True, + ) + ) + + self.merge_split_enabled = bool(tl.merge_split) + self.respect_user_edits = bool(tl.respect_user_edits) + self._intrusion_set_default_confidence = tl.intrusion_set_default_confidence + + def _seed_cursor(self) -> str: + s = (self.import_from_date or "").strip() + if not s: + return "" + try: + datetime.strptime(s, "%Y-%m-%d") + except ValueError: + return "" + return f"{s}T00:00:00.000Z" + + def _publish_connector_info(self, *, mark_last_run: bool) -> None: + """Populate OpenCTI connector status fields and push them to the UI. + + """ + helper = self.helper + try: + duration_period_s = self.config.connector.duration_period.total_seconds() + + try: + helper.check_connector_buffering() + except Exception: + helper.connector_info.queue_threshold = float( + helper.connect_queue_threshold + ) + + if mark_last_run: + helper.last_run_datetime() + helper.next_run_datetime(duration_period_s) + helper.force_ping() + except Exception as ex: + helper.connector_logger.warning( + f"Failed to publish connector status to OpenCTI UI: {ex}" + ) + + def process_message(self) -> None: + timestamp = int(time.time()) + + self._publish_connector_info(mark_last_run=True) + + current_state = self.helper.get_state() or {} + self._cycle(current_state, timestamp) + + self._publish_connector_info(mark_last_run=True) + + if self.helper.connect_run_and_terminate: + self.helper.connector_logger.info("Connector stopped") + self.helper.force_ping() + sys.exit(0) + + def run(self) -> None: + self.helper.connector_logger.info("Starting RST Threat Library connector") + self.helper.connector_logger.info( + f"OpenCTI push mode: {self.opencti_push_mode} " + f"(CONNECTOR_UPDATE_EXISTING_DATA={self.update_existing_data})" + ) + if self._sync_labels: + self.helper.connector_logger.info( + f"Sync labels merged on import: {self._sync_labels}" + ) + if self._reconcile_exclude_labels: + self.helper.connector_logger.info( + "Reconcile exclude-labels (excluded from merge/split): " + f"{self._reconcile_exclude_labels}" + ) + if self._reconcile_allow_created_by: + self.helper.connector_logger.info( + "Reconcile createdBy allowlist (merge/split only these authors): " + f"{self._reconcile_allow_created_by}" + ) + self.helper.connector_logger.info( + f"Intrusion-set merge/split at import: enabled={self.merge_split_enabled}" + ) + self.helper.connector_logger.info( + "Retain local user edits (confidence lock): " + f"enabled={self.respect_user_edits}" + ) + if self.respect_user_edits: + self.helper.connector_logger.info( + "User-edit policy: preserve OpenCTI content when its confidence " + "exceeds Threat Library confidence; otherwise connector overrides" + ) + if self._intrusion_set_default_confidence is not None: + self.helper.connector_logger.info( + "Intrusion-set import confidence override: " + f"{self._intrusion_set_default_confidence}" + ) + if self.merge_split_enabled: + self.helper.connector_logger.info( + "Intrusion-set duplicates use OpenCTI stix.merge (relationships, " + "sightings, notes preserved); merge/split runs after delta import." + ) + + duration_period_s = self.config.connector.duration_period.total_seconds() + self.helper.connector_logger.info( + f"Scheduled execution period: {duration_period_s}s" + ) + self.helper.schedule_process( + message_callback=self.process_message, + duration_period=duration_period_s, + ) + + def _cycle(self, state: Dict[str, Any], timestamp: int) -> None: + seed = self._seed_cursor() + client = ThreatLibraryClient( + self.helper, + base_url=self._client_config["baseurl"], + api_key=self._client_config["apikey"], + auth_header=self._client_config["auth_header"], + connect_timeout=self._client_config["contimeout"], + read_timeout=self._client_config["readtimeout"], + retry=self._client_config["retry"], + ssl_verify=self._client_config["ssl_verify"], + page_size=self._client_config["page_size"], + order_by=self._client_config["order_by"], + order_mode=self._client_config["order_mode"], + proxy=self._client_config["proxy"], + ) + + for obj_type in self.object_types: + self._cycle_type(client, obj_type, state, timestamp, seed) + + if self.merge_split_enabled: + self._apply_intrusion_set_merge_split(client, state, timestamp) + + @staticmethod + def _entity_label_values(entity: Dict[str, Any]) -> List[str]: + out: List[str] = [] + ol = entity.get("objectLabel") + if not ol: + return out + if isinstance(ol, dict) and ol.get("edges") is not None: + for edge in ol.get("edges") or []: + if not isinstance(edge, dict): + continue + node = edge.get("node") + if isinstance(node, dict) and node.get("value"): + out.append(str(node["value"])) + return out + if isinstance(ol, list): + for x in ol: + if isinstance(x, dict): + v = x.get("value") + if v: + out.append(str(v)) + elif isinstance(x, str): + out.append(x) + return out + + @staticmethod + def _created_by_standard_id(entity: Dict[str, Any]) -> Optional[str]: + cb = entity.get("createdBy") + if not cb or not isinstance(cb, dict): + return None + sid = cb.get("standard_id") + return str(sid) if sid else None + + def _sync_state(self, state: Dict[str, Any]) -> Dict[str, Any]: + return state.setdefault("fingerprints", {}) + + def _get_sync_record( + self, state: Dict[str, Any], obj_type_path: str, standard_id: str + ) -> Optional[Dict[str, Any]]: + type_map = self._sync_state(state).get(obj_type_path) or {} + rec = type_map.get(standard_id) + return rec if isinstance(rec, dict) else None + + def _set_sync_record( + self, + state: Dict[str, Any], + obj_type_path: str, + standard_id: str, + record: Dict[str, Any], + ) -> None: + type_map = self._sync_state(state).setdefault(obj_type_path, {}) + type_map[standard_id] = record + + def _record_sync_state( + self, + state: Dict[str, Any], + obj_type_path: str, + api_items: List[Dict[str, Any]], + ) -> None: + if not self.respect_user_edits: + return + for item in api_items: + sid = item.get("standard_id") + if not sid: + continue + self._set_sync_record( + state, + obj_type_path, + sid, + make_sync_record(item), + ) + + def _analyst_locks_entity( + self, + opencti_entity: Dict[str, Any], + *, + obj_type_path: str, + state: Dict[str, Any], + api_item: Optional[Dict[str, Any]] = None, + ) -> bool: + if not self.respect_user_edits: + return False + sid = opencti_entity.get("standard_id") + if not sid: + return False + if api_item is not None: + api_item = self._normalize_api_item(obj_type_path, api_item) + record = self._get_sync_record(state, obj_type_path, sid) + return self._analyst_confidence_wins( + opencti_entity, + obj_type_path=obj_type_path, + api_item=api_item, + stored_record=record, + ) + + def _upstream_confidence_for_lock( + self, + obj_type_path: str, + *, + api_item: Optional[Dict[str, Any]] = None, + stored_record: Optional[Dict[str, Any]] = None, + ) -> int: + if api_item is not None: + return confidence_value(api_item) + if ( + obj_type_path == ThreatObjectType.INTRUSION_SETS + and self._intrusion_set_default_confidence is not None + ): + return self._intrusion_set_default_confidence + stored = upstream_confidence_from_record(stored_record) + return stored if stored is not None else 0 + + def _analyst_confidence_wins( + self, + opencti_entity: Dict[str, Any], + *, + obj_type_path: str, + api_item: Optional[Dict[str, Any]] = None, + stored_record: Optional[Dict[str, Any]] = None, + ) -> bool: + opencti_conf = confidence_value(opencti_entity) + upstream_conf = self._upstream_confidence_for_lock( + obj_type_path, + api_item=api_item, + stored_record=stored_record, + ) + return opencti_conf > upstream_conf + + def _prepare_upsert_item( + self, + obj_type_path: str, + api_item: Dict[str, Any], + state: Dict[str, Any], + ) -> UpsertPrep: + api_item = self._normalize_api_item(obj_type_path, api_item) + if not self.respect_user_edits: + return UpsertPrep(skip=False, api_item=api_item) + + sid = api_item.get("standard_id") + if not sid: + return UpsertPrep(skip=False, api_item=api_item) + + opencti_entity = self._read_opencti_entity(obj_type_path, sid) + if not opencti_entity: + return UpsertPrep(skip=False, api_item=api_item) + + if self._analyst_locks_entity( + opencti_entity, + obj_type_path=obj_type_path, + state=state, + api_item=api_item, + ): + oc_conf = confidence_value(opencti_entity) + api_conf = confidence_value(api_item) + return UpsertPrep( + skip=True, + api_item=api_item, + opencti_entity=opencti_entity, + skip_reason=( + "OpenCTI confidence " + f"({oc_conf}) exceeds Threat Library confidence " + f"({api_conf}) retains analyst edits" + ), + ) + + oc_conf = confidence_value(opencti_entity) + api_conf = confidence_value(api_item) + if oc_conf < api_conf: + self.helper.connector_logger.info( + f"[{obj_type_path}] connector override for {sid} " + f"(name={api_item.get('name')}) — OpenCTI confidence " + f"({oc_conf}) is below Threat Library confidence ({api_conf})" + ) + + return UpsertPrep( + skip=False, + api_item=api_item, + opencti_entity=opencti_entity, + ) + + def _upsert_sdo_from_prep( + self, obj_type_path: str, prep: UpsertPrep + ) -> Optional[Any]: + if prep.skip: + return None + return self._item_to_sdo(prep.api_item, obj_type_path) + + def _read_opencti_entity( + self, obj_type_path: str, standard_id: str + ) -> Optional[Dict[str, Any]]: + api = self.helper.api + try: + if obj_type_path == ThreatObjectType.INTRUSION_SETS: + return api.intrusion_set.read(id=standard_id) + if obj_type_path == ThreatObjectType.MALWARE: + return api.malware.read(id=standard_id) + if obj_type_path == ThreatObjectType.TOOLS: + return api.tool.read(id=standard_id) + if obj_type_path == ThreatObjectType.CAMPAIGNS: + return api.campaign.read(id=standard_id) + except Exception as ex: + self.helper.connector_logger.debug( + f"[{obj_type_path}] read {standard_id} for confidence check: {ex}" + ) + return None + + def _wait_for_opencti_entity( + self, + obj_type_path: str, + standard_id: str, + *, + attempts: int = _MERGE_SURVIVOR_READ_ATTEMPTS, + delay_s: float = _MERGE_SURVIVOR_READ_DELAY_S, + context: str = "merge", + ) -> Optional[Dict[str, Any]]: + """Read an entity, retrying while bundle-mode workers run.""" + if not standard_id: + return None + + entity = self._read_opencti_entity(obj_type_path, standard_id) + if entity: + return entity + + for attempt in range(1, max(attempts, 1)): + self.helper.connector_logger.info( + f"[{obj_type_path}] {context}: waiting for {standard_id} " + f"in OpenCTI (attempt {attempt}/{attempts - 1}, " + f"delay={delay_s:.1f}s)" + ) + time.sleep(delay_s) + entity = self._read_opencti_entity(obj_type_path, standard_id) + if entity: + self.helper.connector_logger.info( + f"[{obj_type_path}] {context}: {standard_id} readable after " + f"wait (attempt {attempt}/{attempts - 1})" + ) + return entity + return None + + def _is_entity_user_edited( + self, + obj_type_path: str, + entity: Dict[str, Any], + state: Dict[str, Any], + api_item: Optional[Dict[str, Any]] = None, + ) -> bool: + return self._analyst_locks_entity( + entity, + obj_type_path=obj_type_path, + state=state, + api_item=api_item, + ) + + def _entity_has_exclude_label(self, entity: Dict[str, Any]) -> bool: + """True when entity carries a label listed in RECONCILE_EXCLUDE_LABELS.""" + if not self._reconcile_exclude_labels: + return False + labels = self._entity_label_values(entity) + return any(block in labels for block in self._reconcile_exclude_labels) + + def _should_allow_reconcile_delete(self, entity: Dict[str, Any]) -> bool: + """Return False if this entity is protected from merge/split fusion.""" + if self._entity_has_exclude_label(entity): + self.helper.connector_logger.debug( + "Merge/split skip fusion (exclude-label): " + f"standard_id={entity.get('standard_id')}" + ) + return False + + allow = self._reconcile_allow_created_by + if allow: + cb = self._created_by_standard_id(entity) + if not cb or cb not in allow: + self.helper.connector_logger.debug( + "Merge/split skip fusion (createdBy not in allowlist): " + f"standard_id={entity.get('standard_id')} createdBy={cb}" + ) + return False + + return True + + def _normalize_api_item( + self, obj_type_path: str, item: Dict[str, Any] + ) -> Dict[str, Any]: + """Apply connector defaults to upstream API payloads before sync.""" + if ( + obj_type_path != ThreatObjectType.INTRUSION_SETS + or self._intrusion_set_default_confidence is None + ): + return item + out = dict(item) + out["confidence"] = self._intrusion_set_default_confidence + return out + + def _list_opencti_domain_objects(self, obj_type_path: str) -> List[Dict[str, Any]]: + api = self.helper.api + try: + if obj_type_path == "intrusion-sets": + data = api.intrusion_set.list(getAll=True) + elif obj_type_path == "malware": + data = api.malware.list(getAll=True) + elif obj_type_path == "tools": + data = api.tool.list(getAll=True) + elif obj_type_path == "campaigns": + data = api.campaign.list(getAll=True) + else: + return [] + return list(data) if data else [] + except Exception as ex: + self.helper.connector_logger.error( + f"[{obj_type_path}] reconcile OpenCTI list failed: {ex}" + ) + return [] + + def _opencti_entities_for_merge_split( + self, + obj_type_path: str, + state: Dict[str, Any], + api_items: Optional[List[Dict[str, Any]]] = None, + ) -> List[Dict[str, Any]]: + """Intrusion sets to compare with the API catalogue for merge/split. + + """ + managed = set((state.get("managed_ids") or {}).get(obj_type_path, [])) + api_idents: Set[str] = set() + if api_items: + for item in api_items: + api_idents |= set(identifiers_from_api_item(item)) + + out: List[Dict[str, Any]] = [] + seen: Set[str] = set() + for entity in self._list_opencti_domain_objects(obj_type_path): + sid = entity.get("standard_id") + if not sid or sid in seen: + continue + if self._entity_has_exclude_label(entity): + self.helper.connector_logger.debug( + f"[{obj_type_path}] merge/split: skip " + f"{entity.get('standard_id')} (exclude-label)" + ) + continue + + labels = self._entity_label_values(entity) + include = False + if self._sync_labels and any(sl in labels for sl in self._sync_labels): + include = True + elif sid in managed: + include = True + elif api_idents and (set(identifiers_from_opencti(entity)) & api_idents): + include = True + if include: + out.append(entity) + seen.add(sid) + return out + + def _split_failure_fingerprint(self, split: SplitCandidate) -> str: + aliases = sorted(str(a) for a in (split.aliases_to_remove or []) if a) + split_off = sorted( + str(item.get("standard_id") or "") + for item in (split.split_off_api_items or []) + if item.get("standard_id") + ) + return "|".join(aliases) + "#" + ",".join(split_off) + + def _get_split_failure_entry( + self, state: Dict[str, Any], obj_type: str, oc_sid: str + ) -> Dict[str, Any]: + failures = state.setdefault(_SPLIT_FAILURES_STATE_KEY, {}) + by_type = failures.setdefault(obj_type, {}) + entry = by_type.get(oc_sid) + if not isinstance(entry, dict): + entry = {"count": 0, "fingerprint": "", "skipped": False} + by_type[oc_sid] = entry + return entry + + def _clear_split_failure( + self, state: Dict[str, Any], obj_type: str, oc_sid: str + ) -> None: + failures = state.get(_SPLIT_FAILURES_STATE_KEY) or {} + by_type = failures.get(obj_type) or {} + if oc_sid in by_type: + del by_type[oc_sid] + if not by_type: + failures.pop(obj_type, None) + if not failures: + state.pop(_SPLIT_FAILURES_STATE_KEY, None) + self.helper.set_state(state) + + def _should_skip_split_after_failures( + self, + split: SplitCandidate, + obj_type: str, + state: Dict[str, Any], + ) -> bool: + """True when this split continues to fail.""" + oc = split.opencti_entity + oc_sid = oc.get("standard_id") + if not oc_sid: + return False + + fingerprint = self._split_failure_fingerprint(split) + entry = self._get_split_failure_entry(state, obj_type, oc_sid) + if entry.get("fingerprint") != fingerprint: + entry["count"] = 0 + entry["fingerprint"] = fingerprint + entry["skipped"] = False + self.helper.set_state(state) + return False + + return bool(entry.get("skipped")) + + def _record_split_failure( + self, + split: SplitCandidate, + obj_type: str, + state: Dict[str, Any], + *, + reason: str, + ) -> None: + """Track consecutive split failures""" + oc = split.opencti_entity + oc_sid = oc.get("standard_id") + if not oc_sid: + return + + fingerprint = self._split_failure_fingerprint(split) + entry = self._get_split_failure_entry(state, obj_type, oc_sid) + if entry.get("fingerprint") != fingerprint: + entry["count"] = 0 + entry["fingerprint"] = fingerprint + entry["skipped"] = False + + entry["count"] = int(entry.get("count") or 0) + 1 + entry["fingerprint"] = fingerprint + count = entry["count"] + + if count >= _SPLIT_FAILURE_SKIP_THRESHOLD and not entry.get("skipped"): + entry["skipped"] = True + self.helper.connector_logger.info( + f"[{obj_type}] split: intrusion set skipped " + f"{oc_sid} (name={oc.get('name')}) — abandoned after " + f"{count} consecutive failures ({reason})" + ) + else: + self.helper.connector_logger.info( + f"[{obj_type}] split: skip {oc_sid} (name={oc.get('name')}) — " + f"{reason} (failure {count}/{_SPLIT_FAILURE_SKIP_THRESHOLD})" + ) + self.helper.set_state(state) + + def _apply_intrusion_set_merge_split( + self, + client: ThreatLibraryClient, + state: Dict[str, Any], + timestamp: int, + ) -> None: + """Run merge/split after import against OpenCTI entities.""" + obj_type = ThreatObjectType.INTRUSION_SETS + try: + api_items = list(client.iter_all_items(obj_type)) + except Exception as ex: + self.helper.connector_logger.error( + f"[{obj_type}] merge/split catalogue fetch failed: {ex}\n" + f"{traceback.format_exc()}" + ) + return + + opencti_entities = self._opencti_entities_for_merge_split( + obj_type, state, api_items=api_items + ) + plan = analyze_intrusion_set_merge_split(api_items, opencti_entities) + + if not plan.splits and not plan.merges: + self.helper.connector_logger.info( + f"[{obj_type}] merge/split: no candidates " + f"(api={len(api_items)}, opencti_scoped={len(opencti_entities)})" + ) + return + + self.helper.connector_logger.info( + f"[{obj_type}] merge/split: {len(plan.splits)} split(s), " + f"{len(plan.merges)} merge(s)" + ) + + for split in plan.splits: + self._execute_intrusion_set_split(split, timestamp, obj_type, state) + + for merge in plan.merges: + self._execute_intrusion_set_merge(merge, timestamp, obj_type, state) + + def _execute_intrusion_set_split( + self, + split: SplitCandidate, + timestamp: int, + obj_type: str, + state: Dict[str, Any], + ) -> None: + oc = split.opencti_entity + oid = oc.get("id") + oc_sid = oc.get("standard_id") + if not oid or not oc_sid: + return + + if self._should_skip_split_after_failures(split, obj_type, state): + self.helper.connector_logger.debug( + f"[{obj_type}] split: intrusion set skipped {oc_sid} " + f"(name={oc.get('name')}) — previously abandoned" + ) + return + + if self._analyst_locks_entity( + oc, + obj_type_path=obj_type, + state=state, + api_item=split.keep_api_item, + ): + self._record_split_failure( + split, + obj_type, + state, + reason=( + "OpenCTI confidence exceeds Threat Library (analyst lock)" + ), + ) + return + + if split.aliases_to_remove: + remaining = [ + a for a in (oc.get("aliases") or []) if a not in split.aliases_to_remove + ] + try: + self.helper.api.stix_domain_object.update_field( + id=oid, + input={"key": "aliases", "value": remaining}, + ) + self.helper.connector_logger.info( + f"[{obj_type}] split: removed aliases {split.aliases_to_remove} " + f"from {oc_sid} (name={oc.get('name')})" + ) + except Exception as ex: + self.helper.connector_logger.error( + f"[{obj_type}] split alias update failed for {oc_sid}: {ex}" + ) + self._record_split_failure( + split, + obj_type, + state, + reason=f"alias update failed: {ex}", + ) + return + + stix_objects: List[Any] = [] + seen_identities: Dict[str, bool] = {} + pushed: List[str] = [] + + refresh_items: List[Dict[str, Any]] = [] + if split.keep_api_item: + refresh_items.append(split.keep_api_item) + refresh_items.extend(split.split_off_api_items) + + pushed_items: List[Dict[str, Any]] = [] + for item in refresh_items: + prep = self._prepare_upsert_item(obj_type, item, state) + if prep.skip: + if prep.skip_reason: + sid = item.get("standard_id") + self.helper.connector_logger.info( + f"[{obj_type}] split: skip upsert for {sid} " + f"(name={item.get('name')}) — {prep.skip_reason}" + ) + continue + sdo = self._upsert_sdo_from_prep(obj_type, prep) + if sdo is None: + continue + stix_objects.append(sdo) + sid = item.get("standard_id") + if sid: + pushed.append(sid) + pushed_items.append(prep.api_item) + cb = item.get("createdBy") or {} + cb_id = cb.get("standard_id") + if cb_id and cb_id not in seen_identities: + identity = self.converter.build_identity(cb) + if identity is not None: + stix_objects.append(identity) + seen_identities[cb_id] = True + + if stix_objects and self._batch_send(stix_objects, timestamp, obj_type): + self._record_sync_state(state, obj_type, pushed_items) + managed = state.setdefault("managed_ids", {}) + cur = set(managed.get(obj_type, [])) + cur.update(pushed) + managed[obj_type] = sorted(cur) + self._clear_split_failure(state, obj_type, oc_sid) + self.helper.set_state(state) + elif not stix_objects and not split.aliases_to_remove: + self._record_split_failure( + split, + obj_type, + state, + reason="no upsertable split-off items", + ) + else: + self._clear_split_failure(state, obj_type, oc_sid) + + def _opencti_fusion_merge_intrusion_sets( + self, + target_entity: Dict[str, Any], + source_entities: List[Dict[str, Any]], + obj_type: str, + state: Dict[str, Any], + ) -> List[str]: + """Fuse duplicate intrusion sets into target via OpenCTI UI merge.""" + target_internal_id = target_entity.get("id") + target_sid = target_entity.get("standard_id") + if not target_internal_id or not target_sid: + return [] + + source_pairs: List[Tuple[str, str]] = [] + for src in source_entities: + src_sid = src.get("standard_id") + src_internal_id = src.get("id") + if not src_sid or not src_internal_id or src_sid == target_sid: + continue + if not self._should_allow_reconcile_delete(src): + self.helper.connector_logger.info( + f"[{obj_type}] merge: skip source {src_sid} " + f"(name={src.get('name')}) — protected by exclude-labels " + "or createdBy allowlist" + ) + continue + if self._is_entity_user_edited(obj_type, src, state): + self.helper.connector_logger.info( + f"[{obj_type}] merge: skip source {src_sid} " + f"(name={src.get('name')}) — OpenCTI confidence exceeds " + "Threat Library (analyst lock)" + ) + continue + source_pairs.append((src_internal_id, src_sid)) + + if not source_pairs: + return [] + + merged: List[str] = [] + for offset in range(0, len(source_pairs), _OPENCTI_MERGE_SOURCE_BATCH): + batch = source_pairs[offset : offset + _OPENCTI_MERGE_SOURCE_BATCH] + chunk_internal = [pair[0] for pair in batch] + chunk_sids = [pair[1] for pair in batch] + try: + self.helper.api.stix.merge( + id=target_internal_id, + object_ids=chunk_internal, + ) + merged.extend(chunk_sids) + self.helper.connector_logger.info( + f"[{obj_type}] OpenCTI merge: fused {chunk_sids} into " + f"{target_sid} (name={target_entity.get('name')})" + ) + except Exception as ex: + self.helper.connector_logger.error( + f"[{obj_type}] OpenCTI merge failed for sources {chunk_sids} " + f"into {target_sid}: {ex}" + ) + return merged + + def _execute_intrusion_set_merge( + self, + merge: MergeCandidate, + timestamp: int, + obj_type: str, + state: Dict[str, Any], + ) -> None: + stix_objects: List[Any] = [] + seen_identities: Dict[str, bool] = {} + pushed: List[str] = [] + + item = merge.target_api_item + target_sid = item.get("standard_id") + + score_candidates: List[Dict[str, Any]] = list( + merge.opencti_entities_to_merge + ) + if target_sid: + existing_api_entity = self._read_opencti_entity(obj_type, target_sid) + if existing_api_entity: + score_candidates.append(existing_api_entity) + + preferred_survivor = pick_opencti_merge_survivor( + target_sid or "", score_candidates + ) + preferred_survivor_sid = ( + preferred_survivor.get("standard_id") if preferred_survivor else None + ) + if ( + preferred_survivor + and preferred_survivor_sid + and preferred_survivor_sid != target_sid + ): + self.helper.connector_logger.info( + f"[{obj_type}] merge: choosing OpenCTI survivor " + f"{preferred_survivor_sid} (name={preferred_survivor.get('name')}, " + f"aliases={len(preferred_survivor.get('aliases') or [])}) over " + f"upstream {target_sid} (name={item.get('name')}) — more aliases" + ) + + survivor_prep = self._prepare_upsert_item(obj_type, item, state) + survivor_pushed: List[Dict[str, Any]] = [] + if not survivor_prep.skip: + sdo = self._upsert_sdo_from_prep(obj_type, survivor_prep) + if sdo is not None: + stix_objects.append(sdo) + if target_sid: + pushed.append(target_sid) + survivor_pushed.append(item) + cb = item.get("createdBy") or {} + cb_id = cb.get("standard_id") + if cb_id and cb_id not in seen_identities: + identity = self.converter.build_identity(cb) + if identity is not None: + stix_objects.append(identity) + seen_identities[cb_id] = True + elif survivor_prep.skip_reason: + self.helper.connector_logger.info( + f"[{obj_type}] merge: skip survivor upsert for {target_sid} " + f"(name={item.get('name')}) — {survivor_prep.skip_reason}" + ) + + if stix_objects and not self._batch_send(stix_objects, timestamp, obj_type): + self.helper.connector_logger.warning( + f"[{obj_type}] merge: survivor upsert failed for {target_sid}; " + "skipping OpenCTI fusion" + ) + return + + if survivor_pushed: + self._record_sync_state(state, obj_type, survivor_pushed) + + if not target_sid: + return + + api_target = self._wait_for_opencti_entity( + obj_type, + target_sid, + context="merge", + ) + if not api_target and survivor_prep.opencti_entity: + api_target = survivor_prep.opencti_entity + self.helper.connector_logger.warning( + f"[{obj_type}] merge: using cached OpenCTI entity for " + f"{target_sid} after post-upsert read miss" + ) + + fusion_target: Optional[Dict[str, Any]] = None + if preferred_survivor_sid and preferred_survivor_sid != target_sid: + preferred_live = self._wait_for_opencti_entity( + obj_type, + preferred_survivor_sid, + context="merge preferred survivor", + ) + if preferred_live: + fusion_target = preferred_live + elif preferred_survivor and preferred_survivor.get("id"): + fusion_target = preferred_survivor + self.helper.connector_logger.warning( + f"[{obj_type}] merge: using cached preferred survivor " + f"{preferred_survivor_sid} after read miss" + ) + elif api_target: + self.helper.connector_logger.warning( + f"[{obj_type}] merge: preferred survivor " + f"{preferred_survivor_sid} no longer readable; " + f"fusing into upstream {target_sid}" + ) + fusion_target = api_target + elif api_target: + fusion_target = api_target + elif preferred_survivor and preferred_survivor.get("id"): + fusion_target = preferred_survivor + self.helper.connector_logger.warning( + f"[{obj_type}] merge: upstream survivor {target_sid} not " + f"readable after upsert; fusing into OpenCTI candidate " + f"{fusion_target.get('standard_id')} " + f"(name={fusion_target.get('name')})" + ) + + if not fusion_target or not fusion_target.get("id"): + self.helper.connector_logger.error( + f"[{obj_type}] merge: survivor {target_sid} not found in OpenCTI " + "after upsert (and no usable OpenCTI fusion candidate)" + ) + return + + fusion_target_sid = fusion_target.get("standard_id") + if self._analyst_locks_entity( + fusion_target, + obj_type_path=obj_type, + state=state, + api_item=merge.target_api_item, + ): + self.helper.connector_logger.info( + f"[{obj_type}] merge: skip OpenCTI fusion into {fusion_target_sid} " + f"(name={fusion_target.get('name')}) — OpenCTI confidence exceeds " + "Threat Library (analyst lock)" + ) + return + + source_by_sid: Dict[str, Dict[str, Any]] = {} + for src in merge.opencti_entities_to_merge: + src_sid = src.get("standard_id") + if src_sid and src_sid != fusion_target_sid: + source_by_sid[src_sid] = src + if target_sid != fusion_target_sid and api_target: + source_by_sid[target_sid] = api_target + + merged_sids = self._opencti_fusion_merge_intrusion_sets( + fusion_target, + list(source_by_sid.values()), + obj_type, + state, + ) + + managed = state.setdefault("managed_ids", {}) + cur = set(managed.get(obj_type, [])) + if pushed: + cur.update(pushed) + for sid in merged_sids: + cur.discard(sid) + if fusion_target_sid: + cur.add(fusion_target_sid) + managed[obj_type] = sorted(cur) + self.helper.set_state(state) + + def _cycle_type( + self, + client: ThreatLibraryClient, + obj_type: str, + state: Dict[str, Any], + timestamp: int, + seed: str, + ) -> None: + cursor_key = f"cursor_{obj_type}" + cursor = state.get(cursor_key) or seed + self.helper.connector_logger.info( + f"[{obj_type}] cycle start (cursor={cursor or '(none)'})" + ) + + stix_objects: List[Any] = [] + seen_identities: Dict[str, bool] = {} + pushed_standard_ids: List[str] = [] + pushed_api_items: List[Dict[str, Any]] = [] + latest = cursor + count = 0 + skipped_user_edit = 0 + + try: + for item in client.iter_new_items(obj_type, cursor): + prep = self._prepare_upsert_item(obj_type, item, state) + mod = item.get("modified") or "" + if mod and (not latest or mod > latest): + latest = mod + count += 1 + if prep.skip: + skipped_user_edit += 1 + if prep.skip_reason: + sid = item.get("standard_id") + self.helper.connector_logger.info( + f"[{obj_type}] skip upsert for {sid} " + f"(name={item.get('name')}) — {prep.skip_reason}" + ) + continue + sdo = self._upsert_sdo_from_prep(obj_type, prep) + if sdo is None: + continue + stix_objects.append(sdo) + sid = item.get("standard_id") + if sid: + pushed_standard_ids.append(sid) + pushed_api_items.append(prep.api_item) + + cb = item.get("createdBy") or {} + cb_id = cb.get("standard_id") + if cb_id and cb_id not in seen_identities: + identity = self.converter.build_identity(cb) + if identity is not None: + stix_objects.append(identity) + seen_identities[cb_id] = True + + if count % 100 == 0: + self.helper.connector_logger.info( + f"[{obj_type}] converted {count} object(s) so far" + ) + except Exception as ex: + self.helper.connector_logger.error( + f"[{obj_type}] fetch failed: {ex}\n{traceback.format_exc()}" + ) + return + + if skipped_user_edit: + self.helper.connector_logger.info( + f"[{obj_type}] skipped {skipped_user_edit} object(s) " + "(analyst confidence lock)" + ) + + if not stix_objects: + if count: + state[cursor_key] = latest or cursor + self.helper.set_state(state) + self.helper.connector_logger.info(f"[{obj_type}] no new objects this cycle") + else: + ok = self._batch_send(stix_objects, timestamp, obj_type) + if ok: + self._record_sync_state(state, obj_type, pushed_api_items) + state[cursor_key] = latest or cursor + managed = state.setdefault("managed_ids", {}) + cur = set(managed.get(obj_type, [])) + cur.update(pushed_standard_ids) + managed[obj_type] = sorted(cur) + # Persist incrementally so a later crash/restart doesn't redo this type. + self.helper.set_state(state) + self.helper.connector_logger.info( + f"[{obj_type}] ingested {count} object(s), cursor now " + f"{state[cursor_key] or '(none)'}" + ) + else: + self.helper.connector_logger.warning( + f"[{obj_type}] OpenCTI push failed ({self.opencti_push_mode}); " + f"cursor not advanced (will retry on next cycle)" + ) + return + + def _item_to_sdo(self, item: Dict[str, Any], obj_type_path: str): + return self.converter.item_to_sdo(item, obj_type_path, self._sync_labels) + + def _batch_send( + self, stix_objects: List[Any], timestamp: int, obj_type: str + ) -> bool: + if self.opencti_push_mode == "api": + return self._batch_send_via_api(stix_objects, timestamp, obj_type) + return self._batch_send_stix_bundle(stix_objects, timestamp, obj_type) + + @staticmethod + def _is_retryable_upload_error(exc: BaseException) -> bool: + return isinstance( + exc, + ( + ConnectionError, + OSError, + TimeoutError, + requests.exceptions.RequestException, + ), + ) + + def _mark_work_failed( + self, work_id: Optional[str], obj_type: str, exc: BaseException + ) -> None: + if not work_id: + return + try: + self.helper.api.work.to_processed( + work_id, + f"[{obj_type}] upload failed: {exc}", + in_error=True, + ) + except Exception as close_ex: + self.helper.connector_logger.warning( + f"[{obj_type}] failed to mark work {work_id} in_error: {close_ex}" + ) + + def _batch_send_stix_bundle( + self, stix_objects: List[Any], timestamp: int, obj_type: str + ) -> bool: + now = datetime.fromtimestamp(timestamp, tz=timezone.utc) + friendly_name = ( + f"RST Threat Library [{obj_type}] @ " f"{now.strftime('%Y-%m-%d %H:%M:%S')}" + ) + self.helper.connector_logger.debug( + f"[{obj_type}] start uploading {len(stix_objects)} object(s) " + f"(mode=bundle)" + ) + + max_retries = max(1, int(self._max_retries)) + retry_delay = self._retry_delay + + for attempt in range(max_retries): + work_id: Optional[str] = None + try: + work_id = self.helper.api.work.initiate_work( + self.helper.connect_id, friendly_name + ) + self.helper.send_stix2_bundle( + self.helper.stix2_create_bundle(stix_objects), + update=self.update_existing_data, + work_id=work_id, + cleanup_inconsistent_bundle=True, + ) + self.helper.connector_logger.info( + f"[{obj_type}] sent bundle of {len(stix_objects)} object(s)" + ) + self.helper.api.work.to_processed( + work_id, + f"Sent bundle of {len(stix_objects)} object(s) for {obj_type}", + ) + return True + + except Exception as ex: + self._mark_work_failed(work_id, obj_type, ex) + if self._is_retryable_upload_error(ex): + self.helper.connector_logger.error( + f"[{obj_type}] push attempt {attempt + 1}/{max_retries} " + f"failed: {ex}" + ) + if attempt < max_retries - 1: + self.helper.connector_logger.info( + f"Retrying in {retry_delay} seconds..." + ) + time.sleep(retry_delay) + retry_delay = ( + int(retry_delay * self._retry_backoff_multiplier) + or retry_delay + ) + continue + self.helper.connector_logger.error( + f"[{obj_type}] failed to upload bundle after " + f"{max_retries} attempts." + ) + return False + + error_message = f"[{obj_type}] unexpected error during upload: {ex}" + self.helper.connector_logger.error(error_message) + raise + + return False + + @staticmethod + def _stix_objects_to_api_order(stix_objects: List[Any]) -> List[Any]: + """Import identities before SDOs that reference created_by_ref.""" + identities: List[Any] = [] + rest: List[Any] = [] + seen_identity: Dict[str, bool] = {} + for obj in stix_objects: + d = json.loads(obj.serialize()) + if d.get("type") == "identity": + oid = d.get("id") + if oid and oid not in seen_identity: + seen_identity[oid] = True + identities.append(obj) + else: + rest.append(obj) + return identities + rest + + def _batch_send_via_api( + self, stix_objects: List[Any], timestamp: int, obj_type: str + ) -> bool: + now = datetime.fromtimestamp(timestamp, tz=timezone.utc) + friendly_name = ( + f"RST Threat Library [{obj_type}] API @ " + f"{now.strftime('%Y-%m-%d %H:%M:%S')}" + ) + ordered = self._stix_objects_to_api_order(stix_objects) + self.helper.connector_logger.debug( + f"[{obj_type}] start API import of {len(ordered)} object(s) " + f"(update_existing={self.update_existing_data})" + ) + + max_retries = max(1, int(self._max_retries)) + retry_delay = self._retry_delay + + for attempt in range(max_retries): + work_id: Optional[str] = None + try: + work_id = self.helper.api.work.initiate_work( + self.helper.connect_id, friendly_name + ) + for obj in ordered: + data = json.loads(obj.serialize()) + self.helper.api.stix2.import_object( + data, update=self.update_existing_data + ) + self.helper.connector_logger.info( + f"[{obj_type}] API-imported {len(ordered)} object(s)" + ) + self.helper.api.work.to_processed( + work_id, + f"API-imported {len(ordered)} object(s) for {obj_type}", + ) + return True + + except Exception as ex: + self._mark_work_failed(work_id, obj_type, ex) + if self._is_retryable_upload_error(ex): + self.helper.connector_logger.error( + f"[{obj_type}] API push attempt {attempt + 1}/{max_retries} " + f"failed: {ex}" + ) + if attempt < max_retries - 1: + self.helper.connector_logger.info( + f"Retrying in {retry_delay} seconds..." + ) + time.sleep(retry_delay) + retry_delay = ( + int(retry_delay * self._retry_backoff_multiplier) + or retry_delay + ) + continue + self.helper.connector_logger.error( + f"[{obj_type}] failed API import after {max_retries} attempts." + ) + return False + + error_message = ( + f"[{obj_type}] unexpected error during API import: {ex}" + ) + self.helper.connector_logger.error(error_message) + raise + + return False diff --git a/external-import/rst-threat-library/src/connector/converter_to_stix.py b/external-import/rst-threat-library/src/connector/converter_to_stix.py new file mode 100644 index 0000000000..ccef74fe21 --- /dev/null +++ b/external-import/rst-threat-library/src/connector/converter_to_stix.py @@ -0,0 +1,177 @@ +"""Convert Threat Library API objects into STIX 2.1 SDOs. + +Threat Library objects have upstream ``standard_id`` values. +STIX objects are built with ``stix2`` directly so those IDs are preserved. The +connectors-sdk SDO models regenerate IDs from names. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +import stix2 +from pycti import OpenCTIConnectorHelper + +from connector.utils import ENTITY_TYPE_TO_STIX, PATH_TO_STIX_TYPE + + +class ConverterToStix: + """Build STIX 2.1 objects from Threat Library API payloads.""" + + def __init__(self, helper: OpenCTIConnectorHelper): + self.helper = helper + + def item_to_sdo( + self, + item: Dict[str, Any], + obj_type_path: str, + sync_labels: List[str], + ) -> Optional[Any]: + entity_type = item.get("entity_type") + stix_type = ENTITY_TYPE_TO_STIX.get(entity_type) or PATH_TO_STIX_TYPE.get( + obj_type_path + ) + builders = { + "intrusion-set": self.build_intrusion_set, + "malware": self.build_malware, + "tool": self.build_tool, + "campaign": self.build_campaign, + } + if not stix_type or stix_type not in builders: + self.helper.connector_logger.warning( + "Skipping object with unsupported entity_type", + { + "entity_type": entity_type, + "path": obj_type_path, + }, + ) + return None + if not item.get("standard_id") or not item.get("name"): + self.helper.connector_logger.warning( + "Skipping object missing standard_id or name", + {"standard_id": item.get("standard_id")}, + ) + return None + try: + from connector.utils import with_sync_labels + + merged = with_sync_labels(dict(item), sync_labels) + return builders[stix_type](merged) + except Exception as exc: + self.helper.connector_logger.error( + "Failed to convert object to STIX", + { + "stix_type": stix_type, + "standard_id": item.get("standard_id"), + "error": str(exc), + }, + ) + return None + + @staticmethod + def build_external_references(refs: List[Dict[str, Any]]) -> List[Any]: + out: List[Any] = [] + for ref in refs or []: + source_name = ref.get("source_name") + if not source_name: + continue + kwargs: Dict[str, Any] = {"source_name": source_name} + if ref.get("url"): + kwargs["url"] = ref["url"] + if ref.get("external_id"): + kwargs["external_id"] = ref["external_id"] + if ref.get("description"): + kwargs["description"] = ref["description"] + out.append(stix2.v21.ExternalReference(**kwargs)) + return out + + def _base_sdo_kwargs(self, item: Dict[str, Any]) -> Dict[str, Any]: + kwargs: Dict[str, Any] = { + "id": item["standard_id"], + "name": item.get("name") or item["standard_id"], + } + for field in ("created", "modified", "description", "revoked"): + if item.get(field) not in (None, ""): + kwargs[field] = item[field] + if item.get("confidence") is not None: + try: + kwargs["confidence"] = int(item["confidence"]) + except (TypeError, ValueError): + pass + if item.get("objectLabel"): + kwargs["labels"] = list(item["objectLabel"]) + + ext = self.build_external_references(item.get("externalReferences") or []) + if ext: + kwargs["external_references"] = ext + + created_by = item.get("createdBy") or {} + if created_by.get("standard_id"): + kwargs["created_by_ref"] = created_by["standard_id"] + + marking_refs = [ + marking["standard_id"] + for marking in (item.get("objectMarking") or []) + if marking.get("standard_id") + ] + if marking_refs: + kwargs["object_marking_refs"] = marking_refs + + return kwargs + + def build_identity(self, created_by: Dict[str, Any]) -> Optional[Any]: + sid = created_by.get("standard_id") + if not sid: + return None + name = created_by.get("name") or sid + identity_class = str(created_by.get("identity_class") or "organization") + return stix2.v21.Identity(id=sid, name=name, identity_class=identity_class) + + def build_intrusion_set(self, item: Dict[str, Any]) -> Any: + kwargs = self._base_sdo_kwargs(item) + for field in ( + "first_seen", + "last_seen", + "primary_motivation", + "resource_level", + ): + if item.get(field) not in (None, ""): + kwargs[field] = item[field] + if item.get("aliases"): + kwargs["aliases"] = list(item["aliases"]) + if item.get("goals"): + kwargs["goals"] = list(item["goals"]) + if item.get("secondary_motivations"): + kwargs["secondary_motivations"] = list(item["secondary_motivations"]) + return stix2.v21.IntrusionSet(**kwargs) + + def build_malware(self, item: Dict[str, Any]) -> Any: + kwargs = self._base_sdo_kwargs(item) + for field in ( + "malware_types", + "capabilities", + "architecture_execution_envs", + "implementation_languages", + ): + if item.get(field): + kwargs[field] = list(item[field]) + if item.get("aliases"): + kwargs["aliases"] = list(item["aliases"]) + if item.get("is_family") is not None: + kwargs["is_family"] = bool(item["is_family"]) + return stix2.v21.Malware(**kwargs) + + def build_tool(self, item: Dict[str, Any]) -> Any: + kwargs = self._base_sdo_kwargs(item) + if item.get("tool_types"): + kwargs["tool_types"] = list(item["tool_types"]) + if item.get("tool_version"): + kwargs["tool_version"] = item["tool_version"] + return stix2.v21.Tool(**kwargs) + + def build_campaign(self, item: Dict[str, Any]) -> Any: + kwargs = self._base_sdo_kwargs(item) + for field in ("first_seen", "last_seen", "objective"): + if item.get(field) not in (None, ""): + kwargs[field] = item[field] + return stix2.v21.Campaign(**kwargs) diff --git a/external-import/rst-threat-library/src/connector/merge_split.py b/external-import/rst-threat-library/src/connector/merge_split.py new file mode 100644 index 0000000000..fdee24dc9a --- /dev/null +++ b/external-import/rst-threat-library/src/connector/merge_split.py @@ -0,0 +1,223 @@ +"""Detect intrusion-set merge/split candidates by comparing alias ownership.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, FrozenSet, Iterable, List, Optional, Set + + +def normalize_identifier(value: str) -> str: + return (value or "").strip().casefold() + + +def identifiers_from_api_item(item: Dict[str, Any]) -> FrozenSet[str]: + out: Set[str] = set() + name = item.get("name") + if name: + out.add(normalize_identifier(str(name))) + for alias in item.get("aliases") or []: + if alias: + out.add(normalize_identifier(str(alias))) + return frozenset(out) + + +def identifiers_from_opencti(entity: Dict[str, Any]) -> FrozenSet[str]: + out: Set[str] = set() + name = entity.get("name") + if name: + out.add(normalize_identifier(str(name))) + for alias in entity.get("aliases") or []: + if alias: + out.add(normalize_identifier(str(alias))) + return frozenset(out) + + +def build_identifier_index( + items: Iterable[Dict[str, Any]], + *, + sid_key: str = "standard_id", + from_opencti: bool = False, +) -> Dict[str, Set[str]]: + """Map normalized name/alias -> set of standard_ids that claim it.""" + index: Dict[str, Set[str]] = {} + id_fn = identifiers_from_opencti if from_opencti else identifiers_from_api_item + for item in items: + sid = item.get(sid_key) + if not sid: + continue + for ident in id_fn(item): + index.setdefault(ident, set()).add(sid) + return index + + +@dataclass +class SplitCandidate: + """OpenCTI intrusion set spans aliases owned by multiple upstream objects.""" + + opencti_entity: Dict[str, Any] + keep_api_item: Optional[Dict[str, Any]] + aliases_to_remove: List[str] + split_off_api_items: List[Dict[str, Any]] = field(default_factory=list) + + +@dataclass +class MergeCandidate: + """Multiple OpenCTI intrusion sets should fuse into one upstream survivor.""" + + target_api_item: Dict[str, Any] + opencti_entities_to_merge: List[Dict[str, Any]] = field(default_factory=list) + + +@dataclass +class MergeSplitPlan: + splits: List[SplitCandidate] = field(default_factory=list) + merges: List[MergeCandidate] = field(default_factory=list) + + +def analyze_intrusion_set_merge_split( + api_items: List[Dict[str, Any]], + opencti_entities: List[Dict[str, Any]], +) -> MergeSplitPlan: + """Compare catalogue vs OpenCTI and classify split/merge/normal paths.""" + plan = MergeSplitPlan() + api_by_sid: Dict[str, Dict[str, Any]] = {} + for item in api_items: + sid = item.get("standard_id") + if sid: + api_by_sid[sid] = item + + opencti_by_sid: Dict[str, Dict[str, Any]] = { + e["standard_id"]: e for e in opencti_entities if e.get("standard_id") + } + + api_index = build_identifier_index(api_items) + oc_index = build_identifier_index(opencti_entities, from_opencti=True) + + split_seen: Set[str] = set() + merge_seen: Set[tuple] = set() + + for oc in opencti_entities: + oc_sid = oc.get("standard_id") + if not oc_sid: + continue + oc_ids = identifiers_from_opencti(oc) + if not oc_ids: + continue + + primary_api = api_by_sid.get(oc_sid) + primary_ids = ( + identifiers_from_api_item(primary_api) if primary_api else frozenset() + ) + + conflicting_aliases: Set[str] = set() + split_off_sids: Set[str] = set() + + for ident in oc_ids: + api_owners = api_index.get(ident, set()) + if not api_owners: + continue + if oc_sid in api_owners and len(api_owners) == 1: + continue + other_owners = api_owners - {oc_sid} + if other_owners: + if ident not in primary_ids: + raw = _raw_alias_value(oc, ident) + if raw: + conflicting_aliases.add(raw) + for owner_sid in other_owners: + split_off_sids.add(owner_sid) + + if not conflicting_aliases and not split_off_sids: + continue + if oc_sid in split_seen: + continue + + split_off_items = [ + api_by_sid[s] + for s in sorted(split_off_sids) + if s in api_by_sid and s != oc_sid + ] + if not conflicting_aliases and not split_off_items: + continue + + plan.splits.append( + SplitCandidate( + opencti_entity=oc, + keep_api_item=primary_api, + aliases_to_remove=sorted(conflicting_aliases), + split_off_api_items=split_off_items, + ) + ) + split_seen.add(oc_sid) + + for api_sid, api_item in api_by_sid.items(): + api_ids = identifiers_from_api_item(api_item) + if not api_ids: + continue + + oc_duplicates: Dict[str, Dict[str, Any]] = {} + for ident in api_ids: + for oc_sid in oc_index.get(ident, set()): + if oc_sid == api_sid: + continue + entity = opencti_by_sid.get(oc_sid) + if entity: + oc_duplicates[oc_sid] = entity + + if not oc_duplicates: + continue + + merge_key = (api_sid, tuple(sorted(oc_duplicates))) + if merge_key in merge_seen: + continue + merge_seen.add(merge_key) + + plan.merges.append( + MergeCandidate( + target_api_item=api_item, + opencti_entities_to_merge=list(oc_duplicates.values()), + ) + ) + + return plan + + +def _raw_alias_value(entity: Dict[str, Any], normalized_ident: str) -> Optional[str]: + name = entity.get("name") + if name and normalize_identifier(str(name)) == normalized_ident: + return str(name) + for alias in entity.get("aliases") or []: + if alias and normalize_identifier(str(alias)) == normalized_ident: + return str(alias) + return None + + +def opencti_alias_count(entity: Dict[str, Any]) -> int: + """Count aliases on an OpenCTI intrusion set (name is not counted).""" + return len([a for a in (entity.get("aliases") or []) if a]) + + +def pick_opencti_merge_survivor( + api_standard_id: str, + candidates: List[Dict[str, Any]], +) -> Optional[Dict[str, Any]]: + """Choose the OpenCTI entity that should survive a fusion merge. + """ + unique: Dict[str, Dict[str, Any]] = {} + for entity in candidates: + sid = entity.get("standard_id") + if sid: + unique[sid] = entity + if not unique: + return None + + def sort_key(entity: Dict[str, Any]) -> tuple: + sid = entity.get("standard_id") or "" + return ( + opencti_alias_count(entity), + 1 if sid == api_standard_id else 0, + len(identifiers_from_opencti(entity)), + sid, + ) + + return max(unique.values(), key=sort_key) diff --git a/external-import/rst-threat-library/src/connector/settings.py b/external-import/rst-threat-library/src/connector/settings.py new file mode 100644 index 0000000000..2df9a694e8 --- /dev/null +++ b/external-import/rst-threat-library/src/connector/settings.py @@ -0,0 +1,217 @@ +from datetime import timedelta +from typing import Literal + +from connectors_sdk import ( + BaseConfigModel, + BaseConnectorSettings, + BaseExternalImportConnectorConfig, +) +from connectors_sdk.settings.annotated_types import ListFromString +from connectors_sdk.settings.deprecations import migrate_deprecated_namespace +from pydantic import Field, HttpUrl, model_validator + + +class ExternalImportConnectorConfig(BaseExternalImportConnectorConfig): + """OpenCTI external-import connector settings.""" + + name: str = Field( + description="The name of the connector.", + default="RST Threat Library", + examples=["RST Threat Library"], + ) + duration_period: timedelta = Field( + description="The period of time to await between two runs of the connector.", + default=timedelta(hours=1), + examples=["PT1H", "PT30M"], + ) + queue_threshold: float = Field( + description=( + "Server capacity: max RabbitMQ queue size (in MB) before the " + "connector pauses ingestion. Surfaced in the OpenCTI UI." + ), + default=500.0, + gt=0, + examples=[500.0], + ) + update_existing_data: bool = Field( + description="Whether to update existing STIX objects in OpenCTI.", + default=True, + examples=[True, False], + ) + auto_create_service_account: bool = Field( + description=( + "Create a dedicated Connectors-group service account for this " + "connector on first start and run subsequent API calls as that user." + ), + default=False, + examples=[True, False], + ) + auto_create_service_account_confidence_level: int = Field( + description=( + "Max confidence level for the auto-created connector service account." + ), + default=50, + ge=0, + le=100, + examples=[50, 80], + ) + + +class RstThreatLibraryConfig(BaseConfigModel): + """RST Cloud Threat Library API and sync behaviour.""" + + baseurl: HttpUrl = Field( + description="RST Cloud API base URL.", + default="https://api.rstcloud.net/v1", + examples=["https://api.rstcloud.net/v1"], + ) + apikey: str = Field( + description="RST Cloud Threat Library API key.", + examples=["ChangeMe"], + ) + auth_header: str = Field( + description="HTTP header name used to send the API key.", + default="x-api-key", + examples=["x-api-key"], + ) + contimeout: int = Field( + description="HTTP connect timeout in seconds.", + default=30, + gt=0, + examples=[30], + ) + readtimeout: int = Field( + description="HTTP read timeout in seconds.", + default=120, + gt=0, + examples=[120, 600], + ) + retry: int = Field( + description="Per-request HTTP retry count.", + default=2, + ge=0, + examples=[2, 10], + ) + ssl_verify: bool = Field( + description="Verify TLS certificates for API requests.", + default=True, + examples=[True, False], + ) + page_size: int = Field( + description="Page size (limit) for Threat Library list requests.", + default=100, + gt=0, + examples=[20, 100], + ) + order_by: str = Field( + description="Sort field for incremental polling.", + default="modified", + examples=["modified"], + ) + order_mode: Literal["asc", "desc"] = Field( + description="Sort direction for incremental polling.", + default="desc", + examples=["desc", "asc"], + ) + proxy: str = Field( + description="Optional forward HTTP proxy URL. Empty means direct egress.", + default="", + examples=["", "http://proxy.example.com:8080"], + ) + max_retries: int = Field( + description="Maximum retries when pushing bundles to OpenCTI.", + default=3, + ge=1, + examples=[3], + ) + retry_delay: int = Field( + description="Initial retry delay in seconds for OpenCTI push failures.", + default=10, + ge=0, + examples=[10], + ) + retry_backoff_multiplier: float = Field( + description="Exponential backoff multiplier for OpenCTI push retries.", + default=2.0, + gt=0, + examples=[2.0], + ) + object_types: ListFromString = Field( + description="Comma-separated threat-object paths to poll.", + default="intrusion-sets,malware,tools,campaigns", + examples=["intrusion-sets,malware,tools,campaigns"], + ) + import_from_date: str = Field( + description="Optional initial backfill cutoff (YYYY-MM-DD).", + default="", + examples=["", "2024-01-01"], + ) + opencti_push_mode: Literal["bundle", "api"] = Field( + description="OpenCTI write path: bundle (worker) or api (GraphQL import).", + default="bundle", + examples=["bundle", "api"], + ) + sync_labels: ListFromString = Field( + description="Labels merged on import; scopes merge/split.", + default="RST Threat Library", + examples=["RST Threat Library"], + ) + reconcile_exclude_labels: ListFromString = Field( + description="Entities with these labels are excluded from merge/split fusion.", + default="", + examples=["", "MITRE,manual"], + ) + reconcile_allow_created_by: ListFromString = Field( + description="If set, merge/split fuses only entities with these createdBy IDs.", + default="", + examples=["", "identity--11111111-1111-1111-1111-111111111111"], + ) + merge_split: bool = Field( + description="Enable intrusion-set merge/split against the full catalogue.", + default=False, + examples=[False, True], + ) + respect_user_edits: bool = Field( + description="Preserve OpenCTI content when its confidence exceeds upstream.", + default=False, + examples=[False, True], + ) + intrusion_set_default_confidence: int | None = Field( + description=( + "When set, replaces Threat Library confidence on imported " + "intrusion sets (STIX confidence / OpenCTI source confidence)." + ), + default=None, + ge=0, + le=100, + examples=[None, 80], + ) + + +class ConnectorSettings(BaseConnectorSettings): + """Connector configuration loaded from environment variables and config.yml.""" + + connector: ExternalImportConnectorConfig = Field( + default_factory=ExternalImportConnectorConfig + ) + rst_threat_library: RstThreatLibraryConfig = Field( + default_factory=RstThreatLibraryConfig + ) + + @classmethod + def _migrate_deprecated_namespaces(cls, data: dict) -> dict: + data = super()._migrate_deprecated_namespaces(data) + migrate_deprecated_namespace( + data, + old_namespace="rst-threat-library", + new_namespace="rst_threat_library", + ) + return data + + @model_validator(mode="after") + def _require_api_key(self) -> "ConnectorSettings": + threat_library_cfg = getattr(self, "rst_threat_library", None) + api_key = getattr(threat_library_cfg, "apikey", "") + if not api_key: + raise ValueError("rst_threat_library.apikey is required.") + return self diff --git a/external-import/rst-threat-library/src/connector/utils.py b/external-import/rst-threat-library/src/connector/utils.py new file mode 100644 index 0000000000..0368f97987 --- /dev/null +++ b/external-import/rst-threat-library/src/connector/utils.py @@ -0,0 +1,39 @@ +"""Shared connector helpers.""" + +from __future__ import annotations + +from typing import Any, Dict, List + + +class ThreatObjectType: + INTRUSION_SETS = "intrusion-sets" + MALWARE = "malware" + TOOLS = "tools" + CAMPAIGNS = "campaigns" + + +PATH_TO_STIX_TYPE = { + ThreatObjectType.INTRUSION_SETS: "intrusion-set", + ThreatObjectType.MALWARE: "malware", + ThreatObjectType.TOOLS: "tool", + ThreatObjectType.CAMPAIGNS: "campaign", +} + +ENTITY_TYPE_TO_STIX = { + "Intrusion-Set": "intrusion-set", + "Malware": "malware", + "Tool": "tool", + "Campaign": "campaign", +} + + +def with_sync_labels(item: Dict[str, Any], sync_labels: List[str]) -> Dict[str, Any]: + if not sync_labels: + return item + out = dict(item) + labels = list(out.get("objectLabel") or []) + for label in sync_labels: + if label not in labels: + labels.append(label) + out["objectLabel"] = labels + return out diff --git a/external-import/rst-threat-library/src/main.py b/external-import/rst-threat-library/src/main.py new file mode 100644 index 0000000000..7b3d923b9f --- /dev/null +++ b/external-import/rst-threat-library/src/main.py @@ -0,0 +1,22 @@ +import sys +import time +import traceback + +from pycti import OpenCTIConnectorHelper + +from connector.connector import RSTThreatLibrary +from connector.settings import ConnectorSettings + +__all__ = ["RSTThreatLibrary"] + + +if __name__ == "__main__": + try: + settings = ConnectorSettings() + helper = OpenCTIConnectorHelper(config=settings.to_helper_config()) + connector = RSTThreatLibrary(config=settings, helper=helper) + connector.run() + except Exception: + traceback.print_exc() + time.sleep(10) + sys.exit(1) diff --git a/external-import/rst-threat-library/src/requirements.txt b/external-import/rst-threat-library/src/requirements.txt new file mode 100644 index 0000000000..7843c64d8f --- /dev/null +++ b/external-import/rst-threat-library/src/requirements.txt @@ -0,0 +1,11 @@ +# Pin pycti to your OpenCTI platform version (OPENCTI_VERSION in root .env). +pycti==7.260706.0 +pydantic~=2.11.3 +pydantic-settings>=2.9.1,<3 +requests~=2.33.0 +validators==0.35.0 +stix2>=3.0.0,<4 +limits>=5.0,<6 + +# Dockerfile installs this line with --no-deps so it cannot override the pycti pin above. +connectors-sdk @ git+https://github.com/OpenCTI-Platform/connectors.git@master#subdirectory=connectors-sdk diff --git a/external-import/rst-threat-library/src/rst_threat_library_client/__init__.py b/external-import/rst-threat-library/src/rst_threat_library_client/__init__.py new file mode 100644 index 0000000000..3785a31808 --- /dev/null +++ b/external-import/rst-threat-library/src/rst_threat_library_client/__init__.py @@ -0,0 +1,3 @@ +from rst_threat_library_client.api_client import ThreatLibraryClient + +__all__ = ["ThreatLibraryClient"] diff --git a/external-import/rst-threat-library/src/rst_threat_library_client/api_client.py b/external-import/rst-threat-library/src/rst_threat_library_client/api_client.py new file mode 100644 index 0000000000..cd65117ab5 --- /dev/null +++ b/external-import/rst-threat-library/src/rst_threat_library_client/api_client.py @@ -0,0 +1,135 @@ +"""RST Cloud Threat Library HTTP client.""" + +from __future__ import annotations + +import time +from typing import Any, Dict, Iterator + +import requests +from pycti import OpenCTIConnectorHelper +from pydantic import HttpUrl + + +class ThreatLibraryClient: + """Paginated /v1/threat-objects/ reader with per-request retry.""" + + def __init__( + self, + helper: OpenCTIConnectorHelper, + *, + base_url: HttpUrl, + api_key: str, + auth_header: str = "x-api-key", + connect_timeout: int = 30, + read_timeout: int = 120, + retry: int = 2, + ssl_verify: bool = True, + page_size: int = 100, + order_by: str = "modified", + order_mode: str = "desc", + proxy: str = "", + ): + self.helper = helper + self.base_url = str(base_url).rstrip("/") + self.api_key = api_key + self.auth_header = auth_header or "x-api-key" + self.connect_timeout = int(connect_timeout) + self.read_timeout = int(read_timeout) + self.retry = int(retry) + self.ssl_verify = bool(ssl_verify) + self.page_size = int(page_size) + self.order_by = order_by or "modified" + self.order_mode = (order_mode or "desc").lower() + self.proxy = (proxy or "").strip() + + self._session = requests.Session() + self._session.headers.update( + { + self.auth_header: self.api_key, + "Accept": "application/json", + "User-Agent": "opencti-connector-rst-threat-library", + } + ) + if self.proxy: + self._session.proxies = {"http": self.proxy, "https": self.proxy} + + def iter_new_items(self, obj_type: str, cursor: str) -> Iterator[Dict[str, Any]]: + for item in self._iter_pages(obj_type, log_label="fetched page"): + modified = item.get("modified") or "" + if self.order_mode == "desc" and cursor and modified and modified <= cursor: + self.helper.connector_logger.info( + f"[{obj_type}] cursor reached at modified={modified}; stopping" + ) + return + if self.order_mode == "asc" and cursor and modified and modified <= cursor: + continue + yield item + + def iter_all_items(self, obj_type: str) -> Iterator[Dict[str, Any]]: + yield from self._iter_pages(obj_type, log_label="catalogue scan") + + def _iter_pages(self, obj_type: str, *, log_label: str) -> Iterator[Dict[str, Any]]: + url = f"{self.base_url}/threat-objects/{obj_type}" + offset = 0 + while True: + params = { + "limit": self.page_size, + "offset": offset, + "orderBy": self.order_by, + "orderMode": self.order_mode, + } + data = self._get_json(url, params) + items = (data or {}).get("data") or [] + total = (data or {}).get("total") + self.helper.connector_logger.info( + f"[{obj_type}] {log_label} offset={offset} size={len(items)}" + + (f" (total upstream={total})" if total is not None else "") + ) + if not items: + return + yield from items + if len(items) < self.page_size: + return + offset += self.page_size + + def _get_json(self, url: str, params: Dict[str, Any]) -> Dict[str, Any]: + last_exc = None + for attempt in range(self.retry + 1): + try: + response = self._session.get( + url, + params=params, + timeout=(self.connect_timeout, self.read_timeout), + verify=self.ssl_verify, + ) + if response.status_code == 429 or 500 <= response.status_code < 600: + raise requests.HTTPError( + f"{response.status_code} {response.reason} for {response.url}", + response=response, + ) + response.raise_for_status() + return response.json() + except Exception as exc: + last_exc = exc + if isinstance(exc, requests.HTTPError): + status = getattr( + getattr(exc, "response", None), "status_code", None + ) + if status is not None and 400 <= status < 500 and status != 429: + raise + if attempt < self.retry: + delay = (attempt + 1) * 2 + self.helper.connector_logger.warning( + "GET request failed; retrying", + { + "url": url, + "attempt": attempt + 1, + "max_attempts": self.retry + 1, + "error": str(exc), + "retry_in_seconds": delay, + }, + ) + time.sleep(delay) + else: + raise + raise last_exc # type: ignore[misc] diff --git a/external-import/rst-threat-library/tests/conftest.py b/external-import/rst-threat-library/tests/conftest.py new file mode 100644 index 0000000000..5ee8fc0e22 --- /dev/null +++ b/external-import/rst-threat-library/tests/conftest.py @@ -0,0 +1,4 @@ +import os +import sys + +sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src")) diff --git a/external-import/rst-threat-library/tests/test-requirements.txt b/external-import/rst-threat-library/tests/test-requirements.txt new file mode 100644 index 0000000000..9670511146 --- /dev/null +++ b/external-import/rst-threat-library/tests/test-requirements.txt @@ -0,0 +1,3 @@ +-r ../src/requirements.txt +pytest==9.0.3 + diff --git a/external-import/rst-threat-library/tests/test_connector/test_api_client.py b/external-import/rst-threat-library/tests/test_connector/test_api_client.py new file mode 100644 index 0000000000..27aabe2a57 --- /dev/null +++ b/external-import/rst-threat-library/tests/test_connector/test_api_client.py @@ -0,0 +1,114 @@ +from unittest.mock import MagicMock + +import pytest +import requests + +from rst_threat_library_client.api_client import ThreatLibraryClient + + +class _FakeResponse: + def __init__(self, status_code: int, payload: dict | None = None): + self.status_code = status_code + self.reason = "Error" + self.url = "http://test.com/threat-objects/malware" + self._payload = payload or {"data": []} + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(f"{self.status_code}", response=self) + + def json(self): + return self._payload + + +def test_get_json_retries_transient_http_errors(monkeypatch): + helper = MagicMock() + helper.connector_logger = MagicMock() + client = ThreatLibraryClient( + helper, + base_url="http://test.com/v1", + api_key="secret", + retry=2, + ) + + responses = [ + _FakeResponse(500), + _FakeResponse(200, {"data": [{"standard_id": "malware--1", "name": "x"}]}), + ] + client._session.get = MagicMock(side_effect=responses) + monkeypatch.setattr( + "rst_threat_library_client.api_client.time.sleep", lambda _: None + ) + + payload = client._get_json( + "http://test.com/v1/threat-objects/malware", {"limit": 1} + ) + + assert payload["data"][0]["standard_id"] == "malware--1" + assert client._session.get.call_count == 2 + + +def test_get_json_raises_after_retry_budget_exhausted(monkeypatch): + helper = MagicMock() + helper.connector_logger = MagicMock() + client = ThreatLibraryClient( + helper, + base_url="http://test.com/v1", + api_key="secret", + retry=1, + ) + client._session.get = MagicMock(return_value=_FakeResponse(500)) + monkeypatch.setattr( + "rst_threat_library_client.api_client.time.sleep", lambda _: None + ) + + with pytest.raises(requests.HTTPError): + client._get_json("http://test.com/v1/threat-objects/malware", {"limit": 1}) + + assert client._session.get.call_count == 2 + + +def test_get_json_does_not_retry_permanent_4xx(monkeypatch): + helper = MagicMock() + helper.connector_logger = MagicMock() + client = ThreatLibraryClient( + helper, + base_url="http://test.com/v1", + api_key="secret", + retry=2, + ) + client._session.get = MagicMock(return_value=_FakeResponse(401)) + sleep = MagicMock() + monkeypatch.setattr("rst_threat_library_client.api_client.time.sleep", sleep) + + with pytest.raises(requests.HTTPError): + client._get_json("http://test.com/v1/threat-objects/malware", {"limit": 1}) + + assert client._session.get.call_count == 1 + sleep.assert_not_called() + + +def test_get_json_retries_429(monkeypatch): + helper = MagicMock() + helper.connector_logger = MagicMock() + client = ThreatLibraryClient( + helper, + base_url="http://test.com/v1", + api_key="secret", + retry=2, + ) + responses = [ + _FakeResponse(429), + _FakeResponse(200, {"data": [{"standard_id": "malware--1", "name": "x"}]}), + ] + client._session.get = MagicMock(side_effect=responses) + monkeypatch.setattr( + "rst_threat_library_client.api_client.time.sleep", lambda _: None + ) + + payload = client._get_json( + "http://test.com/v1/threat-objects/malware", {"limit": 1} + ) + + assert payload["data"][0]["standard_id"] == "malware--1" + assert client._session.get.call_count == 2 diff --git a/external-import/rst-threat-library/tests/test_connector/test_confidence.py b/external-import/rst-threat-library/tests/test_connector/test_confidence.py new file mode 100644 index 0000000000..294327f2f1 --- /dev/null +++ b/external-import/rst-threat-library/tests/test_connector/test_confidence.py @@ -0,0 +1,32 @@ +from connector.confidence import ( + analyst_confidence_wins, + confidence_value, + make_sync_record, +) + + +def test_confidence_value_defaults_missing_to_zero(): + assert confidence_value({}) == 0 + assert confidence_value({"confidence": "42"}) == 42 + + +def test_analyst_confidence_wins_when_opencti_is_higher(): + assert analyst_confidence_wins( + {"confidence": 80}, + api_item={"confidence": 50}, + ) + + +def test_analyst_confidence_does_not_win_when_upstream_is_higher_or_equal(): + assert not analyst_confidence_wins( + {"confidence": 40}, + api_item={"confidence": 60}, + ) + assert not analyst_confidence_wins( + {"confidence": 60}, + api_item={"confidence": 60}, + ) + + +def test_make_sync_record_stores_upstream_confidence(): + assert make_sync_record({"confidence": 55}) == {"upstream_confidence": 55} diff --git a/external-import/rst-threat-library/tests/test_connector/test_connector_batch_send.py b/external-import/rst-threat-library/tests/test_connector/test_connector_batch_send.py new file mode 100644 index 0000000000..9b1d831e19 --- /dev/null +++ b/external-import/rst-threat-library/tests/test_connector/test_connector_batch_send.py @@ -0,0 +1,342 @@ +from unittest.mock import MagicMock + +import pytest + +from connector.connector import RSTThreatLibrary +from connector.settings import ConnectorSettings + + +class StubConnectorSettings(ConnectorSettings): + @classmethod + def _load_config_dict(cls, _, handler): + return handler( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": { + "id": "connector-id", + "name": "RST Threat Library", + "scope": "intrusion-set,malware,tool,campaign", + "log_level": "error", + "duration_period": "PT5M", + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + }, + } + ) + + +class StubConnectorSettingsWithConfidenceOverride(ConnectorSettings): + @classmethod + def _load_config_dict(cls, _, handler): + return handler( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": { + "id": "connector-id", + "name": "RST Threat Library", + "scope": "intrusion-set,malware,tool,campaign", + "log_level": "error", + "duration_period": "PT5M", + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + "intrusion_set_default_confidence": 80, + }, + } + ) + + +class StubConnectorSettingsWithConfidenceLock(ConnectorSettings): + @classmethod + def _load_config_dict(cls, _, handler): + return handler( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": { + "id": "connector-id", + "name": "RST Threat Library", + "scope": "intrusion-set,malware,tool,campaign", + "log_level": "error", + "duration_period": "PT5M", + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + "intrusion_set_default_confidence": 80, + "respect_user_edits": True, + }, + } + ) + + +@pytest.fixture +def connector(): + settings = StubConnectorSettings() + helper = MagicMock() + helper.connector_logger = MagicMock() + helper.connect_id = "connector-id" + helper.api.work.initiate_work.return_value = "work-1" + helper.stix2_create_bundle.return_value = '{"type":"bundle","objects":[]}' + helper.send_stix2_bundle.return_value = ["bundle-sent"] + return RSTThreatLibrary(config=settings, helper=helper) + + +def test_batch_send_stix_bundle_uses_helper_bundle_pattern(connector): + stix_object = MagicMock() + stix_object.serialize.return_value = '{"type":"malware","id":"malware--1"}' + + ok = connector._batch_send_stix_bundle( + [stix_object], timestamp=1_700_000_000, obj_type="malware" + ) + + assert ok is True + connector.helper.stix2_create_bundle.assert_called_once_with([stix_object]) + connector.helper.send_stix2_bundle.assert_called_once() + kwargs = connector.helper.send_stix2_bundle.call_args.kwargs + assert kwargs["cleanup_inconsistent_bundle"] is True + assert kwargs["work_id"] == "work-1" + + +def test_batch_send_stix_bundle_retries_transient_failures(connector, monkeypatch): + stix_object = MagicMock() + connector.helper.send_stix2_bundle.side_effect = [ + ConnectionError("temporary"), + ["bundle-sent"], + ] + monkeypatch.setattr("connector.connector.time.sleep", lambda _: None) + + ok = connector._batch_send_stix_bundle( + [stix_object], timestamp=1_700_000_000, obj_type="malware" + ) + + assert ok is True + assert connector.helper.send_stix2_bundle.call_count == 2 + + +def test_batch_send_stix_bundle_returns_false_after_retry_budget( + connector, monkeypatch +): + stix_object = MagicMock() + connector._max_retries = 2 + connector.helper.send_stix2_bundle.side_effect = ConnectionError("temporary") + monkeypatch.setattr("connector.connector.time.sleep", lambda _: None) + + ok = connector._batch_send_stix_bundle( + [stix_object], timestamp=1_700_000_000, obj_type="malware" + ) + + assert ok is False + assert connector.helper.send_stix2_bundle.call_count == 2 + assert connector.helper.api.work.to_processed.call_count == 2 + for call in connector.helper.api.work.to_processed.call_args_list: + assert call.kwargs.get("in_error") is True + + +def test_batch_send_stix_bundle_retries_requests_exceptions(connector, monkeypatch): + import requests + + stix_object = MagicMock() + connector.helper.send_stix2_bundle.side_effect = [ + requests.exceptions.Timeout("upstream timeout"), + ["bundle-sent"], + ] + monkeypatch.setattr("connector.connector.time.sleep", lambda _: None) + + ok = connector._batch_send_stix_bundle( + [stix_object], timestamp=1_700_000_000, obj_type="malware" + ) + + assert ok is True + assert connector.helper.send_stix2_bundle.call_count == 2 + assert connector.helper.api.work.to_processed.call_count == 2 + first = connector.helper.api.work.to_processed.call_args_list[0] + second = connector.helper.api.work.to_processed.call_args_list[1] + assert first.kwargs.get("in_error") is True + assert second.kwargs.get("in_error") is not True + + +def test_batch_send_stix_bundle_reraises_non_retryable_after_marking_work(connector): + stix_object = MagicMock() + connector.helper.send_stix2_bundle.side_effect = ValueError("bad payload") + + with pytest.raises(ValueError, match="bad payload"): + connector._batch_send_stix_bundle( + [stix_object], timestamp=1_700_000_000, obj_type="malware" + ) + + connector.helper.api.work.to_processed.assert_called_once() + kwargs = connector.helper.api.work.to_processed.call_args.kwargs + assert kwargs.get("in_error") is True + + +def test_batch_send_stix_bundle_performs_one_attempt_when_max_retries_zero(connector): + connector._max_retries = 0 + stix_object = MagicMock() + + ok = connector._batch_send_stix_bundle( + [stix_object], timestamp=1_700_000_000, obj_type="malware" + ) + + assert ok is True + assert connector.helper.send_stix2_bundle.call_count == 1 + + +def test_normalize_api_item_overrides_intrusion_set_confidence(): + settings = StubConnectorSettingsWithConfidenceOverride() + helper = MagicMock() + helper.connector_logger = MagicMock() + connector = RSTThreatLibrary(config=settings, helper=helper) + + item = {"standard_id": "intrusion-set--1", "name": "APT", "confidence": 96} + normalized = connector._normalize_api_item("intrusion-sets", item) + + assert normalized["confidence"] == 80 + assert item["confidence"] == 96 + + +def test_normalize_api_item_leaves_other_types_unchanged(): + settings = StubConnectorSettingsWithConfidenceOverride() + helper = MagicMock() + helper.connector_logger = MagicMock() + connector = RSTThreatLibrary(config=settings, helper=helper) + + item = {"standard_id": "malware--1", "name": "Evil", "confidence": 96} + normalized = connector._normalize_api_item("malware", item) + + assert normalized is item + assert normalized["confidence"] == 96 + + +def test_prepare_upsert_item_respects_confidence_override_for_analyst_lock(): + settings = StubConnectorSettingsWithConfidenceLock() + helper = MagicMock() + helper.connector_logger = MagicMock() + connector = RSTThreatLibrary(config=settings, helper=helper) + connector._read_opencti_entity = MagicMock( + return_value={ + "standard_id": "intrusion-set--1", + "confidence": 85, + } + ) + + prep = connector._prepare_upsert_item( + "intrusion-sets", + {"standard_id": "intrusion-set--1", "name": "APT", "confidence": 96}, + {}, + ) + + assert prep.skip is True + assert prep.api_item["confidence"] == 80 + + +def test_analyst_lock_uses_confidence_override_from_stored_state(): + settings = StubConnectorSettingsWithConfidenceLock() + helper = MagicMock() + helper.connector_logger = MagicMock() + connector = RSTThreatLibrary(config=settings, helper=helper) + state = { + "fingerprints": { + "intrusion-sets": { + "intrusion-set--1": {"upstream_confidence": 96}, + } + } + } + + locked = connector._analyst_locks_entity( + {"standard_id": "intrusion-set--1", "confidence": 85}, + obj_type_path="intrusion-sets", + state=state, + ) + + assert locked is True + + +def test_split_abandons_after_consecutive_analyst_lock_failures(): + from connector.connector import _SPLIT_FAILURE_SKIP_THRESHOLD + from connector.merge_split import SplitCandidate + + settings = StubConnectorSettingsWithConfidenceLock() + helper = MagicMock() + helper.connector_logger = MagicMock() + connector = RSTThreatLibrary(config=settings, helper=helper) + connector._analyst_locks_entity = MagicMock(return_value=True) + + oc = { + "id": "uuid-earth-lusca", + "standard_id": "intrusion-set--8f8886ee-5773-597c-b532-f57efafbaa02", + "name": "Earth Lusca", + "aliases": ["RedHotel"], + "confidence": 90, + } + split = SplitCandidate( + opencti_entity=oc, + keep_api_item={"standard_id": oc["standard_id"], "name": "Earth Lusca"}, + aliases_to_remove=["RedHotel"], + split_off_api_items=[], + ) + state: dict = {} + + for _ in range(_SPLIT_FAILURE_SKIP_THRESHOLD): + connector._execute_intrusion_set_split( + split, timestamp=1, obj_type="intrusion-sets", state=state + ) + + entry = state["split_failures"]["intrusion-sets"][oc["standard_id"]] + assert entry["skipped"] is True + assert entry["count"] == _SPLIT_FAILURE_SKIP_THRESHOLD + + helper.connector_logger.info.reset_mock() + connector._execute_intrusion_set_split( + split, timestamp=1, obj_type="intrusion-sets", state=state + ) + info_msgs = " ".join( + str(c.args[0]) for c in helper.connector_logger.info.call_args_list if c.args + ) + assert "analyst lock" not in info_msgs + assert connector._analyst_locks_entity.call_count == _SPLIT_FAILURE_SKIP_THRESHOLD + + +def test_wait_for_opencti_entity_retries_until_readable(monkeypatch): + settings = StubConnectorSettings() + helper = MagicMock() + helper.connector_logger = MagicMock() + connector = RSTThreatLibrary(config=settings, helper=helper) + + sid = "intrusion-set--36319194-19e1-50ac-9163-778b56a1bf12" + entity = {"id": "uuid-apt29", "standard_id": sid, "name": "APT29"} + reads = [None, None, entity] + connector._read_opencti_entity = MagicMock(side_effect=reads) + monkeypatch.setattr("connector.connector.time.sleep", lambda _: None) + + found = connector._wait_for_opencti_entity( + "intrusion-sets", + sid, + attempts=4, + delay_s=0.01, + context="merge", + ) + + assert found == entity + assert connector._read_opencti_entity.call_count == 3 + + +def test_wait_for_opencti_entity_returns_none_after_retries(monkeypatch): + settings = StubConnectorSettings() + helper = MagicMock() + helper.connector_logger = MagicMock() + connector = RSTThreatLibrary(config=settings, helper=helper) + connector._read_opencti_entity = MagicMock(return_value=None) + monkeypatch.setattr("connector.connector.time.sleep", lambda _: None) + + found = connector._wait_for_opencti_entity( + "intrusion-sets", + "intrusion-set--missing", + attempts=3, + delay_s=0.01, + ) + + assert found is None + assert connector._read_opencti_entity.call_count == 3 diff --git a/external-import/rst-threat-library/tests/test_connector/test_converter_to_stix.py b/external-import/rst-threat-library/tests/test_connector/test_converter_to_stix.py new file mode 100644 index 0000000000..1330ee474e --- /dev/null +++ b/external-import/rst-threat-library/tests/test_connector/test_converter_to_stix.py @@ -0,0 +1,88 @@ +import json +from unittest.mock import MagicMock + +import pytest + +from connector.converter_to_stix import ConverterToStix + + +@pytest.fixture +def converter(): + helper = MagicMock() + helper.connector_logger = MagicMock() + return ConverterToStix(helper=helper) + + +def test_item_to_sdo_builds_intrusion_set_with_upstream_id(converter): + item = { + "standard_id": "intrusion-set--c8d782e1-6566-4c2b-a9f8-87a757c379a4", + "entity_type": "Intrusion-Set", + "name": "APT Example", + "description": "Test intrusion set", + "confidence": 75, + "aliases": ["Example Group"], + "objectLabel": ["RST Threat Library"], + } + + sdo = converter.item_to_sdo(item, "intrusion-sets", ["RST Threat Library"]) + + assert sdo is not None + payload = json.loads(sdo.serialize()) + assert payload["id"] == item["standard_id"] + assert payload["name"] == "APT Example" + assert payload["aliases"] == ["Example Group"] + assert payload["confidence"] == 75 + assert "RST Threat Library" in payload["labels"] + + +def test_item_to_sdo_returns_none_when_standard_id_missing(converter): + item = {"entity_type": "Malware", "name": "No ID Malware"} + + assert converter.item_to_sdo(item, "malware", []) is None + converter.helper.connector_logger.warning.assert_called() + + +def test_build_identity_uses_upstream_standard_id(converter): + identity = converter.build_identity( + { + "standard_id": "identity--a1b2c3d4-e5f6-4789-a012-3456789abcde", + "name": "RST Cloud", + } + ) + + payload = json.loads(identity.serialize()) + assert payload["id"] == "identity--a1b2c3d4-e5f6-4789-a012-3456789abcde" + assert payload["name"] == "RST Cloud" + assert payload["identity_class"] == "organization" + + +def test_build_identity_honors_upstream_identity_class(converter): + identity = converter.build_identity( + { + "standard_id": "identity--a1b2c3d4-e5f6-4789-a012-3456789abcde", + "name": "Analyst", + "identity_class": "individual", + } + ) + + payload = json.loads(identity.serialize()) + assert payload["identity_class"] == "individual" + + +def test_build_external_references_skips_missing_source_name(converter): + refs = converter.build_external_references( + [ + {"url": "https://example.com/no-source"}, + { + "source_name": "RST Cloud", + "url": "https://example.com/ok", + "external_id": "abc", + }, + ] + ) + + assert len(refs) == 1 + payload = json.loads(refs[0].serialize()) + assert payload["source_name"] == "RST Cloud" + assert payload["url"] == "https://example.com/ok" + assert payload["external_id"] == "abc" diff --git a/external-import/rst-threat-library/tests/test_connector/test_merge_split.py b/external-import/rst-threat-library/tests/test_connector/test_merge_split.py new file mode 100644 index 0000000000..e4286a90df --- /dev/null +++ b/external-import/rst-threat-library/tests/test_connector/test_merge_split.py @@ -0,0 +1,106 @@ +from connector.merge_split import ( + analyze_intrusion_set_merge_split, + opencti_alias_count, + pick_opencti_merge_survivor, +) + + +def test_merge_split_detects_duplicate_opencti_entities_for_single_upstream_survivor(): + api_items = [ + { + "standard_id": "intrusion-set--aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "name": "Flax Typhoon", + "aliases": ["Earth Naga"], + } + ] + opencti_entities = [ + { + "standard_id": "intrusion-set--bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", + "name": "Earth Naga", + "aliases": [], + }, + { + "standard_id": "intrusion-set--aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "name": "Flax Typhoon", + "aliases": [], + }, + ] + + plan = analyze_intrusion_set_merge_split(api_items, opencti_entities) + + assert len(plan.merges) == 1 + assert plan.merges[0].target_api_item["standard_id"] == api_items[0]["standard_id"] + assert len(plan.merges[0].opencti_entities_to_merge) == 1 + + +def test_merge_split_detects_alias_conflict_for_split_candidate(): + api_items = [ + { + "standard_id": "intrusion-set--11111111-1111-4111-8111-111111111111", + "name": "Group A", + "aliases": [], + }, + { + "standard_id": "intrusion-set--22222222-2222-4222-8222-222222222222", + "name": "Group B", + "aliases": ["Shared Alias"], + }, + ] + opencti_entities = [ + { + "standard_id": "intrusion-set--11111111-1111-4111-8111-111111111111", + "name": "Group A", + "aliases": ["Shared Alias"], + } + ] + + plan = analyze_intrusion_set_merge_split(api_items, opencti_entities) + + assert len(plan.splits) == 1 + assert "Shared Alias" in plan.splits[0].aliases_to_remove + + +def test_pick_opencti_merge_survivor_prefers_more_aliases(): + """UNC3313-style internal names lose to established names with many aliases.""" + unc = { + "standard_id": "intrusion-set--unc3313-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "name": "UNC3313", + "aliases": ["unc3313_group"], + } + muddy = { + "standard_id": "intrusion-set--muddywa-bbbb-4bbb-8bbb-bbbbbbbbbbbb", + "name": "MuddyWater", + "aliases": [ + "Muddy Water", + "TEMP.Zagros", + "Static Kitten", + "Seedworm", + "COBALT MIRAGE", + ], + } + + survivor = pick_opencti_merge_survivor(unc["standard_id"], [unc, muddy]) + + assert survivor is not None + assert survivor["standard_id"] == muddy["standard_id"] + assert opencti_alias_count(muddy) > opencti_alias_count(unc) + + +def test_pick_opencti_merge_survivor_ties_break_to_api_standard_id(): + api_sid = "intrusion-set--aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + other_sid = "intrusion-set--bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + api_entity = { + "standard_id": api_sid, + "name": "Flax Typhoon", + "aliases": ["Earth Naga"], + } + other = { + "standard_id": other_sid, + "name": "Earth Naga", + "aliases": ["Flax Typhoon"], + } + + survivor = pick_opencti_merge_survivor(api_sid, [other, api_entity]) + + assert survivor is not None + assert survivor["standard_id"] == api_sid diff --git a/external-import/rst-threat-library/tests/test_connector/test_settings.py b/external-import/rst-threat-library/tests/test_connector/test_settings.py new file mode 100644 index 0000000000..386919e2f2 --- /dev/null +++ b/external-import/rst-threat-library/tests/test_connector/test_settings.py @@ -0,0 +1,163 @@ +from typing import Any + +import pytest +from connectors_sdk import ConfigValidationError + +from connector import ConnectorSettings + + +@pytest.mark.parametrize( + "settings_dict", + [ + pytest.param( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": { + "id": "connector-id", + "name": "RST Threat Library", + "scope": "test, connector", + "log_level": "error", + "duration_period": "PT5M", + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + }, + }, + id="full_valid_settings_dict", + ), + pytest.param( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": {"id": "connector-id", "scope": "test, connector"}, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + }, + }, + id="minimal_valid_settings_dict", + ), + ], +) +def test_settings_should_accept_valid_input(settings_dict: dict[str, Any]): + class FakeConnectorSettings(ConnectorSettings): + @classmethod + def _load_config_dict(cls, _, handler) -> dict[str, Any]: + return handler(settings_dict) + + settings = FakeConnectorSettings() + + assert settings.opencti is not None + assert settings.connector is not None + assert settings.rst_threat_library is not None + + +def test_settings_passes_auto_create_service_account_to_helper_config(): + class FakeConnectorSettings(ConnectorSettings): + @classmethod + def _load_config_dict(cls, _, handler) -> dict[str, Any]: + return handler( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": { + "id": "connector-id", + "name": "RST Threat Library", + "scope": "intrusion-set", + "log_level": "error", + "duration_period": "PT5M", + "auto_create_service_account": True, + "auto_create_service_account_confidence_level": 80, + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + }, + } + ) + + settings = FakeConnectorSettings() + helper_config = settings.to_helper_config() + + assert helper_config["connector"]["auto_create_service_account"] is True + assert ( + helper_config["connector"]["auto_create_service_account_confidence_level"] + == 80 + ) + + +@pytest.mark.parametrize( + "settings_dict, field_name", + [ + pytest.param( + {}, + "settings", + id="empty_settings_dict", + ), + pytest.param( + { + "opencti": {"url": "http://localhost:PORT", "token": "test-token"}, + "connector": { + "id": "connector-id", + "name": "RST Threat Library", + "scope": "test, connector", + "log_level": "error", + "duration_period": "PT5M", + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + }, + }, + "opencti.url", + id="invalid_opencti_url", + ), + pytest.param( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": {"name": "RST Threat Library", "scope": "test, connector"}, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + }, + }, + "connector.id", + id="missing_connector_id", + ), + ], +) +def test_settings_should_raise_when_invalid_input( + settings_dict: dict[str, Any], + field_name: str, +): + class FakeConnectorSettings(ConnectorSettings): + @classmethod + def _load_config_dict(cls, _, handler) -> dict[str, Any]: + return handler(settings_dict) + + with pytest.raises(ConfigValidationError) as err: + FakeConnectorSettings() + + assert "Error validating configuration" in str(err.value) + + +def test_settings_rejects_negative_retry(): + class FakeConnectorSettings(ConnectorSettings): + @classmethod + def _load_config_dict(cls, _, handler) -> dict[str, Any]: + return handler( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": { + "id": "connector-id", + "scope": "intrusion-set", + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + "retry": -1, + }, + } + ) + + with pytest.raises(ConfigValidationError): + FakeConnectorSettings() diff --git a/external-import/rst-threat-library/tests/test_main.py b/external-import/rst-threat-library/tests/test_main.py new file mode 100644 index 0000000000..ef1cc2178f --- /dev/null +++ b/external-import/rst-threat-library/tests/test_main.py @@ -0,0 +1,68 @@ +from typing import Any +from unittest.mock import MagicMock + +import pytest +from pycti import OpenCTIConnectorHelper + +from connector import ConnectorSettings +from main import RSTThreatLibrary + + +@pytest.fixture +def mock_opencti_connector_helper(monkeypatch): + """Mock heavy dependencies of OpenCTIConnectorHelper (avoid OpenCTI calls).""" + + module_import_path = "pycti.connector.opencti_connector_helper" + monkeypatch.setattr(f"{module_import_path}.killProgramHook", MagicMock()) + monkeypatch.setattr(f"{module_import_path}.sched.scheduler", MagicMock()) + monkeypatch.setattr(f"{module_import_path}.ConnectorInfo", MagicMock()) + monkeypatch.setattr(f"{module_import_path}.OpenCTIApiClient", MagicMock()) + monkeypatch.setattr(f"{module_import_path}.OpenCTIConnector", MagicMock()) + monkeypatch.setattr(f"{module_import_path}.OpenCTIMetricHandler", MagicMock()) + monkeypatch.setattr(f"{module_import_path}.PingAlive", MagicMock()) + + +class StubConnectorSettings(ConnectorSettings): + """Stub ConnectorSettings with a valid in-memory config dict.""" + + @classmethod + def _load_config_dict(cls, _, handler) -> dict[str, Any]: + return handler( + { + "opencti": { + "url": "http://localhost:8080", + "token": "test-token", + }, + "connector": { + "id": "connector-id", + "name": "Test Connector", + "scope": "test, connector", + "log_level": "error", + "duration_period": "PT5M", + }, + "rst_threat_library": { + "baseurl": "http://test.com", + "apikey": "test-api-key", + "auth_header": "x-api-key", + }, + } + ) + + +def test_connector_settings_is_instantiated(): + settings = StubConnectorSettings() + assert isinstance(settings.to_helper_config(), dict) + + +def test_opencti_connector_helper_is_instantiated(mock_opencti_connector_helper): + settings = StubConnectorSettings() + helper = OpenCTIConnectorHelper(config=settings.to_helper_config()) + assert helper.opencti_url is not None + + +def test_connector_is_instantiated(mock_opencti_connector_helper): + settings = StubConnectorSettings() + helper = OpenCTIConnectorHelper(config=settings.to_helper_config()) + connector = RSTThreatLibrary(config=settings, helper=helper) + assert connector.config == settings + assert connector.helper == helper