diff --git a/internal-enrichment/xposedornot/.build.env b/internal-enrichment/xposedornot/.build.env new file mode 100644 index 00000000000..90e99f7106a --- /dev/null +++ b/internal-enrichment/xposedornot/.build.env @@ -0,0 +1,2 @@ +CONNECTOR_CMD="-m src" +CONNECTOR_WORKDIR="/opt/connector" diff --git a/internal-enrichment/xposedornot/.dockerignore b/internal-enrichment/xposedornot/.dockerignore new file mode 100644 index 00000000000..4cca3bac802 --- /dev/null +++ b/internal-enrichment/xposedornot/.dockerignore @@ -0,0 +1,10 @@ +**/logs +**/*.gql +**/venv +**/.venv +**/__pycache__/ +**/*.egg-info/ +**/config.yml +**/__pycache__ +**/__metadata__ +**/__docs__ diff --git a/internal-enrichment/xposedornot/.env.sample b/internal-enrichment/xposedornot/.env.sample new file mode 100644 index 00000000000..887f23538d6 --- /dev/null +++ b/internal-enrichment/xposedornot/.env.sample @@ -0,0 +1,13 @@ +OPENCTI_URL=http://opencti:8080 +OPENCTI_TOKEN=ChangeMe +CONNECTOR_ID=ChangeMe +CONNECTOR_TYPE=INTERNAL_ENRICHMENT +CONNECTOR_NAME=XposedOrNot +CONNECTOR_SCOPE=Email-Addr +CONNECTOR_LOG_LEVEL=info +CONNECTOR_AUTO=false +# Optional; leave unset to use the free keyless community API +XPOSEDORNOT_API_KEY= +XPOSEDORNOT_API_BASE_URL=https://api.xposedornot.com +XPOSEDORNOT_MAX_TLP=TLP:AMBER +XPOSEDORNOT_TLP_LEVEL=amber diff --git a/internal-enrichment/xposedornot/.gitignore b/internal-enrichment/xposedornot/.gitignore new file mode 100644 index 00000000000..6e221ed5069 --- /dev/null +++ b/internal-enrichment/xposedornot/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.pytest_cache/ +.venv/ +venv/ +config.yml +.env diff --git a/internal-enrichment/xposedornot/Dockerfile b/internal-enrichment/xposedornot/Dockerfile new file mode 100644 index 00000000000..1906cdfbb7e --- /dev/null +++ b/internal-enrichment/xposedornot/Dockerfile @@ -0,0 +1,16 @@ +FROM python:3.12-alpine +ENV CONNECTOR_TYPE=INTERNAL_ENRICHMENT + +# Copy the connector +COPY src /opt/connector/src + +WORKDIR /opt/connector + +# Install Python modules +# hadolint ignore=DL3003 +RUN apk --no-cache add git build-base libmagic libffi-dev && \ + cd /opt/connector/src && \ + pip3 install --no-cache-dir -r /opt/connector/src/requirements.txt && \ + apk del git build-base + +CMD ["python", "-m", "src"] diff --git a/internal-enrichment/xposedornot/README.md b/internal-enrichment/xposedornot/README.md new file mode 100644 index 00000000000..f82961f546b --- /dev/null +++ b/internal-enrichment/xposedornot/README.md @@ -0,0 +1,96 @@ +# OpenCTI XposedOrNot Connector + + + +Table of Contents + +- [OpenCTI XposedOrNot Connector](#opencti-xposedornot-connector) + - [Introduction](#introduction) + - [Installation](#installation) + - [Requirements](#requirements) + - [Configuration variables](#configuration-variables) + - [Deployment](#deployment) + - [Docker Deployment](#docker-deployment) + - [Manual Deployment](#manual-deployment) + - [Usage](#usage) + - [Behavior](#behavior) + - [Legal and privacy notice](#legal-and-privacy-notice) + - [Debugging](#debugging) + +## Introduction + +[XposedOrNot](https://xposedornot.com) is an open, free data-breach search service tracking 760+ known breaches. Given an email address it returns the breaches the address appears in, with per-breach detail: breach date, records exposed, exposed data classes, affected domain, industry and password-storage risk, plus an overall risk score. + +This internal-enrichment connector enriches `Email-Addr` observables with that exposure data. **No API key or registration is required** — the free community API is used by default. An optional commercial key ([console.xposedornot.com](https://console.xposedornot.com)) switches the connector to the Plus API with higher rate limits. + +## Installation + +### Requirements + +- OpenCTI Platform >= 6.8.12 +- No XposedOrNot account or API key needed (optional key for higher volume) + +## Configuration variables + +Full configuration reference: [`__metadata__/CONNECTOR_CONFIG_DOC.md`](__metadata__/CONNECTOR_CONFIG_DOC.md). + +Key parameters: + +| Parameter | Docker env var | Mandatory | Default | Description | +|-----------|----------------|-----------|---------|-------------| +| OpenCTI URL | `OPENCTI_URL` | Yes | | The OpenCTI platform URL. | +| OpenCTI Token | `OPENCTI_TOKEN` | Yes | | Token of the connector user. | +| Connector Auto | `CONNECTOR_AUTO` | No | `false` | Automatic enrichment. The keyless API allows 2 req/s and 25/hour per IP — keep manual, or configure an API key before enabling on busy platforms. | +| API key | `XPOSEDORNOT_API_KEY` | No | *(empty)* | Optional key; switches to the Plus API with higher limits. Fully functional without it. | +| Base URL | `XPOSEDORNOT_API_BASE_URL` | No | `https://api.xposedornot.com` | Free community API base URL. | +| Max TLP | `XPOSEDORNOT_MAX_TLP` | No | `TLP:AMBER` | Maximum TLP of an observable the connector may enrich (the email address is sent to the API). | +| TLP level | `XPOSEDORNOT_TLP_LEVEL` | No | `amber` | TLP marking applied to the produced objects. | + +## Deployment + +### Docker Deployment + +Use the provided `docker-compose.yml` (or add the service to your OpenCTI stack): + +```shell +docker compose up -d +``` + +### Manual Deployment + +```shell +# From the connector root directory (internal-enrichment/xposedornot): +pip3 install -r src/requirements.txt +cp config.yml.sample config.yml # then edit config.yml +python3 -m src +``` + +## Usage + +On an `Email-Addr` observable, click the enrichment button and select the XposedOrNot connector (or set `CONNECTOR_AUTO=true` — mind the rate limits above). The connector is playbook-compatible. + +## Behavior + +For a breached email address, the connector enriches in place — no extra entities are created, keeping the graph clean: + +- the observable's **score** is set from the XposedOrNot risk score (0–100, free API); +- **labels** `data-breach` — and `plaintext-password-exposure` when at least one breach stored passwords in plaintext — are added to the observable; +- an **external reference** to xposedornot.com is attached; +- a markdown **Note** is attached with the full breach table: breach name, date, records exposed, affected domain, exposed data classes, password-storage risk and verification status, plus first/latest exposure years and totals. + +A clean email (not found in any breach) completes with an explicit "no known breach exposure" message and modifies nothing. Rate limiting (HTTP 429) is retried with backoff honoring `Retry-After`; persistent rate limiting fails that single enrichment with a log message recommending the optional key — the connector itself keeps running. + +## Legal and privacy notice + +Only the observable's email address is sent, over TLS, to xposedornot.com — nothing else leaves the platform. Breach exposure tied to an email address is **personal information**: gate what may be enriched with `XPOSEDORNOT_MAX_TLP`, apply a restrictive `XPOSEDORNOT_TLP_LEVEL` to results, and use them only within lawful, authorised investigations (GDPR / legal basis / proper investigative framework). See the [XposedOrNot privacy policy](https://xposedornot.com/privacy). + +## Debugging + +Set `CONNECTOR_LOG_LEVEL=debug`. All API errors are logged through the connector logger with masked context; the API key never appears in logs. Typical messages: + +- `XposedOrNot rate limited (keyless: 2/s, 25/hour); backing off.` — expected under keyless bursts; configure a key for volume. +- `XposedOrNot: API key rejected or missing for the Plus API` — check `XPOSEDORNOT_API_KEY`. diff --git a/internal-enrichment/xposedornot/__metadata__/CONNECTOR_CONFIG_DOC.md b/internal-enrichment/xposedornot/__metadata__/CONNECTOR_CONFIG_DOC.md new file mode 100644 index 00000000000..532b6ffbe20 --- /dev/null +++ b/internal-enrichment/xposedornot/__metadata__/CONNECTOR_CONFIG_DOC.md @@ -0,0 +1,19 @@ +# Connector Configurations + +Below is an exhaustive enumeration of all configurable parameters available, each accompanied by detailed explanations of their purposes, default behaviors, and usage guidelines to help you understand and utilize them effectively. + +### Type: `object` + +| Property | Type | Required | Possible values | Default | Description | +| -------- | ---- | -------- | --------------- | ------- | ----------- | +| OPENCTI_URL | `string` | ✅ | Format: [`uri`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | | The OpenCTI platform URL. | +| OPENCTI_TOKEN | `string` | ✅ | string | | The token of the user who represents the connector in the OpenCTI platform. | +| CONNECTOR_NAME | `string` | | string | `"XposedOrNot"` | Name of the connector. | +| CONNECTOR_SCOPE | `array` | | string | `["Email-Addr"]` | The scope or type of data the connector is importing, either a MIME type or Stix Object (for information only). | +| CONNECTOR_TYPE | `string` | | string | `"INTERNAL_ENRICHMENT"` | Should always be set to INTERNAL_ENRICHMENT for this connector. | +| CONNECTOR_LOG_LEVEL | `string` | | `debug` `info` `warn` `warning` `error` | `"error"` | Determines the verbosity of the logs. | +| CONNECTOR_AUTO | `boolean` | | boolean | `false` | Enables or disables automatic enrichment of observables. The keyless community API is rate limited (2/s, 25/hour per IP); keep disabled or configure an API key before enabling on busy platforms. | +| XPOSEDORNOT_API_KEY | `string` | | Format: [`password`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | | Optional XposedOrNot API key (console.xposedornot.com). When set, the connector uses the commercial Plus API with higher rate limits. The connector is fully functional without it. | +| XPOSEDORNOT_API_BASE_URL | `string` | | Format: [`uri`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | `"https://api.xposedornot.com"` | Base URL of the free XposedOrNot community API. | +| XPOSEDORNOT_MAX_TLP | `string` | | `TLP:CLEAR` `TLP:WHITE` `TLP:GREEN` `TLP:AMBER` `TLP:AMBER+STRICT` `TLP:RED` | `"TLP:AMBER"` | Maximum TLP of an observable the connector is allowed to enrich. The observable's email address is sent to the XposedOrNot API. | +| XPOSEDORNOT_TLP_LEVEL | `string` | | `clear` `green` `amber` `amber+strict` `red` | `"amber"` | Traffic Light Protocol (TLP) level applied to the objects imported into OpenCTI. Results contain personal data; a restrictive TLP is recommended. | diff --git a/internal-enrichment/xposedornot/__metadata__/connector_config_schema.json b/internal-enrichment/xposedornot/__metadata__/connector_config_schema.json new file mode 100644 index 00000000000..9f078b00aae --- /dev/null +++ b/internal-enrichment/xposedornot/__metadata__/connector_config_schema.json @@ -0,0 +1,96 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://www.filigran.io/connectors/xposedornot_config.schema.json", + "type": "object", + "properties": { + "OPENCTI_URL": { + "description": "The OpenCTI platform URL.", + "format": "uri", + "maxLength": 2083, + "minLength": 1, + "type": "string" + }, + "OPENCTI_TOKEN": { + "description": "The token of the user who represents the connector in the OpenCTI platform.", + "type": "string" + }, + "CONNECTOR_NAME": { + "default": "XposedOrNot", + "description": "Name of the connector.", + "type": "string" + }, + "CONNECTOR_SCOPE": { + "default": [ + "Email-Addr" + ], + "description": "The scope or type of data the connector is importing, either a MIME type or Stix Object (for information only).", + "items": { + "type": "string" + }, + "type": "array" + }, + "CONNECTOR_TYPE": { + "default": "INTERNAL_ENRICHMENT", + "description": "Should always be set to INTERNAL_ENRICHMENT for this connector.", + "type": "string" + }, + "CONNECTOR_LOG_LEVEL": { + "default": "error", + "description": "Determines the verbosity of the logs.", + "enum": [ + "debug", + "info", + "warn", + "warning", + "error" + ], + "type": "string" + }, + "CONNECTOR_AUTO": { + "default": false, + "description": "Enables or disables automatic enrichment of observables. The keyless community API is rate limited (2/s, 25/hour per IP); keep disabled or configure an API key before enabling on busy platforms.", + "type": "boolean" + }, + "XPOSEDORNOT_API_KEY": { + "description": "Optional XposedOrNot API key (console.xposedornot.com). When set, the connector uses the commercial Plus API with higher rate limits. The connector is fully functional without it.", + "format": "password", + "type": "string" + }, + "XPOSEDORNOT_API_BASE_URL": { + "default": "https://api.xposedornot.com", + "description": "Base URL of the free XposedOrNot community API.", + "format": "uri", + "type": "string" + }, + "XPOSEDORNOT_MAX_TLP": { + "default": "TLP:AMBER", + "description": "Maximum TLP of an observable the connector is allowed to enrich. The observable's email address is sent to the XposedOrNot API.", + "enum": [ + "TLP:CLEAR", + "TLP:WHITE", + "TLP:GREEN", + "TLP:AMBER", + "TLP:AMBER+STRICT", + "TLP:RED" + ], + "type": "string" + }, + "XPOSEDORNOT_TLP_LEVEL": { + "default": "amber", + "description": "Traffic Light Protocol (TLP) level applied to the objects imported into OpenCTI. Results contain personal data; a restrictive TLP is recommended.", + "enum": [ + "clear", + "green", + "amber", + "amber+strict", + "red" + ], + "type": "string" + } + }, + "required": [ + "OPENCTI_URL", + "OPENCTI_TOKEN" + ], + "additionalProperties": true +} diff --git a/internal-enrichment/xposedornot/__metadata__/connector_manifest.json b/internal-enrichment/xposedornot/__metadata__/connector_manifest.json new file mode 100644 index 00000000000..469c5e9c83e --- /dev/null +++ b/internal-enrichment/xposedornot/__metadata__/connector_manifest.json @@ -0,0 +1,28 @@ +{ + "title": "XposedOrNot", + "slug": "xposedornot", + "description": "XposedOrNot is an open, free data-breach search service tracking 760+ breaches. Given an email address, it returns the breaches the address appears in, with per-breach detail: breach date, records exposed, exposed data classes, affected domain, industry and password-storage risk, plus an overall risk score.\n\nIntegrating XposedOrNot with OpenCTI lets analysts enrich an Email-Addr observable in place: the observable receives a risk-derived score, `data-breach` labels (plus a `plaintext-password-exposure` label when applicable) and an external reference, and a markdown summary note with the full breach table is attached. The connector works without any API key via the free community API (rate limits apply); an optional commercial key raises limits. Breach exposure tied to an email address is personal information and must only be used for lawful, authorised investigations (GDPR / legal basis / proper investigative framework).", + "short_description": "Enrich an email address with its data-breach exposure — breach names, dates, records, exposed data classes and password-storage risk — score and label the observable, and attach a summary note. No API key required.", + "logo": "internal-enrichment/xposedornot/__metadata__/logo.png", + "use_cases": [ + "Brand, Digital Risk & Underground Exposure", + "Detection & Response Enablement" + ], + "solution_categories": [ + "Enrichment & Reputation", + "Digital Risk Protection" + ], + "license_type": "Free", + "contact": null, + "verified": false, + "last_verified_date": null, + "playbook_supported": true, + "max_confidence_level": 75, + "support_version": ">= 6.8.12", + "subscription_link": "https://console.xposedornot.com", + "source_code": "https://github.com/OpenCTI-Platform/connectors/tree/master/internal-enrichment/xposedornot", + "manager_supported": false, + "container_version": "rolling", + "container_image": "opencti/connector-xposedornot", + "container_type": "INTERNAL_ENRICHMENT" +} diff --git a/internal-enrichment/xposedornot/__metadata__/logo.png b/internal-enrichment/xposedornot/__metadata__/logo.png new file mode 100644 index 00000000000..d8662b9c2dc Binary files /dev/null and b/internal-enrichment/xposedornot/__metadata__/logo.png differ diff --git a/internal-enrichment/xposedornot/config.yml.sample b/internal-enrichment/xposedornot/config.yml.sample new file mode 100644 index 00000000000..e0e7998d110 --- /dev/null +++ b/internal-enrichment/xposedornot/config.yml.sample @@ -0,0 +1,17 @@ +opencti: + url: 'http://localhost:8080' + token: 'ChangeMe' + +connector: + type: 'INTERNAL_ENRICHMENT' + id: 'ChangeMe' + name: 'XposedOrNot' # optional (default: 'XposedOrNot') + scope: 'Email-Addr' # optional (default: 'Email-Addr') + log_level: 'info' # optional (default: 'error') + auto: false # optional, auto-enrich observables. Keyless API is rate limited; keep false or set an api_key (default: false) + +xposedornot: + api_key: 'ChangeMe' # optional; remove/leave empty to use the free keyless community API + api_base_url: 'https://api.xposedornot.com' # optional (default: community API) + max_tlp: 'TLP:AMBER' # optional, max TLP of observables the connector may enrich (default: 'TLP:AMBER') + tlp_level: 'amber' # optional, TLP applied to produced objects: clear, green, amber, amber+strict, red (default: 'amber') diff --git a/internal-enrichment/xposedornot/docker-compose.yml b/internal-enrichment/xposedornot/docker-compose.yml new file mode 100644 index 00000000000..866f748f422 --- /dev/null +++ b/internal-enrichment/xposedornot/docker-compose.yml @@ -0,0 +1,17 @@ +version: '3' +services: + connector-xposedornot: + image: opencti/connector-xposedornot:latest + environment: + - OPENCTI_URL=http://localhost + - OPENCTI_TOKEN=ChangeMe + - CONNECTOR_ID=ChangeMe + - CONNECTOR_NAME=XposedOrNot + - CONNECTOR_SCOPE=Email-Addr + - CONNECTOR_AUTO=false # keyless API is rate limited (25/hour); enable with care or configure an API key + - CONNECTOR_LOG_LEVEL=error + - XPOSEDORNOT_API_KEY= # optional; empty uses the free keyless community API + - XPOSEDORNOT_API_BASE_URL=https://api.xposedornot.com + - XPOSEDORNOT_MAX_TLP=TLP:AMBER # maximum TLP of observables the connector may enrich + - XPOSEDORNOT_TLP_LEVEL=amber # clear, green, amber, amber+strict, red + restart: always diff --git a/internal-enrichment/xposedornot/pyproject.toml b/internal-enrichment/xposedornot/pyproject.toml new file mode 100644 index 00000000000..da606462f45 --- /dev/null +++ b/internal-enrichment/xposedornot/pyproject.toml @@ -0,0 +1,3 @@ +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src"] diff --git a/internal-enrichment/xposedornot/src/__init__.py b/internal-enrichment/xposedornot/src/__init__.py new file mode 100644 index 00000000000..40a96afc6ff --- /dev/null +++ b/internal-enrichment/xposedornot/src/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- diff --git a/internal-enrichment/xposedornot/src/__main__.py b/internal-enrichment/xposedornot/src/__main__.py new file mode 100644 index 00000000000..93726f4c285 --- /dev/null +++ b/internal-enrichment/xposedornot/src/__main__.py @@ -0,0 +1,21 @@ +# -*- coding: utf-8 -*- +import sys +import time +import traceback + +from pycti import OpenCTIConnectorHelper + +from .xposedornot import ConnectorSettings, XposedOrNotConnector + +if __name__ == "__main__": + try: + settings = ConnectorSettings() + helper = OpenCTIConnectorHelper( + config=settings.to_helper_config(), + playbook_compatible=True, + ) + XposedOrNotConnector(config=settings, helper=helper).run() + except Exception: + traceback.print_exc() + time.sleep(10) + sys.exit(1) diff --git a/internal-enrichment/xposedornot/src/requirements.txt b/internal-enrichment/xposedornot/src/requirements.txt new file mode 100644 index 00000000000..29fa5c3ea0b --- /dev/null +++ b/internal-enrichment/xposedornot/src/requirements.txt @@ -0,0 +1,4 @@ +pycti==7.260722.0 +pydantic~=2.11.3 +requests>=2.32,<3.0.0 +connectors-sdk @ git+https://github.com/OpenCTI-Platform/connectors.git@master#subdirectory=connectors-sdk diff --git a/internal-enrichment/xposedornot/src/xposedornot/__init__.py b/internal-enrichment/xposedornot/src/xposedornot/__init__.py new file mode 100644 index 00000000000..516893d1f82 --- /dev/null +++ b/internal-enrichment/xposedornot/src/xposedornot/__init__.py @@ -0,0 +1,5 @@ +# -*- coding: utf-8 -*- +from .connector import XposedOrNotConnector +from .settings import ConnectorSettings + +__all__ = ["XposedOrNotConnector", "ConnectorSettings"] diff --git a/internal-enrichment/xposedornot/src/xposedornot/client_api.py b/internal-enrichment/xposedornot/src/xposedornot/client_api.py new file mode 100644 index 00000000000..165a08dae95 --- /dev/null +++ b/internal-enrichment/xposedornot/src/xposedornot/client_api.py @@ -0,0 +1,179 @@ +# -*- coding: utf-8 -*- +"""Client for the XposedOrNot data-breach API. + +Community API (default, no key): + GET https://api.xposedornot.com/v1/breach-analytics?email= +Plus API (used automatically when an API key is configured): + GET https://plus-api.xposedornot.com/v3/check-email/?detailed=true + Auth: header `x-api-key: ` + +Both responses are normalised to a single shape: + {"breaches": [, ...], "risk_label": str|None, "risk_score": int|None} +A clean result (email not found in any breach) is returned as {} — this is a +normal outcome, not an error. Errors return None (and are logged). +""" + +from __future__ import annotations + +import time +from urllib.parse import quote + +import requests + +FREE_BASE_URL = "https://api.xposedornot.com" +PLUS_BASE_URL = "https://plus-api.xposedornot.com" +USER_AGENT = "opencti-xposedornot-connector/1.0 (+https://github.com/XposedOrNot)" + + +def _to_int(value): + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _split_data_classes(value) -> list[str]: + return [item.strip() for item in str(value or "").split(";") if item.strip()] + + +def _normalise_free(data: dict) -> dict: + breaches = [] + exposed = data.get("ExposedBreaches") or {} + for entry in exposed.get("breaches_details") or []: + breaches.append( + { + "name": entry.get("breach"), + "date": entry.get("xposed_date"), + "records": _to_int(entry.get("xposed_records")), + "domain": entry.get("domain"), + "industry": entry.get("industry"), + "password_risk": entry.get("password_risk"), + "verified": entry.get("verified"), + "data_classes": _split_data_classes(entry.get("xposed_data")), + } + ) + if not breaches: + return {} + risk = (data.get("BreachMetrics") or {}).get("risk") or [] + risk_label = ( + risk[0].get("risk_label") if risk and isinstance(risk[0], dict) else None + ) + risk_score = ( + _to_int(risk[0].get("risk_score")) + if risk and isinstance(risk[0], dict) + else None + ) + return {"breaches": breaches, "risk_label": risk_label, "risk_score": risk_score} + + +def _normalise_plus(data: dict) -> dict: + breaches = [] + for entry in data.get("breaches") or []: + breaches.append( + { + "name": entry.get("breach_id"), + "date": entry.get("breached_date"), + "records": _to_int(entry.get("xposed_records")), + "domain": entry.get("domain"), + "industry": entry.get("industry"), + "password_risk": entry.get("password_risk"), + "verified": entry.get("verified"), + "data_classes": _split_data_classes(entry.get("xposed_data")), + } + ) + if not breaches: + return {} + return {"breaches": breaches, "risk_label": None, "risk_score": None} + + +class XposedOrNotClient: + def __init__( + self, + helper, + api_key: str | None = None, + base_url: str | None = None, + timeout: int = 30, + ): + self.helper = helper + self.api_key = (api_key or "").strip() or None + self.base_url = (base_url or FREE_BASE_URL).rstrip("/") + self.timeout = timeout + self.session = requests.Session() + self.session.headers.update( + {"Accept": "application/json", "User-Agent": USER_AGENT} + ) + if self.api_key: + self.session.headers.update({"x-api-key": self.api_key}) + + def lookup(self, email: str) -> dict | None: + """Look up breach exposure for an email. + + Returns the normalised result dict, {} for a clean email, or None on error. + """ + if self.api_key: + url = "%s/v3/check-email/%s" % (PLUS_BASE_URL, quote(email, safe="")) + params = {"detailed": "true"} + else: + url = "%s/v1/breach-analytics" % self.base_url + params = {"email": email} + + max_retries = 3 + for attempt in range(1, max_retries + 1): + try: + resp = self.session.get(url, params=params, timeout=self.timeout) + except requests.RequestException as exc: + self.helper.connector_logger.error( + "XposedOrNot request failed", meta={"error": str(exc)} + ) + return None + + if resp.status_code == 404: + # email not found in any breach -- normal, clean outcome + return {} + + if resp.status_code == 429: + if self.api_key: + message = "XposedOrNot Plus API rate limited; backing off." + else: + message = ( + "XposedOrNot rate limited (keyless: 2/s, 25/hour); backing" + " off. An optional API key raises limits." + ) + self.helper.connector_logger.warning(message, meta={"attempt": attempt}) + if attempt < max_retries: + retry_after = resp.headers.get("Retry-After") + wait = ( + int(retry_after) + if retry_after and retry_after.isdigit() + else 15 + ) + time.sleep(min(wait, 60)) + continue + + if resp.status_code in (401, 403, 422): + self.helper.connector_logger.error( + "XposedOrNot: API key rejected or missing for the Plus API", + meta={"status": resp.status_code}, + ) + return None + if resp.status_code >= 400: + self.helper.connector_logger.error( + "XposedOrNot: error response", + meta={"status": resp.status_code, "body": resp.text[:300]}, + ) + return None + + try: + data = resp.json() + except ValueError: + self.helper.connector_logger.error( + "XposedOrNot: invalid JSON in response." + ) + return None + + return _normalise_plus(data) if self.api_key else _normalise_free(data) + + self.helper.connector_logger.error( + "XposedOrNot: still rate limited after retries." + ) + return None diff --git a/internal-enrichment/xposedornot/src/xposedornot/connector.py b/internal-enrichment/xposedornot/src/xposedornot/connector.py new file mode 100644 index 00000000000..03e22fff592 --- /dev/null +++ b/internal-enrichment/xposedornot/src/xposedornot/connector.py @@ -0,0 +1,162 @@ +# -*- coding: utf-8 -*- +"""OpenCTI internal-enrichment connector for XposedOrNot. + +Enriches Email-Addr observables with data-breach exposure. Works without any +API key via the free community API; an optional key switches to the Plus API. + +Privacy note: the observable's email address (personal information) is sent +over TLS to xposedornot.com. Gate what may leave the platform with +XPOSEDORNOT_MAX_TLP, and use the results only within lawful, authorised +investigations. +""" + +from __future__ import annotations + +import re +import traceback +from copy import deepcopy + +from connectors_sdk.models import TLPMarking +from pycti import OpenCTIConnectorHelper + +from .client_api import XposedOrNotClient +from .converter_to_stix import ConverterToStix +from .settings import ConnectorSettings + +EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") + + +def is_valid_email(value: str) -> bool: + return bool(value) and len(value) <= 254 and bool(EMAIL_RE.match(value)) + + +def observable_tlp(observable: dict) -> str | None: + """Extract the TLP marking definition of the observable, if any.""" + for marking in observable.get("objectMarking") or []: + if str(marking.get("definition_type", "")).upper() == "TLP": + return marking.get("definition") + return None + + +class XposedOrNotConnector: + SCOPES = ["Email-Addr"] + + def __init__(self, config: ConnectorSettings, helper: OpenCTIConnectorHelper): + self.config = config + self.helper = helper + + api_key = ( + config.xposedornot.api_key.get_secret_value() + if config.xposedornot.api_key + else None + ) + self.max_tlp = config.xposedornot.max_tlp + self.tlp = TLPMarking(level=config.xposedornot.tlp_level) + self.client = XposedOrNotClient( + helper, api_key, str(config.xposedornot.api_base_url) + ) + self.converter = ConverterToStix( + author=ConverterToStix.make_author(), tlp=self.tlp + ) + + def _process_message(self, data: dict) -> str: + observable = data["enrichment_entity"] + entity_type = observable.get("entity_type") + if entity_type not in self.SCOPES: + return "Unsupported type: %s" % entity_type + + tlp = observable_tlp(observable) + if not OpenCTIConnectorHelper.check_max_tlp(tlp, self.max_tlp): + return ( + "TLP of the observable (%s) is higher than what the connector is" + " allowed to enrich (%s); skipping." % (tlp or "none", self.max_tlp) + ) + + email = ( + str(observable.get("observable_value") or observable.get("value") or "") + .strip() + .lower() + ) + if not is_valid_email(email): + return "The observable value is not a valid email address." + + result = self.client.lookup(email) + if result is None: + return "XposedOrNot request failed (see logs)." + if not result: + return "No known breach exposure for this email address (XposedOrNot)." + + breaches = result.get("breaches") or [] + + # Update the source observable in place: score, labels, external reference. + stix_objects = data["stix_objects"] + stix_entity = data["stix_entity"] + enriched_entity = deepcopy(stix_entity) + if result.get("risk_score") is not None: + enriched_entity["x_opencti_score"] = result["risk_score"] + labels = enriched_entity.get("labels") or [] + for label in ["data-breach"] + ( + ["plaintext-password-exposure"] + if self.converter.has_plaintext_exposure(breaches) + else [] + ): + if label not in labels: + labels.append(label) + enriched_entity["labels"] = labels + external_references = enriched_entity.get("external_references") or [] + if not any( + ref.get("source_name") == "XposedOrNot" for ref in external_references + ): + # Append a plain dict to match the existing (dict) references on the + # observable; mixing stix2 objects into a plain-dict STIX entity can + # break bundle serialization. + external_references.append( + { + "source_name": "XposedOrNot", + "url": "https://xposedornot.com", + "description": "XposedOrNot breach exposure check", + } + ) + enriched_entity["external_references"] = external_references + enriched_objects = [ + enriched_entity if obj["id"] == enriched_entity["id"] else obj + for obj in stix_objects + ] + + # Per-breach detail as a markdown Note attached to the observable. + note = self.converter.build_note(enriched_entity["id"], result) + enriched_objects += [ + self.converter.author.to_stix2_object(), + self.tlp.to_stix2_object(), + note.to_stix2_object(), + ] + + bundle = self.helper.stix2_create_bundle(enriched_objects) + self.helper.send_stix2_bundle(bundle, update=True) + + first_year, latest_year = self.converter.years(breaches) + span = ( + " (first %s, latest %s)" % (first_year, latest_year) + if first_year and latest_year + else "" + ) + return ( + "Found %d breach(es)%s; observable updated and summary note attached." + % ( + len(breaches), + span, + ) + ) + + def _process_callback(self, data: dict) -> str: + try: + return self._process_message(data) + except Exception: + self.helper.connector_logger.error( + "Error during enrichment", meta={"trace": traceback.format_exc()} + ) + return "Internal error (see logs)." + + def run(self) -> None: + self.helper.connector_logger.info("Starting the XposedOrNot connector.") + self.helper.listen(message_callback=self._process_callback) diff --git a/internal-enrichment/xposedornot/src/xposedornot/converter_to_stix.py b/internal-enrichment/xposedornot/src/xposedornot/converter_to_stix.py new file mode 100644 index 00000000000..4bf9e723318 --- /dev/null +++ b/internal-enrichment/xposedornot/src/xposedornot/converter_to_stix.py @@ -0,0 +1,130 @@ +# -*- coding: utf-8 -*- +"""Convert an XposedOrNot lookup result into STIX objects. + +The enrichment is deliberately conservative to keep graphs clean: + - the source Email-Addr observable is updated in place (score, labels, + external reference) by the connector; + - the per-breach detail lands in one markdown Note attached to the + observable, rendered by OpenCTI. +""" + +from __future__ import annotations + +from typing import Any + +from connectors_sdk.models import Note, OrganizationAuthor, Reference, TLPMarking + + +def _md_cell(value: Any) -> str: + """Make a value safe for a one-line markdown table cell.""" + text = str(value if value is not None else "—") + return text.replace("|", "\\|").replace("\n", " ").strip() or "—" + + +def _fmt_records(value) -> str: + try: + return "{:,}".format(int(value)) + except (TypeError, ValueError): + return "—" + + +class ConverterToStix: + """Build the Note describing the breach exposure of an email address.""" + + def __init__(self, author: OrganizationAuthor, tlp: TLPMarking): + self.author = author + self.tlp = tlp + + @staticmethod + def make_author() -> OrganizationAuthor: + return OrganizationAuthor( + name="XposedOrNot", + description=( + "Breach exposure data provided by the XposedOrNot API" + " (https://xposedornot.com)." + ), + organization_type="vendor", + ) + + @staticmethod + def years(breaches: list[dict]) -> tuple[int | None, int | None]: + years = [] + for breach in breaches: + try: + years.append(int(str(breach.get("date"))[:4])) + except (TypeError, ValueError): + continue + return (min(years), max(years)) if years else (None, None) + + @staticmethod + def has_plaintext_exposure(breaches: list[dict]) -> bool: + return any(b.get("password_risk") == "plaintextpassword" for b in breaches) + + def build_note(self, source_id: str, result: dict) -> Note: + breaches = result.get("breaches") or [] + first_year, latest_year = self.years(breaches) + total_records = sum(b.get("records") or 0 for b in breaches) + + lines = [ + "## XposedOrNot — breach exposure summary", + "", + "**Breaches found:** %d " % len(breaches), + ] + if first_year and latest_year: + lines.append( + "**First exposure:** %s — **Latest:** %s " % (first_year, latest_year) + ) + if total_records: + lines.append( + "**Total records across breaches:** {:,} ".format(total_records) + ) + if result.get("risk_label"): + score = result.get("risk_score") + lines.append( + "**Overall risk:** %s%s " + % ( + result["risk_label"], + " (%s/100)" % score if score is not None else "", + ) + ) + if self.has_plaintext_exposure(breaches): + lines.append("") + lines.append("⚠️ **At least one breach stored passwords in plaintext.**") + lines += [ + "", + "| Breach | Date | Records | Domain | Exposed data | Password risk | Verified |", + "| --- | --- | --- | --- | --- | --- | --- |", + ] + for breach in breaches: + lines.append( + "| %s | %s | %s | %s | %s | %s | %s |" + % ( + _md_cell(breach.get("name")), + _md_cell(breach.get("date")), + _fmt_records(breach.get("records")), + _md_cell(breach.get("domain")), + _md_cell(", ".join(breach.get("data_classes") or [])), + _md_cell(breach.get("password_risk")), + _md_cell(breach.get("verified")), + ) + ) + lines += [ + "", + "_Generated by the XposedOrNot connector" + " ([xposedornot.com](https://xposedornot.com))._", + # Hidden marker: keeps the Note's generated STIX id unique per source + # observable (the SDK derives the id from content/abstract only, not + # object_refs) while staying stable across re-enrichments. Does not + # render in OpenCTI's markdown view. + "" % source_id, + ] + + return Note( + abstract="XposedOrNot — exposed in %d data breach(es)" % len(breaches), + content="\n".join(lines), + objects=[Reference(id=source_id)], + note_types=["analysis"], + labels=["xposedornot", "data-breach"], + author=self.author, + markings=[self.tlp], + ) diff --git a/internal-enrichment/xposedornot/src/xposedornot/settings.py b/internal-enrichment/xposedornot/src/xposedornot/settings.py new file mode 100644 index 00000000000..2f0b905e350 --- /dev/null +++ b/internal-enrichment/xposedornot/src/xposedornot/settings.py @@ -0,0 +1,71 @@ +# -*- coding: utf-8 -*- +"""Typed connector settings (connectors-sdk BaseConnectorSettings).""" + +from typing import Literal + +from connectors_sdk import ( + BaseConfigModel, + BaseConnectorSettings, + BaseInternalEnrichmentConnectorConfig, + ListFromString, +) +from pydantic import Field, HttpUrl, SecretStr + + +class InternalEnrichmentConnectorConfig(BaseInternalEnrichmentConnectorConfig): + """Defaults for the INTERNAL_ENRICHMENT connector block.""" + + name: str = Field( + description="The name of the connector.", + default="XposedOrNot", + ) + scope: ListFromString = Field( + description="The scope of the connector (observable types to enrich).", + default=["Email-Addr"], + ) + + +class XposedOrNotConfig(BaseConfigModel): + """Configuration specific to the XposedOrNot connector.""" + + api_key: SecretStr | None = Field( + description=( + "Optional XposedOrNot API key (console.xposedornot.com). When set, the" + " connector uses the higher-limit Plus API. Fully functional without it." + ), + default=None, + ) + api_base_url: HttpUrl = Field( + description="Base URL of the free XposedOrNot community API.", + default="https://api.xposedornot.com", + ) + max_tlp: Literal[ + "TLP:CLEAR", + "TLP:WHITE", + "TLP:GREEN", + "TLP:AMBER", + "TLP:AMBER+STRICT", + "TLP:RED", + ] = Field( + description=( + "Maximum TLP of an observable the connector is allowed to enrich. The" + " observable's email address is sent to the XposedOrNot API." + ), + default="TLP:AMBER", + ) + tlp_level: Literal["clear", "green", "amber", "amber+strict", "red"] = Field( + description=( + "TLP marking applied to the objects imported into OpenCTI. Results" + " contain personal data; a restrictive TLP is recommended." + ), + default="amber", + ) + + +class ConnectorSettings(BaseConnectorSettings): + """Override BaseConnectorSettings with the connector-specific config blocks.""" + + connector: InternalEnrichmentConnectorConfig = Field( + default_factory=InternalEnrichmentConnectorConfig + ) + xposedornot: XposedOrNotConfig = Field(default_factory=XposedOrNotConfig) diff --git a/internal-enrichment/xposedornot/tests/fixtures/breach_analytics.json b/internal-enrichment/xposedornot/tests/fixtures/breach_analytics.json new file mode 100644 index 00000000000..2a38301b8f5 --- /dev/null +++ b/internal-enrichment/xposedornot/tests/fixtures/breach_analytics.json @@ -0,0 +1,35 @@ +{ + "BreachMetrics": {"risk": [{"risk_label": "Critical", "risk_score": 100}]}, + "ExposedBreaches": { + "breaches_details": [ + { + "breach": "Sysco", + "details": "Sysco, a food distribution company, was targeted...", + "domain": "sysco.com", + "industry": "Food", + "logo": "https://xposedornot.com/static/logos/Sysco.png", + "password_risk": "unknown", + "references": "", + "searchable": "Yes", + "verified": "Yes", + "xposed_data": "Email addresses;Names;Phone numbers", + "xposed_date": "2026", + "xposed_records": "2699339" + }, + { + "breach": "Yahoo", + "domain": "yahoo.com", + "industry": "Tech", + "password_risk": "plaintextpassword", + "verified": "Yes", + "xposed_data": "Email addresses;Passwords", + "xposed_date": "2013", + "xposed_records": "3000000000" + } + ] + }, + "BreachesSummary": {"site": "Sysco;Yahoo"}, + "ExposedPastes": {}, + "PasteMetrics": {}, + "PastesSummary": {} +} diff --git a/internal-enrichment/xposedornot/tests/fixtures/plus_detailed.json b/internal-enrichment/xposedornot/tests/fixtures/plus_detailed.json new file mode 100644 index 00000000000..d395c26c500 --- /dev/null +++ b/internal-enrichment/xposedornot/tests/fixtures/plus_detailed.json @@ -0,0 +1,18 @@ +{ + "email": "user+tag@example.com", + "breaches": [ + { + "breach_id": "Sysco", + "breached_date": "2026", + "domain": "sysco.com", + "industry": "Food", + "logo": "https://xposedornot.com/static/logos/Sysco.png", + "password_risk": "unknown", + "searchable": "Yes", + "verified": "Yes", + "xposed_data": "Email addresses;Names", + "xposed_records": "2699339", + "details": "Sysco breach details" + } + ] +} diff --git a/internal-enrichment/xposedornot/tests/test-requirements.txt b/internal-enrichment/xposedornot/tests/test-requirements.txt new file mode 100644 index 00000000000..b3b2b3ca4cc --- /dev/null +++ b/internal-enrichment/xposedornot/tests/test-requirements.txt @@ -0,0 +1 @@ +pytest==9.0.3 diff --git a/internal-enrichment/xposedornot/tests/test_client_api.py b/internal-enrichment/xposedornot/tests/test_client_api.py new file mode 100644 index 00000000000..707b3b2532e --- /dev/null +++ b/internal-enrichment/xposedornot/tests/test_client_api.py @@ -0,0 +1,134 @@ +# -*- coding: utf-8 -*- +"""Unit tests for the XposedOrNot API client. + +Covers the free and Plus API paths, clean results (404 and empty 200), +rate limiting and error branches — all with a mocked HTTP session, no +real network call. +""" + +import json +import os +import sys +from unittest.mock import MagicMock, patch + +SRC = os.path.join(os.path.dirname(__file__), "..", "src") +sys.path.insert(0, SRC) + +from xposedornot.client_api import XposedOrNotClient # noqa: E402 + +FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures") + + +def _fixture(name): + with open(os.path.join(FIXTURES, name), "r", encoding="utf-8") as fh: + return json.load(fh) + + +class FakeResp: + def __init__( + self, status_code=200, json_data=None, headers=None, text="", bad_json=False + ): + self.status_code = status_code + self._json = json_data + self.headers = headers or {} + self.text = text + self._bad_json = bad_json + + def json(self): + if self._bad_json: + raise ValueError("bad json") + return self._json + + +def _client(api_key=None): + helper = MagicMock() + return XposedOrNotClient(helper, api_key=api_key), helper + + +def test_free_path_url_and_normalisation(): + client, _ = _client() + resp = FakeResp(200, _fixture("breach_analytics.json")) + with patch.object(client.session, "get", return_value=resp) as mocked_get: + result = client.lookup("test@example.com") + mocked_get.assert_called_once_with( + "https://api.xposedornot.com/v1/breach-analytics", + params={"email": "test@example.com"}, + timeout=30, + ) + assert result["risk_label"] == "Critical" and result["risk_score"] == 100 + assert [b["name"] for b in result["breaches"]] == ["Sysco", "Yahoo"] + sysco = result["breaches"][0] + assert sysco["records"] == 2699339 + assert sysco["data_classes"] == ["Email addresses", "Names", "Phone numbers"] + + +def test_plus_path_used_when_key_set_and_email_is_url_encoded(): + client, _ = _client(api_key="SECRET") + assert client.session.headers["x-api-key"] == "SECRET" + resp = FakeResp(200, _fixture("plus_detailed.json")) + with patch.object(client.session, "get", return_value=resp) as mocked_get: + result = client.lookup("user+tag@example.com") + mocked_get.assert_called_once_with( + "https://plus-api.xposedornot.com/v3/check-email/user%2Btag%40example.com", + params={"detailed": "true"}, + timeout=30, + ) + assert [b["name"] for b in result["breaches"]] == ["Sysco"] + + +def test_clean_email_404_and_empty_200_both_return_empty_dict(): + client, _ = _client() + with patch.object(client.session, "get", return_value=FakeResp(404)): + assert client.lookup("clean@example.org") == {} + empty = {"ExposedBreaches": {"breaches_details": []}} + with patch.object(client.session, "get", return_value=FakeResp(200, empty)): + assert client.lookup("clean@example.org") == {} + + +def test_rate_limit_retries_then_gives_up_without_final_sleep(): + client, helper = _client() + resp_429 = FakeResp(429, headers={"Retry-After": "0"}) + with patch.object(client.session, "get", return_value=resp_429) as mocked_get: + with patch("xposedornot.client_api.time.sleep") as mocked_sleep: + assert client.lookup("test@example.com") is None + assert mocked_get.call_count == 3 + # no sleep on the final attempt -- it would only delay the error + assert mocked_sleep.call_count == 2 + assert helper.connector_logger.error.called + warned = str(helper.connector_logger.warning.call_args) + assert "keyless" in warned + + +def test_rate_limit_message_tailored_for_plus_api(): + client, helper = _client(api_key="SECRET") + resp_429 = FakeResp(429, headers={"Retry-After": "0"}) + with patch.object(client.session, "get", return_value=resp_429): + with patch("xposedornot.client_api.time.sleep"): + assert client.lookup("test@example.com") is None + warned = str(helper.connector_logger.warning.call_args) + assert "Plus API" in warned and "keyless" not in warned + + +def test_rate_limit_recovers_after_backoff(): + client, _ = _client() + responses = [FakeResp(429, headers={"Retry-After": "0"}), FakeResp(404)] + with patch.object(client.session, "get", side_effect=responses): + with patch("xposedornot.client_api.time.sleep"): + assert client.lookup("test@example.com") == {} + + +def test_server_error_and_bad_json_return_none(): + client, helper = _client() + with patch.object(client.session, "get", return_value=FakeResp(500, text="boom")): + assert client.lookup("test@example.com") is None + with patch.object(client.session, "get", return_value=FakeResp(200, bad_json=True)): + assert client.lookup("test@example.com") is None + assert helper.connector_logger.error.call_count == 2 + + +def test_plus_auth_errors_logged_without_key_leak(): + client, helper = _client(api_key="SECRET") + with patch.object(client.session, "get", return_value=FakeResp(422)): + assert client.lookup("test@example.com") is None + logged = str(helper.connector_logger.error.call_args) + assert "SECRET" not in logged diff --git a/internal-enrichment/xposedornot/tests/test_connector.py b/internal-enrichment/xposedornot/tests/test_connector.py new file mode 100644 index 00000000000..a01b6190919 --- /dev/null +++ b/internal-enrichment/xposedornot/tests/test_connector.py @@ -0,0 +1,219 @@ +# -*- coding: utf-8 -*- +"""Unit tests for the connector: settings, init, and the enrichment flow. + +Skipped when connectors_sdk / pycti are not installed, mirroring the sibling +connectors' test convention. +""" + +import importlib.util +import os +import sys +from typing import Any +from unittest.mock import MagicMock + +import pytest + +SRC = os.path.join(os.path.dirname(__file__), "..", "src") +sys.path.insert(0, SRC) + +SDK_AVAILABLE = ( + importlib.util.find_spec("connectors_sdk") is not None + and importlib.util.find_spec("pycti") is not None +) + +sdk_required = pytest.mark.skipif( + not SDK_AVAILABLE, + reason="connectors_sdk / pycti not installed in this environment", +) + +if SDK_AVAILABLE: + from xposedornot.connector import ( + XposedOrNotConnector, + is_valid_email, + observable_tlp, + ) + from xposedornot.settings import ConnectorSettings + + class StubConnectorSettings(ConnectorSettings): + """ConnectorSettings with a fake but valid config dict for tests.""" + + _api_key: Any = None + + @classmethod + def _load_config_dict(cls, _, handler) -> dict: + xon = { + "api_base_url": "https://api.xposedornot.com", + "max_tlp": "TLP:AMBER", + "tlp_level": "amber", + } + if cls._api_key: + xon["api_key"] = cls._api_key + return handler( + { + "opencti": {"url": "http://localhost:8080", "token": "test-token"}, + "connector": { + "id": "connector-id", + "name": "XposedOrNot", + "scope": "Email-Addr", + "log_level": "error", + "auto": False, + }, + "xposedornot": xon, + } + ) + + +BREACHED = { + "breaches": [ + { + "name": "Sysco", + "date": "2026", + "records": 2699339, + "domain": "sysco.com", + "industry": "Food", + "password_risk": "plaintextpassword", + "verified": "Yes", + "data_classes": ["Email addresses", "Names"], + } + ], + "risk_label": "Critical", + "risk_score": 100, +} + +OBSERVABLE_ID = "email-addr--11111111-1111-4111-8111-111111111111" + + +def _make_connector(api_key=None): + StubConnectorSettings._api_key = api_key + settings = StubConnectorSettings() + helper = MagicMock() + helper.stix2_create_bundle.return_value = "BUNDLE" + return XposedOrNotConnector(config=settings, helper=helper), helper + + +def _enrichment_data(email="test@example.com", tlp="TLP:AMBER"): + marking = [{"definition_type": "TLP", "definition": tlp}] if tlp else [] + entity = {"id": OBSERVABLE_ID, "labels": []} + observable = { + "entity_type": "Email-Addr", + "observable_value": email, + "objectMarking": marking, + } + return { + "enrichment_entity": observable, + "stix_entity": entity, + "stix_objects": [entity], + } + + +# --------------------------------------------------------------------------- +# pure helpers +# --------------------------------------------------------------------------- +def test_email_validation(): + if not SDK_AVAILABLE: + pytest.skip("sdk not installed") + assert is_valid_email("user@example.com") + assert is_valid_email("user+tag@sub.example.co.uk") + assert not is_valid_email("not-an-email") + assert not is_valid_email("") + assert not is_valid_email("a@b") + assert not is_valid_email("a" * 250 + "@example.com") + + +def test_observable_tlp_extraction(): + if not SDK_AVAILABLE: + pytest.skip("sdk not installed") + observable = { + "objectMarking": [ + {"definition_type": "statement", "definition": "custom"}, + {"definition_type": "TLP", "definition": "TLP:AMBER"}, + ] + } + assert observable_tlp(observable) == "TLP:AMBER" + assert observable_tlp({"objectMarking": []}) is None + assert observable_tlp({}) is None + + +# --------------------------------------------------------------------------- +# settings + init +# --------------------------------------------------------------------------- +@sdk_required +def test_settings_instantiate_and_helper_config(): + StubConnectorSettings._api_key = None + settings = StubConnectorSettings() + assert isinstance(settings, ConnectorSettings) + assert isinstance(settings.to_helper_config(), dict) + assert settings.xposedornot.max_tlp == "TLP:AMBER" + assert settings.xposedornot.api_key is None + + +@sdk_required +def test_connector_init_keyless_and_keyed(): + connector, _ = _make_connector() + assert connector.max_tlp == "TLP:AMBER" + assert connector.client.api_key is None + keyed, _ = _make_connector(api_key="SECRET") + assert keyed.client.api_key == "SECRET" + + +# --------------------------------------------------------------------------- +# enrichment flow +# --------------------------------------------------------------------------- +@sdk_required +def test_process_message_breached_updates_observable_and_sends_bundle(): + connector, helper = _make_connector() + connector.client.lookup = MagicMock(return_value=BREACHED) + result = connector._process_message(_enrichment_data()) + assert "Found 1 breach" in result + # the enriched observable was included in the sent bundle + sent_objects = helper.stix2_create_bundle.call_args[0][0] + enriched = next(o for o in sent_objects if o.get("id") == OBSERVABLE_ID) + assert enriched["x_opencti_score"] == 100 + assert "data-breach" in enriched["labels"] + assert "plaintext-password-exposure" in enriched["labels"] + assert any( + ref["source_name"] == "XposedOrNot" for ref in enriched["external_references"] + ) + assert isinstance(enriched["external_references"][0], dict) + helper.send_stix2_bundle.assert_called_once() + + +@sdk_required +def test_process_message_clean_email_modifies_nothing(): + connector, helper = _make_connector() + connector.client.lookup = MagicMock(return_value={}) + result = connector._process_message(_enrichment_data()) + assert "No known breach exposure" in result + helper.send_stix2_bundle.assert_not_called() + + +@sdk_required +def test_process_message_tlp_exceeded_skips_before_lookup(): + connector, _ = _make_connector() + connector.client.lookup = MagicMock() + result = connector._process_message(_enrichment_data(tlp="TLP:RED")) + assert "higher than" in result + connector.client.lookup.assert_not_called() + + +@sdk_required +def test_process_message_unsupported_type_and_invalid_email(): + connector, _ = _make_connector() + data = _enrichment_data() + data["enrichment_entity"]["entity_type"] = "IPv4-Addr" + assert "Unsupported type" in connector._process_message(data) + + connector.client.lookup = MagicMock() + bad = _enrichment_data(email="not-an-email") + assert "not a valid email" in connector._process_message(bad) + connector.client.lookup.assert_not_called() + + +@sdk_required +def test_process_message_api_failure_and_callback_guard(): + connector, helper = _make_connector() + connector.client.lookup = MagicMock(return_value=None) + assert "request failed" in connector._process_message(_enrichment_data()) + # the callback wrapper converts exceptions into a safe message + connector._process_message = MagicMock(side_effect=RuntimeError("boom")) + assert "Internal error" in connector._process_callback(_enrichment_data()) diff --git a/internal-enrichment/xposedornot/tests/test_converter.py b/internal-enrichment/xposedornot/tests/test_converter.py new file mode 100644 index 00000000000..d83ba8fd842 --- /dev/null +++ b/internal-enrichment/xposedornot/tests/test_converter.py @@ -0,0 +1,90 @@ +# -*- coding: utf-8 -*- +"""Unit tests for the STIX converter (Note building, helpers). + +Skipped when connectors_sdk / pycti are not installed, mirroring the test +convention of sibling connectors. +""" + +import importlib.util +import json +import os +import sys + +import pytest + +SRC = os.path.join(os.path.dirname(__file__), "..", "src") +sys.path.insert(0, SRC) + +SDK_AVAILABLE = ( + importlib.util.find_spec("connectors_sdk") is not None + and importlib.util.find_spec("pycti") is not None +) + +if SDK_AVAILABLE: + from connectors_sdk.models import TLPMarking + from xposedornot.client_api import _normalise_free + from xposedornot.converter_to_stix import ConverterToStix, _md_cell + +sdk_required = pytest.mark.skipif( + not SDK_AVAILABLE, + reason="connectors_sdk / pycti not installed in this environment", +) + +FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures") +SOURCE_ID = "email-addr--3f8a3b62-32e9-5efc-8e18-4b34ba6c4f36" + + +def _result(): + with open( + os.path.join(FIXTURES, "breach_analytics.json"), "r", encoding="utf-8" + ) as fh: + return _normalise_free(json.load(fh)) + + +@sdk_required +def test_note_contains_breach_table_and_summary(): + converter = ConverterToStix( + ConverterToStix.make_author(), TLPMarking(level="amber") + ) + note = converter.build_note(SOURCE_ID, _result()) + stix = note.to_stix2_object() + content = stix["content"] + assert "**Breaches found:** 2" in content + assert "**First exposure:** 2013 — **Latest:** 2026" in content + assert "**Overall risk:** Critical (100/100)" in content + assert "plaintext" in content.lower() + assert "| Sysco | 2026 | 2,699,339 | sysco.com |" in content + assert SOURCE_ID in stix["object_refs"] + assert "XposedOrNot — exposed in 2 data breach(es)" == stix["abstract"] + + +@sdk_required +def test_note_is_deterministic_across_runs(): + converter = ConverterToStix( + ConverterToStix.make_author(), TLPMarking(level="amber") + ) + id_one = converter.build_note(SOURCE_ID, _result()).to_stix2_object()["id"] + id_two = converter.build_note(SOURCE_ID, _result()).to_stix2_object()["id"] + assert id_one == id_two + + +@sdk_required +def test_note_id_is_unique_per_source_observable(): + # Two different observables with an identical breach set must NOT share a + # Note id (the SDK derives the id from content/abstract, not object_refs). + converter = ConverterToStix( + ConverterToStix.make_author(), TLPMarking(level="amber") + ) + other_id = "email-addr--22222222-2222-4222-8222-222222222222" + id_a = converter.build_note(SOURCE_ID, _result()).to_stix2_object()["id"] + id_b = converter.build_note(other_id, _result()).to_stix2_object()["id"] + assert id_a != id_b + + +@sdk_required +def test_helpers(): + result = _result() + assert ConverterToStix.years(result["breaches"]) == (2013, 2026) + assert ConverterToStix.has_plaintext_exposure(result["breaches"]) is True + assert _md_cell("a|b\nc") == "a\\|b c" + assert _md_cell(None) == "—"