diff --git a/.gitignore b/.gitignore index 85953837321..00f6da35382 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,8 @@ Pipfile* venv* venv_* .temp_venv +.temp_venv_run +.temp_venv_review black *.key *.crt diff --git a/external-import/comlaude/src/comlaude/__init__.py b/external-import/comlaude/src/comlaude/__init__.py index b9b58517178..a1679091b1e 100644 --- a/external-import/comlaude/src/comlaude/__init__.py +++ b/external-import/comlaude/src/comlaude/__init__.py @@ -243,6 +243,8 @@ def _response_error(message, response): else: error_message = json.loads(response.text) - raise Exception(f"""Message:{message}. + raise Exception( + f"""Message:{message}. Response code returned:{response.status_code}. - Error message returned:{error_message}.""") + Error message returned:{error_message}.""" + ) diff --git a/external-import/tenable-vuln-management/tests/test_converter_to_stix.py b/external-import/tenable-vuln-management/tests/test_converter_to_stix.py index 01a9b2db43e..a8004cdd7e7 100644 --- a/external-import/tenable-vuln-management/tests/test_converter_to_stix.py +++ b/external-import/tenable-vuln-management/tests/test_converter_to_stix.py @@ -43,7 +43,8 @@ def mock_config(): @pytest.fixture def fake_asset(): - return Asset.model_validate_json(""" + return Asset.model_validate_json( + """ { "device_type": "general-purpose", "fqdn": "sharepoint2016.target.example.com", @@ -59,12 +60,14 @@ def fake_asset(): "tracked": true, "last_scan_target": "192.0.0.1" } - """) + """ + ) @pytest.fixture def fake_plugin(): - return Plugin.model_validate_json(""" + return Plugin.model_validate_json( + """ { "bid": [ 156641 @@ -190,12 +193,14 @@ def fake_plugin(): ], "type": "local" } - """) + """ + ) @pytest.fixture def fake_vuln_finding(): - return VulnerabilityFinding.model_validate_json("""{ + return VulnerabilityFinding.model_validate_json( + """{ "asset": { "device_type": "hypervisor", "fqdn": "vcsa8.target.example.com", @@ -270,7 +275,8 @@ def fake_vuln_finding(): "indexed": "2023-05-04T09:44:55.673359Z", "source": "NESSUS" } - """) + """ + ) def test_tlp_marking_definition_handler_should_fails_with_unsupported_TLP(): diff --git a/internal-enrichment/import-external-reference/src/import-external-reference.py b/internal-enrichment/import-external-reference/src/import-external-reference.py index 42e7df78aa9..0256b04ca81 100644 --- a/internal-enrichment/import-external-reference/src/import-external-reference.py +++ b/internal-enrichment/import-external-reference/src/import-external-reference.py @@ -275,7 +275,8 @@ async def _dismiss_cookies(self, page: Page) -> bool: # Fallback: try to hide/remove banners/popups via JS if nothing was clicked if not found: try: - await page.evaluate(""" + await page.evaluate( + """ const keywords = ["cookie", "consent", "banner", "popup", "gdpr", "privacy", "notice", "modal", "alert", "dialog"]; let removed = false; for (const kw of keywords) { @@ -290,7 +291,8 @@ async def _dismiss_cookies(self, page: Page) -> bool: } } return removed; - """) + """ + ) self.helper.log_debug("Ran JS fallback to hide banners/popups.") except Exception as e: self.helper.log_debug(f"JS fallback for hiding banners failed: {e}") @@ -310,13 +312,15 @@ async def _browser_worker(self): viewport={"width": 1280, "height": 800}, extra_http_headers=self.headers, ) - await ctx.add_init_script(""" + await ctx.add_init_script( + """ Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); Object.defineProperty(navigator, 'platform', { get: () => 'Win32' }); Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); window.chrome = { runtime: {}, loadTimes: () => { return null; }, csi: () => { return {}; } }; - """) + """ + ) page = await ctx.new_page() status_code = None diff --git a/internal-enrichment/polyswarm-enrichment/tests/test_e2e_keydoor.py b/internal-enrichment/polyswarm-enrichment/tests/test_e2e_keydoor.py index 4e6346ffd38..66deddaf993 100644 --- a/internal-enrichment/polyswarm-enrichment/tests/test_e2e_keydoor.py +++ b/internal-enrichment/polyswarm-enrichment/tests/test_e2e_keydoor.py @@ -120,7 +120,8 @@ def delete_observable(sha256: str): # Query helpers # --------------------------------------------------------------------------- def get_observable_by_hash(sha256: str) -> dict | None: - query = """ + query = ( + """ { stixCyberObservables( filters: { @@ -141,14 +142,17 @@ def get_observable_by_hash(sha256: str) -> dict | None: } } } - """ % sha256 + """ + % sha256 + ) result = graphql(query) edges = result["stixCyberObservables"]["edges"] return edges[0]["node"] if edges else None def get_notes_for_observable(observable_id: str) -> list[dict]: - query = """ + query = ( + """ { notes( filters: { @@ -160,7 +164,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]: edges { node { id attribute_abstract content } } } } - """ % observable_id + """ + % observable_id + ) result = graphql(query) return [e["node"] for e in result["notes"]["edges"]] diff --git a/internal-enrichment/polyswarm-enrichment/tests/test_e2e_network_iocs.py b/internal-enrichment/polyswarm-enrichment/tests/test_e2e_network_iocs.py index bd51a1a3d29..3f1a264b087 100644 --- a/internal-enrichment/polyswarm-enrichment/tests/test_e2e_network_iocs.py +++ b/internal-enrichment/polyswarm-enrichment/tests/test_e2e_network_iocs.py @@ -132,7 +132,8 @@ def create_observable_via_pycti(sha256: str, description: str = "") -> dict: # --------------------------------------------------------------------------- def get_observable_by_hash(sha256: str) -> dict | None: """Look up a StixFile observable by SHA-256 hash.""" - query = """ + query = ( + """ { stixCyberObservables( filters: { @@ -153,7 +154,9 @@ def get_observable_by_hash(sha256: str) -> dict | None: } } } - """ % sha256 + """ + % sha256 + ) result = graphql(query) edges = result["stixCyberObservables"]["edges"] return edges[0]["node"] if edges else None @@ -161,7 +164,8 @@ def get_observable_by_hash(sha256: str) -> dict | None: def get_notes_for_observable(observable_id: str) -> list[dict]: """Get Notes linked to an observable.""" - query = """ + query = ( + """ { notes( filters: { @@ -173,7 +177,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]: edges { node { id attribute_abstract content } } } } - """ % observable_id + """ + % observable_id + ) result = graphql(query) return [e["node"] for e in result["notes"]["edges"]] diff --git a/internal-enrichment/polyswarm-enrichment/tests/test_e2e_opencti.py b/internal-enrichment/polyswarm-enrichment/tests/test_e2e_opencti.py index b355fb2ad64..2ec5f4b50ea 100644 --- a/internal-enrichment/polyswarm-enrichment/tests/test_e2e_opencti.py +++ b/internal-enrichment/polyswarm-enrichment/tests/test_e2e_opencti.py @@ -118,7 +118,8 @@ def delete_observable(internal_id: str): # --------------------------------------------------------------------------- def get_observable_by_hash(sha256: str) -> dict | None: """Look up a StixFile observable by SHA-256 hash.""" - query = """ + query = ( + """ { stixCyberObservables( filters: { @@ -143,7 +144,9 @@ def get_observable_by_hash(sha256: str) -> dict | None: } } } - """ % sha256 + """ + % sha256 + ) result = graphql(query) edges = result["stixCyberObservables"]["edges"] return edges[0]["node"] if edges else None @@ -151,7 +154,8 @@ def get_observable_by_hash(sha256: str) -> dict | None: def get_notes_for_observable(observable_id: str) -> list[dict]: """Get Notes linked to an observable.""" - query = """ + query = ( + """ { notes( filters: { @@ -169,7 +173,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]: } } } - """ % observable_id + """ + % observable_id + ) result = graphql(query) return [e["node"] for e in result["notes"]["edges"]] @@ -370,11 +376,13 @@ def feeder_enriched(): """Wait for any feeder-created observable to be enriched.""" deadline = time.time() + ENRICHMENT_TIMEOUT while time.time() < deadline: - result = graphql(""" + result = graphql( + """ { stixCyberObservables(first: 20, orderBy: created_at, orderMode: desc) { edges { node { id observable_value x_opencti_score } } } } - """) + """ + ) for edge in result["stixCyberObservables"]["edges"]: node = edge["node"] if node["x_opencti_score"] and node["x_opencti_score"] > 50: diff --git a/internal-enrichment/polyswarm-sandbox/tests/test_e2e_sandbox.py b/internal-enrichment/polyswarm-sandbox/tests/test_e2e_sandbox.py index b327d55d074..74f935b1265 100644 --- a/internal-enrichment/polyswarm-sandbox/tests/test_e2e_sandbox.py +++ b/internal-enrichment/polyswarm-sandbox/tests/test_e2e_sandbox.py @@ -159,7 +159,8 @@ def get_artifact_by_id(artifact_id: str) -> dict | None: def get_notes_for_observable(observable_id: str) -> list[dict]: - query = """ + query = ( + """ { notes( filters: { @@ -177,7 +178,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]: } } } - """ % observable_id + """ + % observable_id + ) result = graphql(query) return [e["node"] for e in result["notes"]["edges"]] diff --git a/internal-enrichment/reversinglabs-malware-presence/src/connector/connector.py b/internal-enrichment/reversinglabs-malware-presence/src/connector/connector.py index d41fdae2c8a..a288281ddda 100644 --- a/internal-enrichment/reversinglabs-malware-presence/src/connector/connector.py +++ b/internal-enrichment/reversinglabs-malware-presence/src/connector/connector.py @@ -157,7 +157,8 @@ def _generate_stix_note(self, data): scanner_percent = data["rl"]["malware_presence"]["scanner_percent"] abstract = "ReversingLabs Spectra Intelligence Results" - content = textwrap.dedent(f""" + content = textwrap.dedent( + f""" ## ReversingLabs Spectra Intelligence Results Detected hash {hash} is **{status}** | Scanner | | @@ -165,7 +166,8 @@ def _generate_stix_note(self, data): | Count | {scanner_count} | | Match | {scanner_match} | | (%) | {scanner_percent} | - """) + """ + ) # Create Note stix_note = stix2.Note( diff --git a/internal-enrichment/reversinglabs-spectra-analyze/src/main.py b/internal-enrichment/reversinglabs-spectra-analyze/src/main.py index bfdcc8210fb..4ec41bcb755 100644 --- a/internal-enrichment/reversinglabs-spectra-analyze/src/main.py +++ b/internal-enrichment/reversinglabs-spectra-analyze/src/main.py @@ -678,7 +678,8 @@ def _ip_report(self): dl_files_statistics = resp_json.get("downloaded_files_statistics", {}) abstract = "ReversingLabs Spectra Analyze IP address report" - content = textwrap.dedent(f""" + content = textwrap.dedent( + f""" ## ReversingLabs Spectra Analyze IP address report for {self.ip_sample} Third party statistics | Status | Amount | @@ -697,7 +698,8 @@ def _ip_report(self): | SUSPICIOUS | {dl_files_statistics.get("suspicious")} | | UNKNOWN | {dl_files_statistics.get("unknown")} | | TOTAL | {dl_files_statistics.get("total")} | - """) + """ + ) note = stix2.Note( id=Note.generate_id(None, content), @@ -842,11 +844,13 @@ def _domain_reports(self, domain_list): if selected_domains: abstract = "ReversingLabs Spectra Analyze domain statistics" - accumulated_content = textwrap.dedent(""" + accumulated_content = textwrap.dedent( + """ ## ReversingLabs Spectra Analyze domain statistics | Domain | Third party statistics Malicious/Total | Downloaded files statistics Malicious/Total | | ------------- | --------------- | --------------- | - """) + """ + ) for one_domain in selected_domains: domain_name = one_domain.get("requested_domain") @@ -981,11 +985,13 @@ def _url_reports(self, url_list): if selected_urls: abstract = "ReversingLabs Spectra Analyze URL statistics" - accumulated_content = textwrap.dedent(""" + accumulated_content = textwrap.dedent( + """ ## ReversingLabs Spectra Analyze URL statistics | URL | Third party statistics Malicious/Total | Analysis statistics Malicious/Total | | ------------- | --------------- | --------------- | - """) + """ + ) for one_url in selected_urls: url_name = one_url.get("requested_url") @@ -1107,7 +1113,8 @@ def _domain_report(self): ) dl_files_statistics = resp_json.get("downloaded_files_statistics", {}) - content = textwrap.dedent(f""" + content = textwrap.dedent( + f""" ## ReversingLabs Spectra Analyze domain report for `{self.domain_sample}` Third party statistics | Status | Amount | @@ -1126,7 +1133,8 @@ def _domain_report(self): | SUSPICIOUS | {dl_files_statistics.get("suspicious")} | | UNKNOWN | {dl_files_statistics.get("unknown")} | | TOTAL | {dl_files_statistics.get("total")} | - """) + """ + ) note = stix2.Note( id=Note.generate_id(None, content), @@ -1226,7 +1234,8 @@ def _url_report(self): analysis_stats = resp_json.get("analysis", {}).get("statistics", {}) tp_stats = resp_json.get("third_party_reputations", {}).get("statistics", {}) - content = textwrap.dedent(f""" + content = textwrap.dedent( + f""" ## ReversingLabs Spectra Analyze URL report for `{self.url_sample}` Third party statistics @@ -1246,7 +1255,8 @@ def _url_report(self): | SUSPICIOUS | {analysis_stats.get("suspicious")} | | UNKNOWN | {analysis_stats.get("unknown")} | | TOTAL | {analysis_stats.get("total")} | - """) + """ + ) note = stix2.Note( id=Note.generate_id(None, content), diff --git a/internal-enrichment/reversinglabs-spectra-intel-submission/src/connector/connector.py b/internal-enrichment/reversinglabs-spectra-intel-submission/src/connector/connector.py index bc1d9dac6fb..92d8f28d1e0 100644 --- a/internal-enrichment/reversinglabs-spectra-intel-submission/src/connector/connector.py +++ b/internal-enrichment/reversinglabs-spectra-intel-submission/src/connector/connector.py @@ -192,7 +192,8 @@ def _generate_stix_note(self, results): analysis_duration = results["analysis_duration"] abstract = "ReversingLabs Spectra Sandbox Results" - analysis_content = textwrap.dedent(f""" + analysis_content = textwrap.dedent( + f""" # ReversingLabs Spectra Sandbox Analysis metadata Classification: **{classification}** @@ -207,7 +208,8 @@ def _generate_stix_note(self, results): Analysis is executed on **{platform}** operating system with following configuration: **{configuration}** - """) + """ + ) signature_text_header = """ @@ -215,10 +217,12 @@ def _generate_stix_note(self, results): """ - signature_text_content = textwrap.dedent("""\ + signature_text_content = textwrap.dedent( + """\ | Description | Risk Factor | |-----------------|-------------------| - """) + """ + ) signatures_list = results["signatures"] sorted_signatures = sorted( signatures_list, key=lambda x: x["risk_factor"], reverse=True @@ -227,9 +231,11 @@ def _generate_stix_note(self, results): for sig in sorted_signatures: sig_description = sig["description"] sig_risk_factor = sig["risk_factor"] - signature_text_content += textwrap.dedent(f"""\ + signature_text_content += textwrap.dedent( + f"""\ | {sig_description} | {sig_risk_factor} | - """) + """ + ) signature_text_header = textwrap.dedent(signature_text_header) signature_text = signature_text_header + signature_text_content diff --git a/internal-enrichment/whoisfreaks/Dockerfile b/internal-enrichment/whoisfreaks/Dockerfile new file mode 100644 index 00000000000..a29705ba8a8 --- /dev/null +++ b/internal-enrichment/whoisfreaks/Dockerfile @@ -0,0 +1,26 @@ +FROM python:3.12-alpine +ENV CONNECTOR_TYPE=INTERNAL_ENRICHMENT + +# Set working directory inside container +WORKDIR /opt/opencti-connector-whoisfreaks + +# Install system runtime dependencies needed by pycti and stix2 +# Install system runtime/build dependencies needed by pycti and stix2 +# hadolint ignore=DL3003 +RUN apk update && apk upgrade && \ + apk --no-cache add git build-base libmagic libffi-dev + +# Install Python package dependencies +COPY requirements.txt . +RUN pip3 install --no-cache-dir -r requirements.txt && \ + apk del git build-base + +# Copy source code, config, and entrypoint +COPY src ./src +COPY config.yml ./config.yml +COPY entrypoint.sh /entrypoint.sh +# Ensure entrypoint is executable +RUN chmod +x /entrypoint.sh + +# Run startup script +ENTRYPOINT ["/entrypoint.sh"] \ No newline at end of file diff --git a/internal-enrichment/whoisfreaks/README.md b/internal-enrichment/whoisfreaks/README.md new file mode 100644 index 00000000000..75ade030931 --- /dev/null +++ b/internal-enrichment/whoisfreaks/README.md @@ -0,0 +1,61 @@ +# OpenCTI WhoisFreaks Enrichment Connector + +An official internal enrichment connector for the **OpenCTI** (Open Cyber Threat Intelligence) platform powered by the **WhoisFreaks API**. + +This connector enables security analysts to enrich `Domain-Name`, `IPv4-Addr`, and `IPv6-Addr` observables on demand or via automated OpenCTI playbooks. + +--- + +## Features + +- **WHOIS Registration Intelligence**: Fetches live WHOIS records and converts Registrars and Registrants into STIX 2.1 `Identity` SDOs linked via `registered-by` and `owned-by` relationships. +- **DNS Record Mapping**: Parses live DNS responses (A, AAAA, MX, NS, CNAME) and maps resolutions into the OpenCTI graph using `resolves-to` and `related-to` relationships. +- **Subdomain Discovery**: Maps parent-child domain hierarchies and populates subdomains as related `Domain-Name` observables. +- **SSL/TLS Certificate Metadata**: Extracts X.509 certificate metadata and attaches `X509-Certificate` observables to target domains. +- **IP Geolocation & Reverse DNS**: Translates IPv4/IPv6 observables into STIX `Location` SDOs (`located-at`) and maps reverse DNS domain associations. +- **IP Reputation Context**: Evaluates threat scores and attaches contextual security notes to IP entities. + +--- + +## Configuration Variables + +The connector is configured using environment variables (or `.env` / `config.yml`): + +| Variable | Required | Default | Description | +| ---------------------------- | -------- | --------------------------------- | ------------------------------------------------------------------------ | +| `OPENCTI_URL` | Yes | - | Base URL of your OpenCTI platform instance (e.g. `http://opencti:4000`). | +| `OPENCTI_TOKEN` | Yes | - | OpenCTI User / Admin API Token for authentication. | +| `CONNECTOR_ID` | Yes | - | A unique UUIDv4 string generated specifically for this connector. | +| `CONNECTOR_NAME` | No | `WhoisFreaks` | Display name of the connector inside the OpenCTI dashboard. | +| `CONNECTOR_TYPE` | No | `INTERNAL_ENRICHMENT` | Connector operational category in OpenCTI. | +| `CONNECTOR_SCOPE` | No | `Domain-Name,IPv4-Addr,IPv6-Addr` | Supported STIX 2.1 entity types. | +| `CONNECTOR_AUTO` | No | `false` | Enable automatic enrichment on observable creation (`true`/`false`). | +| `CONNECTOR_CONFIDENCE_LEVEL` | No | `100` | Confidence score (0-100) assigned to created STIX objects. | +| `CONNECTOR_LOG_LEVEL` | No | `info` | Log verbosity (`debug`, `info`, `warning`, `error`). | +| `WHOISFREAKS_API_KEY` | Yes | - | Your active WhoisFreaks API key. | + +--- + +## Deployment via Docker Compose + +Add the service block below to your OpenCTI stack `docker-compose.yml`: + +```yaml +connector-whoisfreaks: + image: opencti-connector-whoisfreaks:latest + container_name: opencti-connector-whoisfreaks + environment: + - OPENCTI_URL=http://opencti:4000 + - OPENCTI_TOKEN=${OPENCTI_ADMIN_TOKEN} + - CONNECTOR_ID=${CONNECTOR_KEY} + - CONNECTOR_TYPE=INTERNAL_ENRICHMENT + - CONNECTOR_NAME=WhoisFreaks + - CONNECTOR_SCOPE=Domain-Name,IPv4-Addr,IPv6-Addr + - CONNECTOR_AUTO=false + - CONNECTOR_CONFIDENCE_LEVEL=100 + - CONNECTOR_LOG_LEVEL=info + - WHOISFREAKS_API_KEY=${WHOISFREAKS_API_KEY} + depends_on: + - opencti + restart: always +``` diff --git a/internal-enrichment/whoisfreaks/__metadata__/connector_config_schema.json b/internal-enrichment/whoisfreaks/__metadata__/connector_config_schema.json new file mode 100644 index 00000000000..e9c48a92ef2 --- /dev/null +++ b/internal-enrichment/whoisfreaks/__metadata__/connector_config_schema.json @@ -0,0 +1,81 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://www.filigran.io/connectors/whoisfreaks_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 stable UUID for this connector instance.", + "type": "string" + }, + "CONNECTOR_NAME": { + "default": "WhoisFreaks", + "description": "The name of the connector.", + "type": "string" + }, + "CONNECTOR_SCOPE": { + "default": [ + "Domain-Name", + "IPv4-Addr", + "IPv6-Addr" + ], + "description": "The scope of the connector.", + "items": { + "type": "string" + }, + "type": "array" + }, + "CONNECTOR_LOG_LEVEL": { + "default": "info", + "description": "The minimum level of logs to display.", + "enum": [ + "debug", + "info", + "warn", + "warning", + "error" + ], + "type": "string" + }, + "CONNECTOR_TYPE": { + "const": "INTERNAL_ENRICHMENT", + "default": "INTERNAL_ENRICHMENT", + "type": "string" + }, + "CONNECTOR_AUTO": { + "default": false, + "description": "Whether the connector should run automatically when an entity is created or updated.", + "type": "boolean" + }, + "CONNECTOR_CONFIDENCE_LEVEL": { + "default": 100, + "description": "Default confidence level (0-100) attached to STIX objects created by the connector.", + "maximum": 100, + "minimum": 0, + "type": "integer" + }, + "WHOISFREAKS_API_KEY": { + "description": "API key for accessing the WhoisFreaks service", + "format": "password", + "type": "string", + "writeOnly": true + } + }, + "required": [ + "OPENCTI_URL", + "OPENCTI_TOKEN", + "WHOISFREAKS_API_KEY", + "CONNECTOR_ID" + ], + "additionalProperties": true +} diff --git a/internal-enrichment/whoisfreaks/__metadata__/connector_manifest.json b/internal-enrichment/whoisfreaks/__metadata__/connector_manifest.json new file mode 100644 index 00000000000..08f9fe57182 --- /dev/null +++ b/internal-enrichment/whoisfreaks/__metadata__/connector_manifest.json @@ -0,0 +1,26 @@ +{ + "title": "WhoisFreaks", + "slug": "whoisfreaks", + "description": "An official internal enrichment connector for OpenCTI powered by WhoisFreaks API. Enriches Domain-Name, IPv4-Addr, and IPv6-Addr observables with live WHOIS registration data, DNS records, SSL/TLS certificate metadata, IP geolocation, and threat context.", + "short_description": "Enrich domain, IP, and SSL observables with live WHOIS, DNS, Geolocation, and threat intelligence from WhoisFreaks.", + "logo": "internal-enrichment/whoisfreaks/__metadata__/logo.png", + "use_cases": [ + "Infrastructure & Attack Surface Visibility" + ], + "solution_categories": [ + "Enrichment & Reputation" + ], + "contact": null, + "license_type": "Commercial", + "verified": false, + "last_verified_date": null, + "playbook_supported": true, + "max_confidence_level": 100, + "support_version": ">=6.0.0", + "subscription_link": "https://whoisfreaks.com", + "source_code": "https://github.com/OpenCTI-Platform/connectors/tree/master/internal-enrichment/whoisfreaks", + "manager_supported": true, + "container_version": "rolling", + "container_image": "opencti/connector-whoisfreaks", + "container_type": "INTERNAL_ENRICHMENT" +} diff --git a/internal-enrichment/whoisfreaks/__metadata__/logo.png b/internal-enrichment/whoisfreaks/__metadata__/logo.png new file mode 100644 index 00000000000..d1d5bf395af Binary files /dev/null and b/internal-enrichment/whoisfreaks/__metadata__/logo.png differ diff --git a/internal-enrichment/whoisfreaks/config.yml b/internal-enrichment/whoisfreaks/config.yml new file mode 100644 index 00000000000..e0aef8679c0 --- /dev/null +++ b/internal-enrichment/whoisfreaks/config.yml @@ -0,0 +1,15 @@ +opencti: + url: "http://localhost:8080" + token: "ChangeMe" + +connector: + id: "ChangeMe" # Must be a valid UUIDv4 (e.g. generated via `uuidgen`) + type: "INTERNAL_ENRICHMENT" + name: "WhoisFreaks" + scope: "Domain-Name,IPv4-Addr,IPv6-Addr" + auto: false # Manual trigger recommended to manage API quota + confidence_level: 100 + log_level: "info" + +whoisfreaks: + api_key: "ChangeMe" diff --git a/internal-enrichment/whoisfreaks/docker-compose.yml b/internal-enrichment/whoisfreaks/docker-compose.yml new file mode 100644 index 00000000000..e451f7f3038 --- /dev/null +++ b/internal-enrichment/whoisfreaks/docker-compose.yml @@ -0,0 +1,87 @@ +version: "3" + +services: + redis: + image: redis:7.2-alpine + restart: always + + elasticsearch: + image: docker.elastic.co/elasticsearch/elasticsearch:8.12.2 + environment: + - discovery.type=single-node + - xpack.security.enabled=false + - "ES_JAVA_OPTS=-Xms512m -Xmx512m" + restart: always + + minio: + image: minio/minio:RELEASE.2024-02-17T01-15-57Z + command: server /data + environment: + - MINIO_ROOT_USER=${MINIO_ROOT_USER:-opencti} + - MINIO_ROOT_PASSWORD=${MINIO_ROOT_PASSWORD} + restart: always + + rabbitmq: + image: rabbitmq:3.12-management-alpine + restart: always + + opencti: + image: opencti/platform:6.0.5 + environment: + - NODE_OPTIONS=--max-old-space-size=4096 + - APP__ADMIN__EMAIL=${OPENCTI_ADMIN_EMAIL} + - APP__ADMIN__PASSWORD=${OPENCTI_ADMIN_PASSWORD} + - APP__ADMIN__TOKEN=${OPENCTI_ADMIN_TOKEN} + - APP__BASE_URL=${OPENCTI_BASE_URL} + - REDIS__HOSTNAME=redis + - REDIS__PORT=6379 + - ELASTICSEARCH__URL=http://elasticsearch:9200 + - MINIO__ENDPOINT=minio + - MINIO__PORT=9000 + - MINIO__USE_SSL=false + - MINIO__ACCESS_KEY=opencti + - MINIO__SECRET_KEY=${MINIO_ROOT_PASSWORD} + - RABBITMQ__HOSTNAME=rabbitmq + - RABBITMQ__PORT=5672 + - RABBITMQ__PORT_MANAGEMENT=15672 + - RABBITMQ__USER=guest + - RABBITMQ__PASSWORD=guest + ports: + - "8080:4000" + depends_on: + - redis + - elasticsearch + - minio + - rabbitmq + restart: always + + worker: + image: opencti/worker:6.0.5 + environment: + - OPENCTI_URL=http://opencti:4000 + - OPENCTI_TOKEN=${OPENCTI_ADMIN_TOKEN} + - WORKER_LOG_LEVEL=INFO + depends_on: + - opencti + restart: always + + connector-whoisfreaks: + build: + context: . + dockerfile: Dockerfile + image: opencti-connector-whoisfreaks:latest + container_name: opencti-connector-whoisfreaks + environment: + - OPENCTI_URL=http://opencti:4000 + - OPENCTI_TOKEN=${OPENCTI_ADMIN_TOKEN} + - CONNECTOR_ID=${CONNECTOR_KEY} + - CONNECTOR_TYPE=INTERNAL_ENRICHMENT + - CONNECTOR_NAME=WhoisFreaks + - CONNECTOR_SCOPE=Domain-Name,IPv4-Addr,IPv6-Addr + - CONNECTOR_AUTO=false + - CONNECTOR_CONFIDENCE_LEVEL=100 + - CONNECTOR_LOG_LEVEL=info + - WHOISFREAKS_API_KEY=${WHOISFREAKS_API_KEY} + depends_on: + - opencti + restart: always \ No newline at end of file diff --git a/internal-enrichment/whoisfreaks/entrypoint.sh b/internal-enrichment/whoisfreaks/entrypoint.sh new file mode 100644 index 00000000000..2744f930110 --- /dev/null +++ b/internal-enrichment/whoisfreaks/entrypoint.sh @@ -0,0 +1,7 @@ +#!/bin/sh + +# Navigate to app directory +cd /opt/opencti-connector-whoisfreaks + +# Start the OpenCTI WhoisFreaks connector main script +exec python3 src/main.py \ No newline at end of file diff --git a/internal-enrichment/whoisfreaks/requirements.txt b/internal-enrichment/whoisfreaks/requirements.txt new file mode 100644 index 00000000000..aca3b476cbc --- /dev/null +++ b/internal-enrichment/whoisfreaks/requirements.txt @@ -0,0 +1,4 @@ +pycti==6.0.5 +stix2>=3.0.1 +requests>=2.31.0 +pyyaml>=6.0.1 \ No newline at end of file diff --git a/internal-enrichment/whoisfreaks/src/__init__.py b/internal-enrichment/whoisfreaks/src/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/internal-enrichment/whoisfreaks/src/__pycache__/__init__.cpython-312.pyc b/internal-enrichment/whoisfreaks/src/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 00000000000..87a2bad50e3 Binary files /dev/null and b/internal-enrichment/whoisfreaks/src/__pycache__/__init__.cpython-312.pyc differ diff --git a/internal-enrichment/whoisfreaks/src/__pycache__/builder.cpython-312.pyc b/internal-enrichment/whoisfreaks/src/__pycache__/builder.cpython-312.pyc new file mode 100644 index 00000000000..38c14390836 Binary files /dev/null and b/internal-enrichment/whoisfreaks/src/__pycache__/builder.cpython-312.pyc differ diff --git a/internal-enrichment/whoisfreaks/src/__pycache__/client.cpython-312.pyc b/internal-enrichment/whoisfreaks/src/__pycache__/client.cpython-312.pyc new file mode 100644 index 00000000000..ad5d9a0d575 Binary files /dev/null and b/internal-enrichment/whoisfreaks/src/__pycache__/client.cpython-312.pyc differ diff --git a/internal-enrichment/whoisfreaks/src/__pycache__/configVariables.cpython-312.pyc b/internal-enrichment/whoisfreaks/src/__pycache__/configVariables.cpython-312.pyc new file mode 100644 index 00000000000..04ad16517e2 Binary files /dev/null and b/internal-enrichment/whoisfreaks/src/__pycache__/configVariables.cpython-312.pyc differ diff --git a/internal-enrichment/whoisfreaks/src/__pycache__/config_variables.cpython-312.pyc b/internal-enrichment/whoisfreaks/src/__pycache__/config_variables.cpython-312.pyc new file mode 100644 index 00000000000..411c3603fac Binary files /dev/null and b/internal-enrichment/whoisfreaks/src/__pycache__/config_variables.cpython-312.pyc differ diff --git a/internal-enrichment/whoisfreaks/src/__pycache__/main.cpython-312.pyc b/internal-enrichment/whoisfreaks/src/__pycache__/main.cpython-312.pyc new file mode 100644 index 00000000000..49354211ac4 Binary files /dev/null and b/internal-enrichment/whoisfreaks/src/__pycache__/main.cpython-312.pyc differ diff --git a/internal-enrichment/whoisfreaks/src/builder.py b/internal-enrichment/whoisfreaks/src/builder.py new file mode 100644 index 00000000000..11ace2807de --- /dev/null +++ b/internal-enrichment/whoisfreaks/src/builder.py @@ -0,0 +1,427 @@ +import logging +from typing import Any, Dict, List, Optional + +import pycti +import stix2 + +logger = logging.getLogger(__name__) + + +class WhoisFreaksStixBuilder: + """ + A builder for creating valid STIX 2.1 bundles from WhoisFreaks API responses. + """ + + def __init__(self, author_name: str = "WhoisFreaks"): + self.author = stix2.Identity( + id=pycti.Identity.generate_id( + name=author_name, identity_class="organization" + ), + name=author_name, + identity_class="organization", + description="WhoisFreaks Domain & IP Threat Intelligence Provider.", + ) + + def build_whois_bundle( + self, domain_name: str, whois_data: Dict[str, Any] + ) -> Optional[stix2.Bundle]: + if not whois_data: + return None + + objects: List[Any] = [self.author] + clean_domain = domain_name.strip().lower() + domain_stix = stix2.DomainName(value=clean_domain) + objects.append(domain_stix) + + records = whois_data.get("whois_domains_historical") or [whois_data] + + # Registrar + registrar_name = next( + ( + rec.get("domain_registrar", {}).get("registrar_name") + for rec in records + if isinstance(rec, dict) and rec.get("domain_registrar") + ), + whois_data.get("registrar_name"), + ) + + if registrar_name and str(registrar_name).strip(): + registrar_stix = stix2.Identity( + id=pycti.Identity.generate_id( + name=str(registrar_name).strip(), identity_class="organization" + ), + name=str(registrar_name).strip(), + identity_class="organization", + created_by_ref=self.author.id, + ) + objects.append(registrar_stix) + objects.append( + stix2.Relationship( + id=pycti.StixCoreRelationship.generate_id( + source_ref=domain_stix.id, + relationship_type="registered-by", + target_ref=registrar_stix.id, + ), + source_ref=domain_stix.id, + target_ref=registrar_stix.id, + relationship_type="registered-by", + created_by_ref=self.author.id, + ) + ) + + # Registrant + registrant_data = next( + ( + rec.get("registrant_contact", {}) + for rec in records + if isinstance(rec, dict) and rec.get("registrant_contact") + ), + whois_data.get("registrant", {}), + ) + + registrant_name = ( + registrant_data.get("name") + if isinstance(registrant_data, dict) + else str(registrant_data) + ) + if registrant_name and not self.is_privacy_protected(str(registrant_name)): + registrant_stix = stix2.Identity( + id=pycti.Identity.generate_id( + name=str(registrant_name).strip(), + identity_class=( + "organization" + if isinstance(registrant_data, dict) + and registrant_data.get("company") + else "individual" + ), + ), + name=str(registrant_name).strip(), + identity_class=( + "organization" + if isinstance(registrant_data, dict) + and registrant_data.get("company") + else "individual" + ), + created_by_ref=self.author.id, + ) + objects.append(registrant_stix) + objects.append( + stix2.Relationship( + id=pycti.StixCoreRelationship.generate_id( + source_ref=domain_stix.id, + relationship_type="owned-by", + target_ref=registrant_stix.id, + ), + source_ref=domain_stix.id, + target_ref=registrant_stix.id, + relationship_type="owned-by", + created_by_ref=self.author.id, + ) + ) + + # Name Servers (Domain -> Domain uses 'related-to') + for ns in whois_data.get("name_servers", []): + if ns and str(ns).strip() and str(ns).strip().lower() != clean_domain: + ns_stix = stix2.DomainName(value=str(ns).strip().lower()) + objects.append(ns_stix) + objects.append( + stix2.Relationship( + id=pycti.StixCoreRelationship.generate_id( + source_ref=domain_stix.id, + relationship_type="related-to", + target_ref=ns_stix.id, + ), + source_ref=domain_stix.id, + target_ref=ns_stix.id, + relationship_type="related-to", + created_by_ref=self.author.id, + ) + ) + + return stix2.Bundle(objects=objects, allow_custom=True) + + def build_dns_bundle( + self, domain_or_ip: str, dns_data: Dict[str, Any] + ) -> Optional[stix2.Bundle]: + if not dns_data: + return None + + objects: List[Any] = [self.author] + clean_value = domain_or_ip.strip().lower() + + # Unified source observable handler + if self.is_ip_address(clean_value): + source_stix = ( + stix2.IPv6Address(value=clean_value) + if ":" in clean_value + else stix2.IPv4Address(value=clean_value) + ) + else: + source_stix = stix2.DomainName(value=clean_value) + objects.append(source_stix) + + dns_records = dns_data.get("dns_records") or dns_data.get("dnsRecords") or [] + if not dns_records and "historicalDnsRecords" in dns_data: + hist = dns_data["historicalDnsRecords"] + if hist and isinstance(hist, list): + dns_records = hist[0].get("dnsRecords", []) + + for record in dns_records: + record_type = (record.get("dnsType") or record.get("type", "")).upper() + record_value = record.get("address") or record.get("value") + if not record_type or not record_value: + continue + + if record_type == "A": + target_stix = stix2.IPv4Address(value=record_value.strip()) + rel_type = "resolves-to" + elif record_type == "AAAA": + target_stix = stix2.IPv6Address(value=record_value.strip()) + rel_type = "resolves-to" + elif record_type in ["CNAME", "MX", "NS"]: + target_stix = stix2.DomainName(value=record_value.strip().lower()) + rel_type = "related-to" + else: + continue + + # Skip self-resolving records to avoid self-referential relationships + if target_stix.id == source_stix.id: + continue + + objects.append(target_stix) + objects.append( + stix2.Relationship( + id=pycti.StixCoreRelationship.generate_id( + source_ref=source_stix.id, + relationship_type=rel_type, + target_ref=target_stix.id, + ), + source_ref=source_stix.id, + target_ref=target_stix.id, + relationship_type=rel_type, + created_by_ref=self.author.id, + ) + ) + + return stix2.Bundle(objects=objects, allow_custom=True) + + def build_ssl_bundle( + self, target: str, ssl_data: Dict[str, Any] + ) -> Optional[stix2.Bundle]: + if not ssl_data: + return None + + objects: List[Any] = [self.author] + clean_target = target.strip().lower() + + if self.is_ip_address(clean_target): + target_stix = ( + stix2.IPv6Address(value=clean_target) + if ":" in clean_target + else stix2.IPv4Address(value=clean_target) + ) + else: + target_stix = stix2.DomainName(value=clean_target) + objects.append(target_stix) + ssl_certificates = ssl_data.get("sslCertificates") + if not ssl_certificates: + return None + for cert in ssl_certificates: + cert_info = cert.get("certificate_info") or cert + issuer = cert_info.get("issuer_dn") or "Unknown Issuer" + subject = cert_info.get("subject_dn") or clean_target + serial = cert_info.get("serial_number") + cert_kwargs = {"issuer": str(issuer), "subject": str(subject)} + if serial: + cert_kwargs["serial_number"] = str(serial) + cert_stix = stix2.X509Certificate(**cert_kwargs) + objects.append(cert_stix) + objects.append( + stix2.Relationship( + id=pycti.StixCoreRelationship.generate_id( + source_ref=target_stix.id, + relationship_type="related-to", + target_ref=cert_stix.id, + ), + source_ref=target_stix.id, + target_ref=cert_stix.id, + relationship_type="related-to", + created_by_ref=self.author.id, + ) + ) + + return stix2.Bundle(objects=objects, allow_custom=True) + + def build_ip_geolocation_bundle( + self, ip_address: str, geolocation_data: Dict[str, Any] + ) -> Optional[stix2.Bundle]: + if not geolocation_data: + return None + + objects: List[Any] = [self.author] + clean_ip = ip_address.strip() + ip_stix = ( + stix2.IPv6Address(value=clean_ip) + if ":" in clean_ip + else stix2.IPv4Address(value=clean_ip) + ) + objects.append(ip_stix) + + loc_data = geolocation_data.get("location") or geolocation_data + country = loc_data.get("country_name") or loc_data.get("country_code") + city = loc_data.get("city") + + if country or city: + location_type = "City" if city else "Country" + location_name = city or country + location_stix = stix2.Location( + id=pycti.Location.generate_id(location_name, location_type), + name=location_name, + country=country, + city=city, + latitude=( + float(loc_data["latitude"]) if loc_data.get("latitude") else None + ), + longitude=( + float(loc_data["longitude"]) if loc_data.get("longitude") else None + ), + custom_properties={"x_opencti_location_type": location_type}, + created_by_ref=self.author.id, + ) + objects.append(location_stix) + objects.append( + stix2.Relationship( + id=pycti.StixCoreRelationship.generate_id( + source_ref=ip_stix.id, + relationship_type="located-at", + target_ref=location_stix.id, + ), + source_ref=ip_stix.id, + target_ref=location_stix.id, + relationship_type="located-at", + created_by_ref=self.author.id, + ) + ) + + return stix2.Bundle(objects=objects, allow_custom=True) + + def build_subdomains_bundle( + self, domain: str, subdomains_data: Dict[str, Any] + ) -> Optional[stix2.Bundle]: + if not subdomains_data: + return None + + objects: List[Any] = [self.author] + clean_domain = domain.strip().lower() + domain_stix = stix2.DomainName(value=clean_domain) + objects.append(domain_stix) + + subdomains = subdomains_data.get("subdomains", []) + for subdomain in subdomains: + sub_val = ( + subdomain.get("subdomain") + if isinstance(subdomain, dict) + else str(subdomain) + ) + if sub_val and sub_val.strip(): + sub_stix = stix2.DomainName(value=sub_val.strip().lower()) + objects.append(sub_stix) + objects.append( + stix2.Relationship( + id=pycti.StixCoreRelationship.generate_id( + source_ref=sub_stix.id, + relationship_type="related-to", + target_ref=domain_stix.id, + ), + source_ref=sub_stix.id, + target_ref=domain_stix.id, + relationship_type="related-to", + created_by_ref=self.author.id, + ) + ) + + return stix2.Bundle(objects=objects, allow_custom=True) + + def build_ip_reputation_bundle( + self, ip_address: str, reputation_data: Dict[str, Any] + ) -> Optional[stix2.Bundle]: + if not reputation_data: + return None + + objects: List[Any] = [self.author] + clean_ip = ip_address.strip() + ip_stix = ( + stix2.IPv6Address(value=clean_ip) + if ":" in clean_ip + else stix2.IPv4Address(value=clean_ip) + ) + objects.append(ip_stix) + + sec_data = reputation_data.get("security") or reputation_data + threat_score = sec_data.get("threat_score") + if threat_score is None: + threat_score = sec_data.get("score") + if threat_score is not None: + note_stix = stix2.Note( + id=pycti.Note.generate_id( + created=None, + content=f"Reputation Analysis for IP {clean_ip}: Threat Score: {threat_score}", + ), + abstract=f"WhoisFreaks Threat Score: {threat_score}", + content=f"Reputation Analysis for IP {clean_ip}:\nThreat Score: {threat_score}\nDetails: {sec_data}", + object_refs=[ip_stix.id], + created_by_ref=self.author.id, + ) + objects.append(note_stix) + + return stix2.Bundle(objects=objects, allow_custom=True) + + def build_domain_reputation_bundle( + self, domain: str, reputation_data: Dict[str, Any] + ) -> Optional[stix2.Bundle]: + if not reputation_data: + return None + + objects: List[Any] = [self.author] + clean_dom = domain.strip().lower() + dom_stix = stix2.DomainName(value=clean_dom) + objects.append(dom_stix) + + score = reputation_data.get("reputation_score") or reputation_data.get("score") + if score is not None: + note_stix = stix2.Note( + id=pycti.Note.generate_id( + created=None, + content=f"Domain Reputation Analysis for {clean_dom}: Score: {score}", + ), + abstract=f"WhoisFreaks Domain Reputation Score: {score}", + content=f"Domain Reputation Analysis for {clean_dom}:\nScore: {score}", + object_refs=[dom_stix.id], + created_by_ref=self.author.id, + ) + objects.append(note_stix) + + return stix2.Bundle(objects=objects, allow_custom=True) + + @staticmethod + def is_privacy_protected(name: str) -> bool: + privacy_indicators = [ + "privacy", + "protected", + "redacted", + "anonymous", + "whoisguard", + "contact privacy", + "data protected", + ] + return any(indicator in name.lower() for indicator in privacy_indicators) + + @staticmethod + def is_ip_address(value: str) -> bool: + import ipaddress + + try: + ipaddress.ip_address(value.strip()) + return True + except ValueError: + return False diff --git a/internal-enrichment/whoisfreaks/src/client.py b/internal-enrichment/whoisfreaks/src/client.py new file mode 100644 index 00000000000..6985d11991b --- /dev/null +++ b/internal-enrichment/whoisfreaks/src/client.py @@ -0,0 +1,257 @@ +import logging +from typing import Any, Dict, Optional + +import requests + +logger = logging.getLogger(__name__) + + +class WhoisFreaksClient: + """ + Client for interacting with WhoisFreaks REST APIs. + Docs: https://whoisfreaks.com/documentation + """ + + BASE_URL = "https://api.whoisfreaks.com" + + def __init__(self, api_key: str, timeout: Optional[int] = 30): + self.api_key = api_key + self.timeout = timeout + self.session = requests.Session() + self.session.headers.update( + { + "User-Agent": "OpenCTI-WhoisFreaks-Connector/1.0", + "Accept": "application/json", + } + ) + + def _post( + self, + endpoint: str, + params: Optional[dict] = None, + body: Optional[dict] = None, + timeout: Optional[int] = None, + ) -> Optional[Dict[str, Any]]: + url = f"{self.BASE_URL}{endpoint}" + if params is None: + params = {} + params["apiKey"] = self.api_key + if timeout is None: + timeout = self.timeout + + try: + response = self.session.post(url, params=params, json=body, timeout=timeout) + if response.status_code == 200: + return response.json() + else: + logger.error( + f"[WhoisFreaks API] POST {url} failed HTTP {response.status_code}: {response.text}" + ) + return None + except requests.exceptions.Timeout: + logger.error(f"[WhoisFreaks API] Timeout reaching {url}.") + return None + except requests.RequestException as e: + logger.error(f"[WhoisFreaks API] Error during POST {url}: {e}") + return None + + def _get( + self, endpoint: str, params: Optional[dict] = None, timeout: Optional[int] = None + ) -> Optional[Dict[str, Any]]: + url = f"{self.BASE_URL}{endpoint}" + if params is None: + params = {} + params["apiKey"] = self.api_key + if timeout is None: + timeout = self.timeout + + try: + response = self.session.get(url, params=params, timeout=timeout) + + if response.status_code == 200: + return response.json() + elif response.status_code == 404: + logger.info(f"[WhoisFreaks API] No record found at {endpoint}") + return None + elif response.status_code == 401: + logger.error( + "[WhoisFreaks API] Unauthorized - Check WHOISFREAKS_API_KEY." + ) + return None + elif response.status_code == 429: + logger.error( + "[WhoisFreaks API] Rate limit reached or insufficient credits." + ) + return None + else: + logger.error( + f"[WhoisFreaks API] GET {url} failed HTTP {response.status_code}: {response.text}" + ) + return None + except requests.exceptions.Timeout: + logger.error(f"[WhoisFreaks API] Timeout reaching {url}.") + return None + except requests.RequestException as e: + logger.error(f"[WhoisFreaks API] Error during GET {url}: {e}") + return None + + # --- API ENDPOINT WRAPPERS --- + def live_whois_lookup( + self, domain: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v2.0/whois/live", params={"format": format, "domainName": domain} + ) + + def historical_whois_lookup( + self, domain: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v2.0/whois", + params={"format": format, "domainName": domain, "whois": "historical"}, + ) + + def reverse_whois_lookup( + self, keyword: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v1.0/whois", + params={"format": format, "keyword": keyword, "whois": "reverse"}, + ) + + def bulk_whois_lookup( + self, domains: list[str], format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._post( + "/v2.0/bulkwhois/live", + params={"format": format}, + body={"domainNames": domains}, + ) + + def live_dns_lookup( + self, domain: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v2.0/dns/live", + params={"format": format, "domainName": domain, "type": "all"}, + ) + + def historical_dns_lookup( + self, domain: str, format: Optional[str] = "json", page: Optional[int] = 1 + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v2.0/dns/historical", + params={ + "format": format, + "domainName": domain, + "type": "all", + "page": page, + }, + ) + + def reverse_dns_lookup( + self, ip_address: str, format: Optional[str] = "json", page: Optional[int] = 1 + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v2.0/dns/reverse", + params={ + "format": format, + "value": ip_address, + "page": page, + "type": "a", + "exact": True, + }, + ) + + def bulk_dns_lookup( + self, domains: list[str], ip_addresses: list[str], format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._post( + "/v2.0/dns/bulk/live", + params={"format": format, "type": "all"}, + body={"domainNames": domains, "ipAddresses": ip_addresses}, + ) + + def domain_availability_lookup( + self, domain: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v1.0/domain/availability", + params={"format": format, "domainName": domain, "sug": False}, + ) + + def bulk_domain_availability_lookup( + self, domains: list[str], format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._post( + "/v1.0/domain/availability", + params={"format": format}, + body={"domainNames": domains}, + ) + + def typosquatting_lookup(self, keyword: str) -> Optional[Dict[str, Any]]: + return self._get("/v3.0/domain/typos", params={"keyword": keyword}) + + def ssl_lookup( + self, domain: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v1.0/ssl/live", + params={ + "format": format, + "domainName": domain, + "chain": True, + "sslRaw": False, + }, + ) + + def ip_geolocation_lookup( + self, ip_address: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v1.0/geolocation", params={"format": format, "ip": ip_address} + ) + + def bulk_ip_geolocation_lookup( + self, ip_addresses: list[str] + ) -> Optional[Dict[str, Any]]: + return self._post("/v1.0/geolocation", body={"ips": ip_addresses}) + + def subdomains_lookup( + self, domain: str, format: Optional[str] = "json", page: Optional[int] = 1 + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v1.0/subdomains", + params={"format": format, "domain": domain, "page": page}, + ) + + def ip_reputation_lookup(self, ip_address: str) -> Optional[Dict[str, Any]]: + return self._get("/v1.0/security", params={"ip": ip_address}) + + def bulk_ip_reputation_lookup( + self, ip_addresses: list[str] + ) -> Optional[Dict[str, Any]]: + return self._post("/v1.0/security", body={"ips": ip_addresses}) + + def asn_whois_lookup( + self, asn: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get("/v2.0/asn-whois", params={"asn": asn, "format": format}) + + def ip_whois_lookup( + self, ip_address: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get("/v1.0/ip-whois", params={"ip": ip_address, "format": format}) + + def domain_reputation_lookup( + self, domain: str, format: Optional[str] = "json" + ) -> Optional[Dict[str, Any]]: + return self._get( + "/v1.0/domain-reputation", params={"domainName": domain, "format": format} + ) + + def account_usage(self) -> Optional[Dict[str, Any]]: + return self._get("/v1.0/whoisapi/usage") + + def rotate_api_key(self) -> Optional[Dict[str, Any]]: + return self._post("/v1.0/api-key/rotate") diff --git a/internal-enrichment/whoisfreaks/src/config_variables.py b/internal-enrichment/whoisfreaks/src/config_variables.py new file mode 100644 index 00000000000..0cb4ebd6c2b --- /dev/null +++ b/internal-enrichment/whoisfreaks/src/config_variables.py @@ -0,0 +1,92 @@ +import sys +from pathlib import Path + +import yaml +from pycti import get_config_variable + + +class ConfigVariables: + """ + Configuration loader for WhoisFreaks OpenCTI Connector. + Reads settings from config.yml or environment variables. + """ + + def __init__(self): + + config_file_path = Path(__file__).parents[1] / "config.yml" + config = {} + + if config_file_path.is_file(): + with open(config_file_path, "r", encoding="utf-8") as file: + config = yaml.safe_load(file) or {} + + self.opencti_url = get_config_variable( + "OPENCTI_URL", ["opencti", "url"], config + ) + + self.opencti_token = get_config_variable( + "OPENCTI_TOKEN", ["opencti", "token"], config + ) + + self.connector_id = get_config_variable( + "CONNECTOR_ID", ["connector", "id"], config + ) + self.connector_type = get_config_variable( + "CONNECTOR_TYPE", + ["connector", "type"], + config, + default="INTERNAL_ENRICHMENT", + ) + self.connector_name = get_config_variable( + "CONNECTOR_NAME", + ["connector", "name"], + config, + default="WhoisFreaks", + ) + self.connector_scope = get_config_variable( + "CONNECTOR_SCOPE", + ["connector", "scope"], + config, + default="Domain-Name,IPv4-Addr,IPv6-Addr", + ) + self.connector_confidence_level = get_config_variable( + "CONNECTOR_CONFIDENCE_LEVEL", + ["connector", "confidence_level"], + config, + isNumber=True, + default=100, + ) + self.connector_auto = get_config_variable( + "CONNECTOR_AUTO", ["connector", "auto"], config, default=False + ) + # Force log level to uppercase string (e.g. 'INFO') + raw_log_level = get_config_variable( + "CONNECTOR_LOG_LEVEL", ["connector", "log_level"], config, default="info" + ) + self.connector_log_level = ( + str(raw_log_level).upper() if raw_log_level else "INFO" + ) + self.whoisfreaks_api_key = get_config_variable( + "WHOISFREAKS_API_KEY", ["whoisfreaks", "api_key"], config + ) + + self._validate() + + def _validate(self) -> None: + """Validates that all required configuration values are present.""" + missing = [] + if not self.opencti_url or self.opencti_url == "ChangeMe": + missing.append("OPENCTI_URL / opencti.url") + if not self.opencti_token or self.opencti_token == "ChangeMe": + missing.append("OPENCTI_TOKEN / opencti.token") + if not self.connector_id or self.connector_id == "ChangeMe": + missing.append("CONNECTOR_ID / connector.id") + if not self.whoisfreaks_api_key or self.whoisfreaks_api_key == "ChangeMe": + missing.append("WHOISFREAKS_API_KEY / whoisfreaks.api_key") + + if missing: + print( + f"[ERROR] Missing or default configuration values for: {', '.join(missing)}" + ) + print("[ERROR] Please update config.yml or pass environment variables.") + sys.exit(1) diff --git a/internal-enrichment/whoisfreaks/src/main.py b/internal-enrichment/whoisfreaks/src/main.py new file mode 100644 index 00000000000..4c8deef67ac --- /dev/null +++ b/internal-enrichment/whoisfreaks/src/main.py @@ -0,0 +1,182 @@ +import logging +import sys +from typing import Any, Dict, List, Optional, Tuple + +from builder import WhoisFreaksStixBuilder +from client import WhoisFreaksClient +from config_variables import ConfigVariables +from pycti import OpenCTIConnectorHelper + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + handlers=[logging.StreamHandler(sys.stdout)], +) +logger = logging.getLogger(__name__) + + +class WhoisFreaksConnector: + """ + OpenCTI Internal Enrichment Connector for WhoisFreaks. + Enriches Domain-Name and IPv4/IPv6 Observables with WHOIS, DNS, SSL, and Geolocation data. + """ + + def __init__(self): + logger.info("[WhoisFreaks Connector] Initializing configuration...") + self.config = ConfigVariables() + + logging.getLogger().setLevel(self.config.connector_log_level) + + self.helper = OpenCTIConnectorHelper( + { + "opencti": { + "url": self.config.opencti_url, + "token": self.config.opencti_token, + }, + "connector": { + "name": self.config.connector_name, + "id": self.config.connector_id, + "type": self.config.connector_type, + "scope": self.config.connector_scope, + "auto": self.config.connector_auto, + "log_level": self.config.connector_log_level, + "confidence_level": self.config.connector_confidence_level, + }, + }, + playbook_compatible=True, + ) + + self.client = WhoisFreaksClient(api_key=self.config.whoisfreaks_api_key) + self.builder = WhoisFreaksStixBuilder(author_name="WhoisFreaks") + + def _enrich_domain(self, domain_name: str) -> List[Any]: + """Executes all WhoisFreaks lookups for Domain-Name entities.""" + bundles = [] + + lookups = [ + (self.client.live_whois_lookup, self.builder.build_whois_bundle), + (self.client.live_dns_lookup, self.builder.build_dns_bundle), + (self.client.ssl_lookup, self.builder.build_ssl_bundle), + (self.client.subdomains_lookup, self.builder.build_subdomains_bundle), + ] + + for fetch_fn, build_fn in lookups: + resp = fetch_fn(domain_name) + if resp: + bundle = build_fn(domain_name, resp) + if bundle: + bundles.append(bundle) + + return bundles + + def _enrich_ip(self, ip_address: str) -> List[Any]: + """Executes all WhoisFreaks lookups for IP entities.""" + bundles = [] + + lookups = [ + ( + self.client.ip_geolocation_lookup, + self.builder.build_ip_geolocation_bundle, + ), + (self.client.ip_reputation_lookup, self.builder.build_ip_reputation_bundle), + (self.client.reverse_dns_lookup, self.builder.build_dns_bundle), + ] + + for fetch_fn, build_fn in lookups: + resp = fetch_fn(ip_address) + if resp: + bundle = build_fn(ip_address, resp) + if bundle: + bundles.append(bundle) + + return bundles + + def _get_entity_info(self, entity_id: str) -> Tuple[Optional[str], Optional[str]]: + """Reads entity from OpenCTI and returns entity type and observable value.""" + opencti_entity = self.helper.api.stix_cyber_observable.read(id=entity_id) + if not opencti_entity: + opencti_entity = self.helper.api.stix_domain_object.read(id=entity_id) + + if not opencti_entity: + return None, None + + obs_type = opencti_entity.get("entity_type") + obs_val = ( + opencti_entity.get("observable_value") + or opencti_entity.get("value") + or opencti_entity.get("name") + ) + return obs_type, obs_val + + def process_message(self, msg: Dict[str, Any]) -> str: + """Callback executed whenever an enrichment task is received from RabbitMQ.""" + entity_id = msg.get("entity_id") + if not entity_id: + logger.error("[WhoisFreaks Connector] Missing entity_id in message") + return "Entity ID missing" + observable_type, observable_value = self._get_entity_info(entity_id) + if not observable_type or not observable_value: + logger.error( + f"[WhoisFreaks Connector] Invalid or missing entity for ID: {entity_id}" + ) + return "Entity or value missing" + + logger.info( + f"[WhoisFreaks Connector] Processing enrichment for {observable_type}: '{observable_value}'" + ) + + work_id = self.helper.api.work.initiate_work( + connector_id=self.config.connector_id, + friendly_name=f"WhoisFreaks enrichment for {observable_value}", + ) + + try: + if observable_type == "Domain-Name": + bundles = self._enrich_domain(observable_value) + elif observable_type in ["IPv4-Addr", "IPv6-Addr"]: + bundles = self._enrich_ip(observable_value) + else: + logger.warning( + f"[WhoisFreaks Connector] Unsupported type: {observable_type}" + ) + self.helper.api.work.to_processed( + work_id, "Unsupported observable type" + ) + return "Unsupported observable type" + + if bundles: + for bundle in bundles: + self.helper.send_stix2_bundle( + bundle=bundle.serialize(), + work_id=work_id, + ) + + message = f"Successfully enriched {observable_value} with {len(bundles)} STIX bundles." + logger.info(f"[WhoisFreaks Connector] {message}") + self.helper.api.work.to_processed(work_id, message) + return message + + message = f"No threat intelligence data found on WhoisFreaks for {observable_value}." + logger.info(f"[WhoisFreaks Connector] {message}") + self.helper.api.work.to_processed(work_id, message) + return message + + except Exception as e: + error_msg = f"Error during processing of {observable_value}: {str(e)}" + logger.exception(f"[WhoisFreaks Connector] {error_msg}") + self.helper.api.work.to_processed(work_id, error_msg, in_error=True) + return error_msg + + def start(self): + """Starts the connector worker and listens to RabbitMQ queue.""" + logger.info("[WhoisFreaks Connector] Starting connector listener loop...") + self.helper.listen(self.process_message) + + +if __name__ == "__main__": # pragma: no cover + try: + connector = WhoisFreaksConnector() + connector.start() + except Exception as e: + logger.fatal(f"[WhoisFreaks Connector] Unhandled startup failure: {str(e)}") + sys.exit(1) diff --git a/internal-enrichment/whoisfreaks/tests/conftest.py b/internal-enrichment/whoisfreaks/tests/conftest.py new file mode 100644 index 00000000000..3f0bef63e56 --- /dev/null +++ b/internal-enrichment/whoisfreaks/tests/conftest.py @@ -0,0 +1,7 @@ +import sys +from pathlib import Path + +# Add the 'src' directory to sys.path so tests can import modules directly +src_path = Path(__file__).resolve().parent.parent / "src" +if str(src_path) not in sys.path: + sys.path.insert(0, str(src_path)) diff --git a/internal-enrichment/whoisfreaks/tests/test-requirements.txt b/internal-enrichment/whoisfreaks/tests/test-requirements.txt new file mode 100644 index 00000000000..19740d26d95 --- /dev/null +++ b/internal-enrichment/whoisfreaks/tests/test-requirements.txt @@ -0,0 +1,3 @@ +-r ../requirements.txt +pytest==9.0.3 +requests-mock diff --git a/internal-enrichment/whoisfreaks/tests/test_builder.py b/internal-enrichment/whoisfreaks/tests/test_builder.py new file mode 100644 index 00000000000..4f729335710 --- /dev/null +++ b/internal-enrichment/whoisfreaks/tests/test_builder.py @@ -0,0 +1,274 @@ +import pytest +import stix2 +from builder import WhoisFreaksStixBuilder + + +@pytest.fixture +def builder(): + return WhoisFreaksStixBuilder() + + +def test_builder_init(builder): + assert builder.author.name == "WhoisFreaks" + assert builder.author.identity_class == "organization" + + +def test_is_privacy_protected(): + assert WhoisFreaksStixBuilder.is_privacy_protected("WHOISGuard Protected") is True + assert WhoisFreaksStixBuilder.is_privacy_protected("REDACTED FOR PRIVACY") is True + assert WhoisFreaksStixBuilder.is_privacy_protected("John Doe") is False + + +def test_is_ip_address(): + assert WhoisFreaksStixBuilder.is_ip_address("1.2.3.4") is True + assert WhoisFreaksStixBuilder.is_ip_address("2001:db8::1") is True + assert WhoisFreaksStixBuilder.is_ip_address("example.com") is False + assert WhoisFreaksStixBuilder.is_ip_address(" 1.2.3.4 ") is True + + +def test_build_whois_bundle_empty(builder): + assert builder.build_whois_bundle("example.com", {}) is None + + +def test_build_whois_bundle_full(builder): + whois_data = { + "registrar_name": "GoDaddy.com, LLC", + "registrant": {"name": "John Doe", "company": "Example Inc."}, + "name_servers": ["ns1.example.com", "ns2.example.com"], + } + bundle = builder.build_whois_bundle("example.com", whois_data) + assert isinstance(bundle, stix2.Bundle) + + # Extract objects from the bundle + objects = {obj["type"]: obj for obj in bundle.objects} + assert "identity" in objects + assert "domain-name" in objects + + # Check registrar and registrant identities + identities = [obj for obj in bundle.objects if obj["type"] == "identity"] + names = [id_obj.name for id_obj in identities] + assert "WhoisFreaks" in names + assert "GoDaddy.com, LLC" in names + assert "John Doe" in names + + # Check relationships + relationships = [obj for obj in bundle.objects if obj["type"] == "relationship"] + rel_types = [rel.relationship_type for rel in relationships] + assert "registered-by" in rel_types + assert "owned-by" in rel_types + assert "related-to" in rel_types + + +def test_build_whois_bundle_privacy_protected(builder): + whois_data = { + "registrar_name": "GoDaddy.com, LLC", + "registrant": {"name": "REDACTED FOR PRIVACY", "company": "Privacy Inc."}, + } + bundle = builder.build_whois_bundle("example.com", whois_data) + assert isinstance(bundle, stix2.Bundle) + + identities = [obj for obj in bundle.objects if obj["type"] == "identity"] + names = [id_obj.name for id_obj in identities] + assert "WhoisFreaks" in names + assert "GoDaddy.com, LLC" in names + # REDACTED FOR PRIVACY should be filtered out + assert "REDACTED FOR PRIVACY" not in names + + relationships = [obj for obj in bundle.objects if obj["type"] == "relationship"] + rel_types = [rel.relationship_type for rel in relationships] + assert "registered-by" in rel_types + assert "owned-by" not in rel_types + + +def test_build_dns_bundle_empty(builder): + assert builder.build_dns_bundle("example.com", {}) is None + + +def test_build_dns_bundle_ipv4_source(builder): + dns_data = { + "dns_records": [ + {"type": "A", "address": "1.2.3.4"}, + {"type": "AAAA", "address": "2001:db8::1"}, + {"type": "CNAME", "value": "alias.example.com"}, + {"type": "MX", "value": "mail.example.com"}, + {"type": "NS", "value": "ns.example.com"}, + { + "type": "TXT", + "value": "v=spf1 ...", + }, # Unsupported type, should be skipped + ] + } + bundle = builder.build_dns_bundle("1.2.3.4", dns_data) + assert isinstance(bundle, stix2.Bundle) + + # Check source observable + ipv4s = [obj for obj in bundle.objects if obj["type"] == "ipv4-addr"] + assert len(ipv4s) == 1 # Only the source "1.2.3.4" (self-resolve records are skipped) + + ipv6s = [obj for obj in bundle.objects if obj["type"] == "ipv6-addr"] + assert len(ipv6s) == 1 + assert ipv6s[0].value == "2001:db8::1" + + domains = [obj for obj in bundle.objects if obj["type"] == "domain-name"] + domain_vals = [d.value for d in domains] + assert "alias.example.com" in domain_vals + assert "mail.example.com" in domain_vals + assert "ns.example.com" in domain_vals + + relationships = [obj for obj in bundle.objects if obj["type"] == "relationship"] + rel_types = [rel.relationship_type for rel in relationships] + assert "resolves-to" in rel_types + assert "related-to" in rel_types + + +def test_build_dns_bundle_historical(builder): + dns_data = { + "historicalDnsRecords": [ + {"dnsRecords": [{"dnsType": "A", "address": "5.6.7.8"}]} + ] + } + bundle = builder.build_dns_bundle("example.com", dns_data) + assert isinstance(bundle, stix2.Bundle) + + ipv4s = [obj for obj in bundle.objects if obj["type"] == "ipv4-addr"] + assert len(ipv4s) == 1 + assert ipv4s[0].value == "5.6.7.8" + + +def test_build_ssl_bundle_empty(builder): + assert builder.build_ssl_bundle("example.com", {}) is None + + +def test_build_ssl_bundle_success(builder): + ssl_data = { + "sslCertificates": [ + { + "certificate_info": { + "issuer_dn": "CN=DigiCert, O=DigiCert Inc", + "subject_dn": "CN=example.com, O=Example Org", + "serial_number": "1234567890", + } + } + ] + } + bundle = builder.build_ssl_bundle("example.com", ssl_data) + assert isinstance(bundle, stix2.Bundle) + + certs = [obj for obj in bundle.objects if obj["type"] == "x509-certificate"] + assert len(certs) == 1 + assert certs[0].issuer == "CN=DigiCert, O=DigiCert Inc" + assert certs[0].subject == "CN=example.com, O=Example Org" + assert certs[0].serial_number == "1234567890" + + relationships = [obj for obj in bundle.objects if obj["type"] == "relationship"] + assert len(relationships) == 1 + assert relationships[0].relationship_type == "related-to" + + +def test_build_ip_geolocation_bundle_empty(builder): + assert builder.build_ip_geolocation_bundle("1.2.3.4", {}) is None + + +def test_build_ip_geolocation_bundle_success(builder): + geo_data = { + "location": { + "country_name": "United States", + "country_code": "US", + "city": "Austin", + "latitude": "30.2672", + "longitude": "-97.7431", + } + } + bundle = builder.build_ip_geolocation_bundle("1.2.3.4", geo_data) + assert isinstance(bundle, stix2.Bundle) + + locations = [obj for obj in bundle.objects if obj["type"] == "location"] + assert len(locations) == 1 + assert locations[0].name == "Austin" + assert locations[0].country == "United States" + assert locations[0].city == "Austin" + assert locations[0].latitude == 30.2672 + assert locations[0].longitude == -97.7431 + + relationships = [obj for obj in bundle.objects if obj["type"] == "relationship"] + assert len(relationships) == 1 + assert relationships[0].relationship_type == "located-at" + + +def test_build_subdomains_bundle_empty(builder): + assert builder.build_subdomains_bundle("example.com", {}) is None + + +def test_build_subdomains_bundle_success(builder): + sub_data = { + "subdomains": [ + {"subdomain": "www.example.com"}, + {"subdomain": "api.example.com"}, + ] + } + bundle = builder.build_subdomains_bundle("example.com", sub_data) + assert isinstance(bundle, stix2.Bundle) + + domains = [obj for obj in bundle.objects if obj["type"] == "domain-name"] + domain_vals = [d.value for d in domains] + assert "example.com" in domain_vals + assert "www.example.com" in domain_vals + assert "api.example.com" in domain_vals + + relationships = [obj for obj in bundle.objects if obj["type"] == "relationship"] + assert len(relationships) == 2 + assert relationships[0].relationship_type == "related-to" + + +def test_build_ip_reputation_bundle_empty(builder): + assert builder.build_ip_reputation_bundle("1.2.3.4", {}) is None + + +def test_build_ip_reputation_bundle_success(builder): + rep_data = {"security": {"threat_score": 85}} + bundle = builder.build_ip_reputation_bundle("1.2.3.4", rep_data) + assert isinstance(bundle, stix2.Bundle) + + notes = [obj for obj in bundle.objects if obj["type"] == "note"] + assert len(notes) == 1 + assert "Threat Score: 85" in notes[0].abstract + assert "Threat Score: 85" in notes[0].content + + +def test_build_domain_reputation_bundle_empty(builder): + assert builder.build_domain_reputation_bundle("example.com", {}) is None + + +def test_build_domain_reputation_bundle_success(builder): + rep_data = {"reputation_score": 45} + bundle = builder.build_domain_reputation_bundle("example.com", rep_data) + assert isinstance(bundle, stix2.Bundle) + + notes = [obj for obj in bundle.objects if obj["type"] == "note"] + assert len(notes) == 1 + assert "Score: 45" in notes[0].abstract + assert "Score: 45" in notes[0].content + +def test_build_dns_bundle_unsupported_record_type(builder): + """Triggers line 173 in builder.py by passing an unhandled DNS record type (e.g., TXT).""" + dns_data = { + "dns_records": [ + {"type": "TXT", "value": "v=spf1 include:_spf.google.com ~all"} + ] + } + bundle = builder.build_dns_bundle("example.com", dns_data) + assert bundle is not None + # Only Identity and Domain-Name objects are present; the TXT record was safely skipped + assert len(bundle.objects) == 2 + + +def test_build_ssl_bundle_empty_certs(builder): + """Triggers line 218 in builder.py by passing an empty sslCertificates list.""" + bundle = builder.build_ssl_bundle("example.com", {"sslCertificates": []}) + assert bundle is None + + +def test_is_ip_address_invalid_string(builder): + """Triggers line 363 in builder.py by passing a non-IP string to is_ip_address.""" + assert builder.is_ip_address("example.com") is False + assert builder.is_ip_address("not-an-ip-address") is False \ No newline at end of file diff --git a/internal-enrichment/whoisfreaks/tests/test_client.py b/internal-enrichment/whoisfreaks/tests/test_client.py new file mode 100644 index 00000000000..90cc41da6da --- /dev/null +++ b/internal-enrichment/whoisfreaks/tests/test_client.py @@ -0,0 +1,316 @@ +import pytest +import requests +import requests_mock +from client import WhoisFreaksClient + + +@pytest.fixture +def client(): + return WhoisFreaksClient(api_key="test-api-key", timeout=10) + + +def test_client_init(client): + assert client.api_key == "test-api-key" + assert client.timeout == 10 + assert client.session.headers["User-Agent"] == "OpenCTI-WhoisFreaks-Connector/1.0" + assert client.session.headers["Accept"] == "application/json" + + +def test_client_get_success(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois/live?apiKey=test-api-key&format=json&domainName=example.com", + json={"status": "success"}, + ) + res = client.live_whois_lookup("example.com") + assert res == {"status": "success"} + + +def test_client_get_404(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois/live?apiKey=test-api-key&format=json&domainName=example.com", + status_code=404, + ) + res = client.live_whois_lookup("example.com") + assert res is None + + +def test_client_get_401(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois/live?apiKey=test-api-key&format=json&domainName=example.com", + status_code=401, + ) + res = client.live_whois_lookup("example.com") + assert res is None + + +def test_client_get_429(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois/live?apiKey=test-api-key&format=json&domainName=example.com", + status_code=429, + ) + res = client.live_whois_lookup("example.com") + assert res is None + + +def test_client_get_500(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois/live?apiKey=test-api-key&format=json&domainName=example.com", + status_code=500, + text="Internal Server Error", + ) + res = client.live_whois_lookup("example.com") + assert res is None + + +def test_client_get_timeout(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois/live?apiKey=test-api-key&format=json&domainName=example.com", + exc=requests.exceptions.Timeout, + ) + res = client.live_whois_lookup("example.com") + assert res is None + + +def test_client_get_request_exception(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois/live?apiKey=test-api-key&format=json&domainName=example.com", + exc=requests.RequestException, + ) + res = client.live_whois_lookup("example.com") + assert res is None + + +def test_client_post_success(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v2.0/bulkwhois/live?apiKey=test-api-key&format=json", + json={"bulk": "success"}, + ) + res = client.bulk_whois_lookup(["example.com"]) + assert res == {"bulk": "success"} + assert m.last_request.json() == {"domainNames": ["example.com"]} + + +def test_client_post_error(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v2.0/bulkwhois/live?apiKey=test-api-key&format=json", + status_code=500, + ) + res = client.bulk_whois_lookup(["example.com"]) + assert res is None + + +def test_client_post_timeout(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v2.0/bulkwhois/live?apiKey=test-api-key&format=json", + exc=requests.exceptions.Timeout, + ) + res = client.bulk_whois_lookup(["example.com"]) + assert res is None + + +def test_client_post_exception(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v2.0/bulkwhois/live?apiKey=test-api-key&format=json", + exc=requests.RequestException, + ) + res = client.bulk_whois_lookup(["example.com"]) + assert res is None + + +# Tests for all endpoint wrappers to ensure correctness of URL mappings and HTTP verbs + + +def test_historical_whois_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/whois?apiKey=test-api-key&format=json&domainName=example.com&whois=historical", + json={"data": "whois"}, + ) + assert client.historical_whois_lookup("example.com") == {"data": "whois"} + + +def test_reverse_whois_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/whois?apiKey=test-api-key&format=json&keyword=example&whois=reverse", + json={"data": "rev"}, + ) + assert client.reverse_whois_lookup("example") == {"data": "rev"} + + +def test_live_dns_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/dns/live?apiKey=test-api-key&format=json&domainName=example.com&type=all", + json={"data": "dns"}, + ) + assert client.live_dns_lookup("example.com") == {"data": "dns"} + + +def test_historical_dns_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/dns/historical?apiKey=test-api-key&format=json&domainName=example.com&type=all&page=2", + json={"data": "hist"}, + ) + assert client.historical_dns_lookup("example.com", page=2) == {"data": "hist"} + + +def test_reverse_dns_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/dns/reverse?apiKey=test-api-key&format=json&value=1.2.3.4&page=1&type=a&exact=True", + json={"data": "rev"}, + ) + assert client.reverse_dns_lookup("1.2.3.4") == {"data": "rev"} + + +def test_bulk_dns_lookup(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v2.0/dns/bulk/live?apiKey=test-api-key&format=json&type=all", + json={"data": "bulk"}, + ) + assert ( + client.bulk_dns_lookup(domains=["example.com"], ip_addresses=["1.2.3.4"]) + == {"data": "bulk"} + ) + + +def test_domain_availability_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/domain/availability?apiKey=test-api-key&format=json&domainName=example.com&sug=False", + json={"data": "avail"}, + ) + assert client.domain_availability_lookup("example.com") == {"data": "avail"} + + +def test_bulk_domain_availability_lookup(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v1.0/domain/availability?apiKey=test-api-key&format=json", + json={"data": "bulk_avail"}, + ) + assert client.bulk_domain_availability_lookup(["example.com"]) == { + "data": "bulk_avail" + } + + +def test_typosquatting_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v3.0/domain/typos?apiKey=test-api-key&keyword=example", + json={"data": "typo"}, + ) + assert client.typosquatting_lookup("example") == {"data": "typo"} + + +def test_ssl_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/ssl/live?apiKey=test-api-key&format=json&domainName=example.com&chain=True&sslRaw=False", + json={"data": "ssl"}, + ) + assert client.ssl_lookup("example.com") == {"data": "ssl"} + + +def test_ip_geolocation_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/geolocation?apiKey=test-api-key&format=json&ip=1.2.3.4", + json={"data": "geo"}, + ) + assert client.ip_geolocation_lookup("1.2.3.4") == {"data": "geo"} + + +def test_bulk_ip_geolocation_lookup(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v1.0/geolocation?apiKey=test-api-key", + json={"data": "bulk_geo"}, + ) + assert client.bulk_ip_geolocation_lookup(["1.2.3.4"]) == {"data": "bulk_geo"} + + +def test_subdomains_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/subdomains?apiKey=test-api-key&format=json&domain=example.com&page=2", + json={"data": "sub"}, + ) + assert client.subdomains_lookup("example.com", page=2) == {"data": "sub"} + + +def test_ip_reputation_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/security?apiKey=test-api-key&ip=1.2.3.4", + json={"data": "rep"}, + ) + assert client.ip_reputation_lookup("1.2.3.4") == {"data": "rep"} + + +def test_bulk_ip_reputation_lookup(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v1.0/security?apiKey=test-api-key", + json={"data": "bulk_rep"}, + ) + assert client.bulk_ip_reputation_lookup(["1.2.3.4"]) == {"data": "bulk_rep"} + + +def test_asn_whois_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v2.0/asn-whois?apiKey=test-api-key&asn=AS15169&format=json", + json={"data": "asn"}, + ) + assert client.asn_whois_lookup("AS15169") == {"data": "asn"} + + +def test_ip_whois_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/ip-whois?apiKey=test-api-key&ip=1.2.3.4&format=json", + json={"data": "ip_whois"}, + ) + assert client.ip_whois_lookup("1.2.3.4") == {"data": "ip_whois"} + + +def test_domain_reputation_lookup(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/domain-reputation?apiKey=test-api-key&domainName=example.com&format=json", + json={"data": "dom_rep"}, + ) + assert client.domain_reputation_lookup("example.com") == {"data": "dom_rep"} + + +def test_account_usage(client): + with requests_mock.Mocker() as m: + m.get( + "https://api.whoisfreaks.com/v1.0/whoisapi/usage?apiKey=test-api-key", + json={"data": "usage"}, + ) + assert client.account_usage() == {"data": "usage"} + + +def test_rotate_api_key(client): + with requests_mock.Mocker() as m: + m.post( + "https://api.whoisfreaks.com/v1.0/api-key/rotate?apiKey=test-api-key", + json={"data": "rotate"}, + ) + assert client.rotate_api_key() == {"data": "rotate"} diff --git a/internal-enrichment/whoisfreaks/tests/test_config_variables.py b/internal-enrichment/whoisfreaks/tests/test_config_variables.py new file mode 100644 index 00000000000..d3081969dff --- /dev/null +++ b/internal-enrichment/whoisfreaks/tests/test_config_variables.py @@ -0,0 +1,61 @@ +import pytest +from config_variables import ConfigVariables + + +@pytest.fixture +def valid_env(monkeypatch): + """Provides a complete set of valid environment variables.""" + env = { + "OPENCTI_URL": "http://localhost:8080", + "OPENCTI_TOKEN": "test-token-uuid-1234", + "CONNECTOR_ID": "whoisfreaks-connector-id-1234", + "CONNECTOR_NAME": "WhoisFreaks", + "CONNECTOR_SCOPE": "Domain-Name,IPv4-Addr,IPv6-Addr", + "CONNECTOR_LOG_LEVEL": "info", + "WHOISFREAKS_API_KEY": "test_whoisfreaks_api_key_xyz", + } + for key, value in env.items(): + monkeypatch.setenv(key, value) + return env + + +def test_config_variables_success(valid_env): + """Tests successful instantiation when all environment variables are present.""" + config = ConfigVariables() + assert config.opencti_url == "http://localhost:8080" + assert config.opencti_token == "test-token-uuid-1234" + assert config.whoisfreaks_api_key == "test_whoisfreaks_api_key_xyz" + + +def test_config_variables_missing_opencti_url(valid_env, monkeypatch): + """Executes validation failure lines when OPENCTI_URL is set to default/invalid value.""" + monkeypatch.setenv("OPENCTI_URL", "ChangeMe") + with pytest.raises(SystemExit): + ConfigVariables() + + +def test_config_variables_missing_opencti_token(valid_env, monkeypatch): + """Executes validation failure lines when OPENCTI_TOKEN is set to default/invalid value.""" + monkeypatch.setenv("OPENCTI_TOKEN", "ChangeMe") + with pytest.raises(SystemExit): + ConfigVariables() + + +def test_config_variables_missing_api_key(valid_env, monkeypatch): + """Executes validation failure lines when WHOISFREAKS_API_KEY is set to default/invalid value.""" + monkeypatch.setenv("WHOISFREAKS_API_KEY", "ChangeMe") + with pytest.raises(SystemExit): + ConfigVariables() + + +def test_config_variables_empty_values(valid_env, monkeypatch): + """Executes validation checks when required variables are empty strings.""" + monkeypatch.setenv("WHOISFREAKS_API_KEY", "") + with pytest.raises(SystemExit): + ConfigVariables() + +def test_config_variables_missing_connector_id(valid_env, monkeypatch): + """Executes validation failure lines when CONNECTOR_ID is set to ChangeMe.""" + monkeypatch.setenv("CONNECTOR_ID", "ChangeMe") + with pytest.raises(SystemExit): + ConfigVariables() \ No newline at end of file diff --git a/internal-enrichment/whoisfreaks/tests/test_connector.py b/internal-enrichment/whoisfreaks/tests/test_connector.py new file mode 100644 index 00000000000..0964bcba5df --- /dev/null +++ b/internal-enrichment/whoisfreaks/tests/test_connector.py @@ -0,0 +1,208 @@ +from unittest.mock import MagicMock + +import main +import pytest +from main import WhoisFreaksConnector + + +@pytest.fixture +def mock_dependencies(monkeypatch): + # Mock ConfigVariables + mock_cfg = MagicMock() + mock_cfg.opencti_url = "http://localhost:8080" + mock_cfg.opencti_token = "mock-token" + mock_cfg.connector_id = "mock-connector-id" + mock_cfg.connector_type = "INTERNAL_ENRICHMENT" + mock_cfg.connector_name = "WhoisFreaks" + mock_cfg.connector_scope = "Domain-Name,IPv4-Addr,IPv6-Addr" + mock_cfg.connector_auto = False + mock_cfg.connector_log_level = "INFO" + mock_cfg.connector_confidence_level = 100 + mock_cfg.whoisfreaks_api_key = "mock-api-key" + + # Mock OpenCTIConnectorHelper + mock_h = MagicMock() + mock_h.api.work.initiate_work.return_value = "mock-work-id" + + # Apply monkeypatch mocks + monkeypatch.setattr(main, "ConfigVariables", MagicMock(return_value=mock_cfg)) + monkeypatch.setattr(main, "OpenCTIConnectorHelper", MagicMock(return_value=mock_h)) + + return mock_cfg, mock_h + + +def test_connector_init(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + assert connector.config == cfg + assert connector.helper == helper + assert connector.client.api_key == "mock-api-key" + assert connector.builder.author.name == "WhoisFreaks" + + +def test_get_entity_info_cyber_observable(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + + # Mock cyber observable read + helper.api.stix_cyber_observable.read.return_value = { + "entity_type": "Domain-Name", + "observable_value": "example.com", + } + + obs_type, obs_val = connector._get_entity_info("entity-123") + assert obs_type == "Domain-Name" + assert obs_val == "example.com" + helper.api.stix_cyber_observable.read.assert_called_once_with(id="entity-123") + + +def test_get_entity_info_domain_object(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + + # Mock cyber observable read returns None, domain object read returns entity + helper.api.stix_cyber_observable.read.return_value = None + helper.api.stix_domain_object.read.return_value = { + "entity_type": "Domain-Name", + "value": "domain-obj.com", + } + + obs_type, obs_val = connector._get_entity_info("entity-456") + assert obs_type == "Domain-Name" + assert obs_val == "domain-obj.com" + helper.api.stix_cyber_observable.read.assert_called_once_with(id="entity-456") + helper.api.stix_domain_object.read.assert_called_once_with(id="entity-456") + + +def test_get_entity_info_not_found(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + + helper.api.stix_cyber_observable.read.return_value = None + helper.api.stix_domain_object.read.return_value = None + + obs_type, obs_val = connector._get_entity_info("entity-789") + assert obs_type is None + assert obs_val is None + + +def test_process_message_empty_entity(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + connector._get_entity_info = MagicMock(return_value=(None, None)) + + result = connector.process_message({"entity_id": "entity-123"}) + assert result == "Entity or value missing" + + +def test_process_message_unsupported_type(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + connector._get_entity_info = MagicMock(return_value=("File", "test.exe")) + + result = connector.process_message({"entity_id": "entity-123"}) + assert result == "Unsupported observable type" + helper.api.work.to_processed.assert_called_once_with( + "mock-work-id", "Unsupported observable type" + ) + + +def test_process_message_domain_enrichment(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + connector._get_entity_info = MagicMock(return_value=("Domain-Name", "example.com")) + + # Mock client lookups to return valid dicts + connector.client.live_whois_lookup = MagicMock( + return_value={"registrar_name": "GoDaddy"} + ) + connector.client.live_dns_lookup = MagicMock( + return_value={"dns_records": [{"type": "A", "address": "1.2.3.4"}]} + ) + connector.client.ssl_lookup = MagicMock(return_value={"sslCertificates": []}) + connector.client.subdomains_lookup = MagicMock(return_value={"subdomains": []}) + + result = connector.process_message({"entity_id": "entity-123"}) + + assert "Successfully enriched example.com with" in result + assert helper.send_stix2_bundle.call_count == 3 + helper.api.work.to_processed.assert_called_once() + + +def test_process_message_domain_enrichment_no_data(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + connector._get_entity_info = MagicMock(return_value=("Domain-Name", "example.com")) + + # Mock client lookups to return None + connector.client.live_whois_lookup = MagicMock(return_value=None) + connector.client.live_dns_lookup = MagicMock(return_value=None) + connector.client.ssl_lookup = MagicMock(return_value=None) + connector.client.subdomains_lookup = MagicMock(return_value=None) + + result = connector.process_message({"entity_id": "entity-123"}) + + assert "No threat intelligence data found on WhoisFreaks for example.com." in result + helper.send_stix2_bundle.assert_not_called() + helper.api.work.to_processed.assert_called_once_with( + "mock-work-id", + "No threat intelligence data found on WhoisFreaks for example.com.", + ) + + +def test_process_message_ip_enrichment(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + connector._get_entity_info = MagicMock(return_value=("IPv4-Addr", "1.2.3.4")) + + # Mock client lookups + connector.client.ip_geolocation_lookup = MagicMock( + return_value={"location": {"country_name": "United States"}} + ) + connector.client.ip_reputation_lookup = MagicMock( + return_value={"security": {"threat_score": 85}} + ) + connector.client.reverse_dns_lookup = MagicMock(return_value={"dns_records": []}) + + result = connector.process_message({"entity_id": "entity-123"}) + + assert "Successfully enriched 1.2.3.4 with" in result + assert helper.send_stix2_bundle.call_count == 3 + helper.api.work.to_processed.assert_called_once() + + +def test_process_message_exception(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + connector._get_entity_info = MagicMock(return_value=("Domain-Name", "example.com")) + + # Cause lookup to raise exception + connector.client.live_whois_lookup = MagicMock(side_effect=Exception("API failure")) + + result = connector.process_message({"entity_id": "entity-123"}) + + assert "Error during processing of example.com: API failure" in result + helper.api.work.to_processed.assert_called_once_with( + "mock-work-id", + "Error during processing of example.com: API failure", + in_error=True, + ) + + +def test_connector_start(mock_dependencies): + cfg, helper = mock_dependencies + connector = WhoisFreaksConnector() + + connector.start() + helper.listen.assert_called_once_with(connector.process_message) + +def test_process_message_unsupported_observable_type(mock_dependencies, monkeypatch): + """Executes lines 115-116 in main.py by processing an unsupported observable type with a valid work_id.""" + from main import WhoisFreaksConnector + + connector = WhoisFreaksConnector() + monkeypatch.setattr( + connector, "_get_entity_info", lambda entity_id: ("StixFile", "sample.exe") + ) + result = connector.process_message({"entity_id": "file--1234", "work_id": "work-1234"}) + assert result == "Unsupported observable type" \ No newline at end of file diff --git a/internal-import-file/import-document-ai/src/import_doc_ai/client_api.py b/internal-import-file/import-document-ai/src/import_doc_ai/client_api.py index ac7724917ac..531ef9677d9 100644 --- a/internal-import-file/import-document-ai/src/import_doc_ai/client_api.py +++ b/internal-import-file/import-document-ai/src/import_doc_ai/client_api.py @@ -17,13 +17,20 @@ def __init__(self, helper, config: ConfigConnector): self.helper = helper self.config = config - self._opencti_instance_id = self.helper.api.query(""" + self._opencti_instance_id = ( + self.helper.api.query( + """ query SettingsQuery { settings { id } } - """).get("data", {}).get("settings", {}).get("id", "") + """ + ) + .get("data", {}) + .get("settings", {}) + .get("id", "") + ) # Define headers in session for legacy direct mode headers = {} diff --git a/shared/pylint_plugins/tests/test_check_stix_plugin/test_linter_stix_generator.py b/shared/pylint_plugins/tests/test_check_stix_plugin/test_linter_stix_generator.py index 3503c08c5b8..a92f1bf4841 100644 --- a/shared/pylint_plugins/tests/test_check_stix_plugin/test_linter_stix_generator.py +++ b/shared/pylint_plugins/tests/test_check_stix_plugin/test_linter_stix_generator.py @@ -93,7 +93,8 @@ def test_find_constructor_calls_should_detect_constructor_call(example_code): def test_find_constructor_calls_should_detect_multiple_constructor_calls(): # Given a Python script several calls to a STIX2 Domain Object constructors - module = parse(""" + module = parse( + """ import stix2 loc = stix2.Location( id=generate_id("name"), @@ -102,7 +103,8 @@ def test_find_constructor_calls_should_detect_multiple_constructor_calls(): author = stix2.Identity( name="author" ) - """) + """ + ) # When find_constructor_calls is called calls = list(find_constructor_calls(module, ["Location", "Identity"], "stix2")) @@ -150,10 +152,12 @@ def test_StixIdGeneratorChecker_should_raise_message_when_no_given_id_in_stix2_d self, ): # Given: A STIX2 domain object constructor call without an 'id' argument - call_node = extract_node(""" + call_node = extract_node( + """ from stix2 import Location loc = Location(name="example") #@ - """) + """ + ) # When: the checker visits the constructor call node with self.assertAddsMessages( @@ -172,12 +176,14 @@ def test_StixIdGeneratorChecker_should_not_raise_message_when_no_stix2_domain_ob self, ): # Given: A Python script with no STIX2 domain object constructor call - call_node = extract_node(""" + call_node = extract_node( + """ class Location: def __init__(self): pass loc = Location() #@ - """) + """ + ) # When: the checker visits the method without constructor calls with self.assertNoMessages(): @@ -187,10 +193,12 @@ def test_StixIdGeneratorChecker_should_not_raise_message_when_given_in_stix2_dom self, ): # Given: A STIX2 domain object constructor call with an 'id' argument - call_node = extract_node(""" + call_node = extract_node( + """ from stix2 import Location loc = Location(id=generate_id("example"), name="example") #@ - """) + """ + ) # When: the checker visits the constructor call node with an id argument # Then: No message is added