diff --git a/internal-enrichment/censys-enrichmentapis/.gitignore b/internal-enrichment/censys-enrichmentapis/.gitignore new file mode 100644 index 00000000000..226cd78fb7a --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/.gitignore @@ -0,0 +1,18 @@ +# Ignore Visual Studio Code settings and workspace files +.vscode +.claude +.venv + + +# folders to ignore +docs/ +src/censys_enrichmentapis/__pycache__ +src/censys_enrichmentapis/converters/__pycache__ +tests/censys_enrichmentapis/__pycache__ +tests/__pycache__ +.pytest_cache/ + +# configuration file +config.yml +src/config.yml +Taskfile.yml diff --git a/internal-enrichment/censys-enrichmentapis/Dockerfile b/internal-enrichment/censys-enrichmentapis/Dockerfile new file mode 100644 index 00000000000..527bcf73487 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/Dockerfile @@ -0,0 +1,30 @@ +ARG PYTHON_VERSION=3.12 +FROM python:${PYTHON_VERSION}-alpine AS base + +# Prevents Python from writing pyc files. +ENV PYTHONDONTWRITEBYTECODE=1 + +# Keeps Python from buffering stdout and stderr to avoid situations where +# the application crashes without emitting any logs due to buffering. +ENV PYTHONUNBUFFERED=1 + +# Install Python modules +# hadolint ignore=DL3003 +RUN apk update && apk upgrade && \ + apk --no-cache add git build-base libmagic libffi-dev libxml2-dev libxslt-dev + +# Copy the connector without local configuration or generated files +COPY --exclude=config.yml \ + --exclude=.DS_Store \ + --exclude=**/.DS_Store \ + --exclude=__pycache__ \ + --exclude=**/__pycache__ \ + src /opt/opencti-connector-censys-enrichmentapis +WORKDIR /opt/opencti-connector-censys-enrichmentapis + +# Install Python dependencies and remove build dependencies +RUN pip3 install --no-cache-dir -r requirements.txt && \ + apk del git build-base + +# Expose and entrypoint +ENTRYPOINT ["python", "main.py"] diff --git a/internal-enrichment/censys-enrichmentapis/README.md b/internal-enrichment/censys-enrichmentapis/README.md new file mode 100644 index 00000000000..19355e94da3 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/README.md @@ -0,0 +1,335 @@ +# OpenCTI Censys Connector + +| Status | Date | Comment | +|--------|------|---------| +| Community | - | - | + +The Censys EnrichmentAPIs connector enriches IP addresses, domains, and certificates with internet scanning data from the Censys Platform, providing geolocation, ASN, services, software, infrastructure information, host reputation score, host threat labels if present. + +## Table of Contents + +- [OpenCTI Censys Connector](#opencti-censys-connector) + - [Table of Contents](#table-of-contents) + - [Introduction](#introduction) + - [Installation](#installation) + - [Requirements](#requirements) + - [Configuration variables](#configuration-variables) + - [OpenCTI environment variables](#opencti-environment-variables) + - [Base connector environment variables](#base-connector-environment-variables) + - [Connector extra parameters environment variables](#connector-extra-parameters-environment-variables) + - [Deployment](#deployment) + - [Docker Deployment](#docker-deployment) + - [Manual Deployment](#manual-deployment) + - [Usage](#usage) + - [Behavior](#behavior) + - [Note Types](#note-types) + - [Debugging](#debugging) + - [Additional information](#additional-information) + +## Introduction + +Censys is an internet intelligence platform that continuously scans the global IPv4 address space and provides comprehensive data about internet-connected devices, services, and certificates. The Censys Search API offers detailed host information including open ports, running services, TLS certificates, geolocation, and autonomous system data. + +This connector integrates Censys Search with OpenCTI to enrich: +- **IP addresses** (IPv4/IPv6): Geolocation, ASN, services, software, hostnames, certificates, reputation score, threat labels. +- **Domain names**: Resolving hosts with all associated IP enrichment data +- **X509 Certificates**: Certificate metadata from Censys certificate database + +## Installation + +### Requirements + +- OpenCTI Platform >= 6.8.11 +- Censys account with API access (Organisation ID and Token) + +## Configuration variables + +There are a number of configuration options, which are set either in `docker-compose.yml` (for Docker), `.env` file, or in `config.yml` (for manual deployment). + +### OpenCTI environment variables + +| Parameter | config.yml | Docker environment variable | Mandatory | Description | +|---------------|------------|-----------------------------|-----------|------------------------------------------------------| +| OpenCTI URL | url | `OPENCTI_URL` | Yes | The URL of the OpenCTI platform. | +| OpenCTI Token | token | `OPENCTI_TOKEN` | Yes | The default admin token set in the OpenCTI platform. | + +### Base connector environment variables + +| Parameter | config.yml | Docker environment variable | Default | Mandatory | Description | +|-----------------|------------|-----------------------------|-----------------------------------------|-----------|------------------------------------------------------------------------------| +| Connector ID | id | `CONNECTOR_ID` | censys-enrichmentapis--674403d0-... | No | A unique `UUIDv4` identifier for this connector instance. | +| Connector Name | name | `CONNECTOR_NAME` | Censys EnrichmentAPIs | No | Name of the connector. | +| Connector Scope | scope | `CONNECTOR_SCOPE` | IPv4-Addr,IPv6-Addr,X509-Certificate,Domain-Name | No | The scope of observables the connector will enrich. | +| Connector Type | type | `CONNECTOR_TYPE` | INTERNAL_ENRICHMENT | Yes | Should always be `INTERNAL_ENRICHMENT` for this connector. | +| Log Level | log_level | `CONNECTOR_LOG_LEVEL` | error | No | Determines the verbosity of the logs: `debug`, `info`, `warn`, or `error`. | +| Auto Mode | auto | `CONNECTOR_AUTO` | false | No | Enables or disables automatic enrichment of observables. | + +### Connector extra parameters environment variables + +| Parameter | config.yml | Docker environment variable | Default | Mandatory | Description | +|-----------------|------------------------------------|-----------------------------------------|------------|-----------|--------------------------------------------------------------------| +| Organisation ID | censys_enrichment.organisation_id | `CENSYS_ENRICHMENT_ORGANISATION_ID` | | Yes | Your Censys organisation ID for API authentication. | +| API Token | censys_enrichment.token | `CENSYS_ENRICHMENT_TOKEN` | | Yes | Your Censys API token for authentication. | +| Max TLP | censys_enrichment.max_tlp | `CENSYS_ENRICHMENT_MAX_TLP` | TLP:AMBER | No | Maximum TLP level for observables to be enriched. | + +## Deployment + +### Docker Deployment + +Build the Docker image: + +```bash +docker build -t opencti/connector-censys-enrichmentapis:latest . +``` + +Configure the connector in `docker-compose.yml`: + +```yaml + connector-censys-enrichmentapis: + image: opencti/connector-censys-enrichmentapis:latest + environment: + - OPENCTI_URL=http://localhost + - OPENCTI_TOKEN=ChangeMe + - CONNECTOR_ID=ChangeMe_UUID4 + - CONNECTOR_NAME=Censys EnrichmentAPIs + - CONNECTOR_SCOPE=IPv4-Addr,IPv6-Addr,X509-Certificate,Domain-Name + - CONNECTOR_LOG_LEVEL=error + - CONNECTOR_AUTO=false + - CENSYS_ENRICHMENT_ORGANISATION_ID=ChangeMe + - CENSYS_ENRICHMENT_TOKEN=ChangeMe + - CENSYS_ENRICHMENT_MAX_TLP=TLP:AMBER + restart: always +``` + +Start the connector: + +```bash +docker compose up -d +``` + +### Manual Deployment + +1. Copy `src/config.yaml.sample` to `src/config.yml` and configure with your credentials. + +2. Install dependencies: + +```bash +pip3 install -r src/requirements.txt +``` + +3. Start the connector from the `src` directory: + +```bash +python3 main.py +``` + +## Usage + +The connector enriches IP addresses, domains, and certificates with Censys internet scanning data. + +**Observations → Observables** + +Select an IPv4-Addr, IPv6-Addr, Domain-Name, or X509-Certificate observable, then click the enrichment button and choose Censys EnrichmentAPIs. + +## Behavior + The connector enriches the following observable types: + + ### IPv4/IPv6 Addresses + - Retrieves host information including geolocation, ASN, services, and reputation + - Creates location entities (City, Country, Region, Administrative Area) + - Links autonomous systems and organizations + - Extracts DNS names associated with the IP + - Creates software entities for detected services + - Includes comprehensive service notes with: + - Service protocol and scan time + - Detected service labels (e.g., REMOTE_ACCESS) + - Associated threats and security information + - Generates reputation notes with: + - Host reputation score and risk level + - Model version information + + ### Domain Names + - Searches for hosts with the domain in their DNS records + - Creates IP address observables for discovered hosts + - **Discovers X.509 certificates** that reference the domain in their Subject Alternative Names (SANs) or Common Name (CN) + - Creates certificate entities with full metadata (issuer, validity, extensions) + - Links certificates to the domain for infrastructure mapping + + This comprehensive domain enrichment is particularly useful for: + - Certificate transparency monitoring + - Threat actor infrastructure discovery + - Identifying shared hosting or certificate patterns + - Detecting potential phishing domains using similar certificates + + ### X.509 Certificates + - Enriches certificates by their hash values (MD5, SHA-1, SHA-256) + - Extracts detailed certificate metadata including extensions and key information + + **Note**: Certificate discovery for domains adds an additional API call per domain enrichment. Be mindful of Censys API rate limits. + +The connector queries the Censys Search API and creates related entities based on the data returned. + +### Data Flow + +```mermaid +graph LR + subgraph OpenCTI Input + IP[IPv4-Addr / IPv6-Addr] + Domain[Domain-Name] + Cert[X509-Certificate] + end + + subgraph Censys API + hostEnrichmentAPI[Host Enrichment API] + CertAPI[Certificate API] + end + + subgraph OpenCTI Output + City[City Location] + Country[Country Location] + Region[Region Location] + AdminArea[Administrative Area] + Hostname[Hostname Observable] + Software[Software Entity] + Certificate[X509-Certificate] + AS[Autonomous System] + Org[Organization Identity] + ServiceNote[Service Information Note] + ReputationNote[Host Reputation Note] + IPOut[IPv4/IPv6 Observable] + end + + IP --> hostEnrichmentAPI + Domain --> hostEnrichmentAPI + Cert --> CertAPI + hostEnrichmentAPI --> City + hostEnrichmentAPI --> Country + hostEnrichmentAPI --> Region + hostEnrichmentAPI --> AdminArea + hostEnrichmentAPI --> Hostname + hostEnrichmentAPI --> Software + hostEnrichmentAPI --> Certificate + hostEnrichmentAPI --> AS + hostEnrichmentAPI --> Org + hostEnrichmentAPI --> ServiceNote + hostEnrichmentAPI --> ReputationNote + hostEnrichmentAPI --> IPOut + CertAPI --> Certificate +``` + +### Enrichment Mapping + +| Censys Data | OpenCTI Entity | Description | +|---------------------------|--------------------------|-------------------------------------------------------------| +| location.city | City (Location) | City where the host is located | +| location.country | Country (Location) | Country where the host is located | +| location.continent | Region (Location) | Continent/region of the host | +| location.province | Administrative Area | Province/state with coordinates | +| dns.names | Hostname | DNS hostnames resolving to the IP | +| services.protocol | Note | Service protocol information | +| services.scan_time | Note | Service scan timestamp | +| services.labels | Note | Service labels (e.g., REMOTE_ACCESS, WEB) | +| services.threats | Note | Associated threats and security information | +| services.software | Software | Running software with vendor and CPE | +| services.vulns | Vulnerability | CVEs detected for the service, with CVSS/EPSS/CWE data | +| services.cert | X509-Certificate | TLS certificates from services | +| reputation.score | Note | Host reputation score and risk level | +| reputation.model_version | Note | Reputation model version used | +| autonomous_system.asn | Autonomous System | ASN number | +| autonomous_system.name | Organization (Identity) | Organization operating the AS | +| Certificate fingerprints | X509-Certificate | Certificate with SHA-1, SHA-256, MD5 hashes | +| Certificate parsed data | X509-Certificate | Subject, issuer, validity, key info, extensions | + +### Entity Mapping by Observable Type + +| Input Type | Generated Entities | +|------------------|---------------------------------------------------------------------------------| +| IPv4-Addr | Locations, Hostnames, Software, Certificates, ASN, Organization, Notes | +| IPv6-Addr | Locations, Hostnames, Software, Certificates, ASN, Organization, Notes | +| Domain-Name | Related IPs + all enrichment data for each resolved IP | +| X509-Certificate | Certificate entity with full parsed metadata | + +### Relationships Created + +| Relationship Type | Source | Target | Description | +|--------------------|---------------------|---------------------|---------------------------------------| +| `located-at` | IP Observable | City/Country/Region | Geolocation relationship | +| `located-at` | IP Observable | Administrative Area | Province/state location | +| `resolves-to` | Hostname | IP Observable | DNS resolution | +| `related-to` | IP Observable | Organization | Operating organization | +| `belongs-to` | IP Observable | Autonomous System | ASN membership | +| `related-to` | IP Observable | Software | Running software | +| `has` | Software | Vulnerability | CVE affecting the detected service | +| `related-to` | IP Observable | X509-Certificate | Associated TLS certificates | +| `related-to` | Autonomous System | Organization | AS operator | +| `related-to` | Autonomous System | Country | AS country location | +| `related-to` | Domain-Name | IPv4/IPv6 Address | Resolved IP addresses | + +### Note Types + +The connector generates two types of notes with detailed information: + +#### Service Information Notes +Each detected service generates a comprehensive note containing: +- **Service Protocol**: The network protocol (e.g., SSH, HTTP, HTTPS) +- **Scan Time**: ISO 8601 timestamp of when the service was scanned +- **Service Labels**: Security classification labels (e.g., REMOTE_ACCESS, WEB) +- **Threats**: Any associated security threats or warnings + +*Note Format Example:* +``` +- Scan Time: 2025-11-03T12:35:48Z + +- Labels + - REMOTE_ACCESS + +## Threats +- [Threat details if any] +``` + +#### Reputation Notes +Host reputation information is documented in external notes containing: +- **Score**: Numeric reputation score (0-1 range) +- **Score Level**: Risk classification (LOW, MEDIUM_RISK, HIGH, CRITICAL) +- **Model Version**: Version of the reputation model used + +*Note Format Example:* +``` +- Score: 42 +- Score level: MEDIUM_RISK +- Model version: 2.0.0 +``` + +### Processing Details + +1. **TLP Check**: Validates observable TLP against `max_tlp` setting +2. **API Query**: Queries appropriate Censys endpoint based on observable type +3. **Location Processing**: Creates hierarchical location entities +4. **DNS Processing**: Creates hostname observables with resolution relationships +5. **Service Processing**: Creates comprehensive service notes with protocol, labels, and threats +6. **Reputation Processing**: Generates reputation notes with score and risk level +7. **ASN Processing**: Creates autonomous system with organization relationship +8. **Certificate Processing**: Full certificate parsing with all available metadata + +## Debugging + +Enable verbose logging by setting: + +```env +CONNECTOR_LOG_LEVEL=debug +``` + +Log output includes: +- API request details +- Entity generation progress +- Relationship creation status +- Error handling information + +## Additional information + +- **API Reference**: [Censys Search API Documentation](https://search.censys.io/api) +- **Rate Limits**: API calls are subject to Censys rate limits based on subscription tier +- **Data Freshness**: Censys continuously scans the internet; data freshness depends on scan frequency +- **TLP Handling**: Observables with TLP above `MAX_TLP` will not be sent to Censys +- **Playbook Support**: This connector supports OpenCTI playbook automation +- **Roadmap**: Potential future support for additional observable types (e.g., URL) diff --git a/internal-enrichment/censys-enrichmentapis/__metadata__/CONNECTOR_CONFIG_DOC.md b/internal-enrichment/censys-enrichmentapis/__metadata__/CONNECTOR_CONFIG_DOC.md new file mode 100644 index 00000000000..146acabda3f --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/__metadata__/CONNECTOR_CONFIG_DOC.md @@ -0,0 +1,18 @@ +# 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 base URL of the OpenCTI instance. | +| OPENCTI_TOKEN | `string` | ✅ | Format: [`password`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | | The API token to connect to OpenCTI. | +| CENSYS_ENRICHMENT_ORGANISATION_ID | `string` | ✅ | Format: [`password`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | | Censys organisation ID. | +| CENSYS_ENRICHMENT_TOKEN | `string` | ✅ | Format: [`password`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | | Censys API token. | +| CONNECTOR_NAME | `string` | | string | `"Censys EnrichmentAPIs"` | The name of the connector. | +| CONNECTOR_SCOPE | `array` | | string | `["IPv4-Addr", "IPv6-Addr", "X509-Certificate", "Domain-Name"]` | The scope of the connector. Must be a subset of: ['Domain-Name', 'IPv4-Addr', 'IPv6-Addr', 'X509-Certificate']. | +| CONNECTOR_LOG_LEVEL | `string` | | `debug` `info` `warn` `warning` `error` | `"error"` | The minimum level of logs to display. | +| CONNECTOR_TYPE | `const` | | `INTERNAL_ENRICHMENT` | `"INTERNAL_ENRICHMENT"` | | +| CONNECTOR_AUTO | `boolean` | | boolean | `false` | Whether the connector should run automatically when an entity is created or updated. | +| CENSYS_ENRICHMENT_MAX_TLP | `string` | | `TLP:WHITE` `TLP:CLEAR` `TLP:GREEN` `TLP:AMBER` `TLP:AMBER+STRICT` `TLP:RED` | `"TLP:AMBER"` | The maximum TLP level allowed for enrichment. | diff --git a/internal-enrichment/censys-enrichmentapis/__metadata__/connector_config_schema.json b/internal-enrichment/censys-enrichmentapis/__metadata__/connector_config_schema.json new file mode 100644 index 00000000000..c17362efd62 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/__metadata__/connector_config_schema.json @@ -0,0 +1,92 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://www.filigran.io/connectors/censys-enrichmentapis_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.", + "format": "password", + "type": "string", + "writeOnly": true + }, + "CONNECTOR_NAME": { + "default": "Censys EnrichmentAPIs", + "description": "The name of the connector.", + "type": "string" + }, + "CONNECTOR_SCOPE": { + "default": [ + "IPv4-Addr", + "IPv6-Addr", + "X509-Certificate", + "Domain-Name" + ], + "description": "The scope of the connector. Must be a subset of: ['Domain-Name', 'IPv4-Addr', 'IPv6-Addr', 'X509-Certificate'].", + "items": { + "type": "string" + }, + "type": "array" + }, + "CONNECTOR_LOG_LEVEL": { + "default": "error", + "description": "The minimum level of logs to display.", + "enum": [ + "debug", + "info", + "warn", + "warning", + "error" + ], + "type": "string" + }, + "CONNECTOR_TYPE": { + "const": "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" + }, + "CENSYS_ENRICHMENT_MAX_TLP": { + "default": "TLP:AMBER", + "description": "The maximum TLP level allowed for enrichment.", + "enum": [ + "TLP:WHITE", + "TLP:CLEAR", + "TLP:GREEN", + "TLP:AMBER", + "TLP:AMBER+STRICT", + "TLP:RED" + ], + "type": "string" + }, + "CENSYS_ENRICHMENT_ORGANISATION_ID": { + "description": "Censys organisation ID.", + "format": "password", + "type": "string", + "writeOnly": true + }, + "CENSYS_ENRICHMENT_TOKEN": { + "description": "Censys API token.", + "format": "password", + "type": "string", + "writeOnly": true + } + }, + "required": [ + "OPENCTI_URL", + "OPENCTI_TOKEN", + "CENSYS_ENRICHMENT_ORGANISATION_ID", + "CENSYS_ENRICHMENT_TOKEN" + ], + "additionalProperties": true +} \ No newline at end of file diff --git a/internal-enrichment/censys-enrichmentapis/__metadata__/connector_manifest.json b/internal-enrichment/censys-enrichmentapis/__metadata__/connector_manifest.json new file mode 100644 index 00000000000..a3998467c3e --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/__metadata__/connector_manifest.json @@ -0,0 +1,27 @@ +{ + "title": "Censys EnrichmentAPIs", + "slug": "censys-enrichmentapis", + "description": "The Censys EnrichmentAPIs connector allows OpenCTI to enrich observables (such as domains, IP addresses, and certificates) using data from the Censys search and intelligence platform. It retrieves detailed information on hosts, certificates, services, and organizations to enhance context and visibility within investigations.", + "short_description": "Enriches domains, IP addresses, and certificates in OpenCTI with Censys data, including host services, open ports, and certificate details for infrastructure visibility.", + "logo": "internal-enrichment/censys-enrichmentapis/__metadata__/logo.png", + "use_cases": [ + "Infrastructure & Attack Surface Visibility", + "Detection & Response Enablement" + ], + "solution_categories": [ + "Enrichment & Reputation" + ], + "license_type": "Commercial", + "contact": null, + "verified": false, + "last_verified_date": "2026-09-12", + "playbook_supported": true, + "max_confidence_level": 50, + "support_version": ">=6.8.11", + "subscription_link": "https://platform.censys.io/", + "source_code": "https://github.com/OpenCTI-Platform/connectors/tree/master/internal-enrichment/censys-enrichmentapis", + "manager_supported": true, + "container_version": "rolling", + "container_image": "opencti/connector-censys-enrichmentapis", + "container_type": "INTERNAL_ENRICHMENT" +} \ No newline at end of file diff --git a/internal-enrichment/censys-enrichmentapis/__metadata__/logo.png b/internal-enrichment/censys-enrichmentapis/__metadata__/logo.png new file mode 100644 index 00000000000..d412b4d3aea Binary files /dev/null and b/internal-enrichment/censys-enrichmentapis/__metadata__/logo.png differ diff --git a/internal-enrichment/censys-enrichmentapis/docker-compose.yml b/internal-enrichment/censys-enrichmentapis/docker-compose.yml new file mode 100644 index 00000000000..bf6ae7ab06b --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/docker-compose.yml @@ -0,0 +1,22 @@ +version: '3' +services: + connector-censys-enrichmentapis: + build: + context: . + dockerfile: Dockerfile + image: connector-censys-enrichmentapis:latest + environment: + # OpenCTI configuration + - OPENCTI_URL=${OPENCTI_URL:-http://localhost:8080} + - OPENCTI_TOKEN=${OPENCTI_TOKEN} + # Common connector configuration + - CONNECTOR_ID=${CONNECTOR_ID} + - CONNECTOR_NAME=${CONNECTOR_NAME:-Censys EnrichmentAPIs} + - CONNECTOR_SCOPE=${CONNECTOR_SCOPE:-IPv4-Addr,IPv6-Addr,X509-Certificate,Domain-Name} + - CONNECTOR_LOG_LEVEL=${CONNECTOR_LOG_LEVEL:-error} + - CONNECTOR_AUTO=${CONNECTOR_AUTO:-false} + # Censys configuration + - CENSYS_ENRICHMENT_ORGANISATION_ID=${CENSYS_ENRICHMENT_ORGANISATION_ID} + - CENSYS_ENRICHMENT_TOKEN=${CENSYS_ENRICHMENT_TOKEN} + - CENSYS_ENRICHMENT_MAX_TLP=${CENSYS_ENRICHMENT_MAX_TLP:-TLP:AMBER} + restart: always diff --git a/internal-enrichment/censys-enrichmentapis/src/__init__.py b/internal-enrichment/censys-enrichmentapis/src/__init__.py new file mode 100644 index 00000000000..82109e30bc3 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/__init__.py @@ -0,0 +1,5 @@ +from censys_enrichmentapis.settings import ConfigLoader + +__all__ = [ + "ConfigLoader", +] diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/__init__.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builder.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builder.py new file mode 100644 index 00000000000..9be5936c0f9 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builder.py @@ -0,0 +1,38 @@ +from censys_enrichmentapis.builders import ( + CertificateStixBuilder, + GeographyStixBuilder, + NetworkStixBuilder, + ServiceStixBuilder, +) +from censys_enrichmentapis.builders.base import StixBuildContext +from connectors_sdk.models import BaseObject, OrganizationAuthor, TLPMarking + + +class CensysStixBuilder: + """Coordinate area-specific STIX builders over one shared bundle.""" + + def __init__(self) -> None: + self._context = StixBuildContext() + self.geography = GeographyStixBuilder(self._context) + self.network = NetworkStixBuilder(self._context) + self.certificates = CertificateStixBuilder(self._context) + self.services = ServiceStixBuilder(self._context) + + @property + def author(self) -> OrganizationAuthor: + return self._context.author + + @property + def marking(self) -> TLPMarking: + return self._context.marking + + @property + def bundle(self) -> list[BaseObject]: + return self._context.bundle + + def reset(self) -> None: + self._context.reset() + self.services.reset() + + def add_author_and_marking(self) -> None: + self._context.add_author_and_marking() diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/__init__.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/__init__.py new file mode 100644 index 00000000000..1a7a5c241d3 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/__init__.py @@ -0,0 +1,11 @@ +from censys_enrichmentapis.builders.certificate import CertificateStixBuilder +from censys_enrichmentapis.builders.geography import GeographyStixBuilder +from censys_enrichmentapis.builders.network import NetworkStixBuilder +from censys_enrichmentapis.builders.service import ServiceStixBuilder + +__all__ = [ + "CertificateStixBuilder", + "GeographyStixBuilder", + "NetworkStixBuilder", + "ServiceStixBuilder", +] diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/base.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/base.py new file mode 100644 index 00000000000..d7feb5e9c61 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/base.py @@ -0,0 +1,63 @@ +from connectors_sdk.models import ( + BaseObject, + OrganizationAuthor, + Reference, + Relationship, + TLPMarking, +) +from connectors_sdk.models.enums import RelationshipType, TLPLevel + + +class StixBuildContext: + """Shared bundle state and metadata used by all area builders.""" + + def __init__(self) -> None: + self.author = OrganizationAuthor(name="Censys EnrichmentAPIs Connector") + self.marking = TLPMarking(level=TLPLevel.CLEAR) + self.common_props = {"author": self.author, "markings": [self.marking]} + self.bundle: list[BaseObject] = [] + + def reset(self) -> None: + # Replace rather than clear so bundles already returned to callers remain stable. + self.bundle = [] + + def add_author_and_marking(self) -> None: + self.bundle.extend([self.author, self.marking]) + + def add_relationship( + self, + source: Reference, + target: Reference, + relationship_type: RelationshipType, + ) -> None: + self.bundle.append( + Relationship( + source=source, + target=target, + type=relationship_type, + **self.common_props, + ) + ) + + +class AreaStixBuilder: + """Base class providing area builders access to shared build state.""" + + def __init__(self, context: StixBuildContext) -> None: + self._context = context + + @property + def bundle(self) -> list[BaseObject]: + return self._context.bundle + + @property + def common_props(self) -> dict[str, object]: + return self._context.common_props + + def add_relationship( + self, + source: Reference, + target: Reference, + relationship_type: RelationshipType, + ) -> None: + self._context.add_relationship(source, target, relationship_type) diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/certificate.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/certificate.py new file mode 100644 index 00000000000..57e601779f8 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/certificate.py @@ -0,0 +1,84 @@ +from censys_enrichmentapis.builders.base import AreaStixBuilder +from censys_platform import Certificate +from connectors_sdk.models import Reference, X509Certificate +from connectors_sdk.models.enums import HashAlgorithm, RelationshipType + + +class CertificateStixBuilder(AreaStixBuilder): + def _add_parsed_fields( + self, certificate: X509Certificate, cert: Certificate + ) -> None: + certificate.serial_number = cert.parsed.serial_number + certificate.issuer = cert.parsed.issuer_dn + certificate.subject = cert.parsed.subject_dn + if cert.parsed.signature: + certificate.signature_algorithm = ( + cert.parsed.signature.signature_algorithm.name + ) + if cert.parsed.validity_period: + certificate.validity_not_before = cert.parsed.validity_period.not_before + certificate.validity_not_after = cert.parsed.validity_period.not_after + if cert.parsed.subject_key_info: + certificate.subject_public_key_algorithm = ( + cert.parsed.subject_key_info.key_algorithm.name + ) + if cert.parsed.subject_key_info.rsa: + certificate.subject_public_key_modulus = ( + cert.parsed.subject_key_info.rsa.modulus + ) + certificate.subject_public_key_exponent = ( + cert.parsed.subject_key_info.rsa.exponent + ) + + def _add_extensions( + self, certificate: X509Certificate, cert: Certificate + ) -> None: + if cert.parsed.extensions.key_usage: + certificate.key_usage = cert.parsed.extensions.key_usage.model_dump_json() + if cert.parsed.extensions.basic_constraints: + certificate.basic_constraints = ( + cert.parsed.extensions.basic_constraints.model_dump_json() + ) + certificate.crl_distribution_points = str( + cert.parsed.extensions.crl_distribution_points + ) + certificate.authority_key_identifier = cert.parsed.extensions.authority_key_id + if cert.parsed.extensions.extended_key_usage: + certificate.extended_key_usage = ( + cert.parsed.extensions.extended_key_usage.model_dump_json() + ) + certificate.certificate_policies = str( + cert.parsed.extensions.certificate_policies + ) + + def add_certificate( + self, + cert: Certificate | None, + *, + related_observable: Reference | None = None, + ) -> X509Certificate | None: + if not cert or not ( + cert.fingerprint_sha256 or cert.fingerprint_sha1 or cert.fingerprint_md5 + ): + return None + + hashes = { + algorithm: fingerprint + for algorithm, fingerprint in ( + (HashAlgorithm.SHA1, cert.fingerprint_sha1), + (HashAlgorithm.SHA256, cert.fingerprint_sha256), + (HashAlgorithm.MD5, cert.fingerprint_md5), + ) + if fingerprint + } + certificate = X509Certificate(hashes=hashes or None, **self.common_props) + if cert.parsed: + self._add_parsed_fields(certificate=certificate, cert=cert) + if cert.parsed.extensions: + self._add_extensions(certificate=certificate, cert=cert) + self.bundle.append(certificate) + if related_observable: + self.add_relationship( + certificate, related_observable, RelationshipType.RELATED_TO + ) + return certificate diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/geography.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/geography.py new file mode 100644 index 00000000000..4c3508649da --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/geography.py @@ -0,0 +1,96 @@ +from censys_enrichmentapis.builders.base import AreaStixBuilder +from censys_platform import Coordinates +from connectors_sdk.models import ( + AdministrativeArea, + City, + Country, + Reference, + Region, + Relationship, +) +from connectors_sdk.models.enums import RelationshipType + + +class GeographyStixBuilder(AreaStixBuilder): + def add_city(self, observable: Reference, name: str | None) -> None: + if not name: + return + + city = City(name=name, **self.common_props) + self.bundle.extend( + [ + city, + Relationship( + source=observable, + target=city, + type=RelationshipType.LOCATED_AT, + **self.common_props, + ), + ] + ) + + def add_country(self, observable: Reference, name: str | None) -> Country | None: + if not name: + return None + + country = Country(name=name, **self.common_props) + self.bundle.extend( + [ + country, + Relationship( + source=observable, + target=country, + type=RelationshipType.LOCATED_AT, + **self.common_props, + ), + ] + ) + return country + + def add_region(self, observable: Reference, name: str | None) -> None: + if not name: + return + + region = Region(name=name, **self.common_props) + self.bundle.extend( + [ + region, + Relationship( + source=observable, + target=region, + type=RelationshipType.LOCATED_AT, + **self.common_props, + ), + ] + ) + + def add_administrative_area( + self, + observable: Reference, + name: str | None, + coordinates: Coordinates | None, + ) -> None: + if not name: + return + + administrative_area = ( + AdministrativeArea( + name=name, + latitude=coordinates.latitude, + longitude=coordinates.longitude, + **self.common_props, + ) + if coordinates + else AdministrativeArea(name=name, **self.common_props) + ) + self.bundle.extend( + [ + administrative_area, + Relationship( + source=observable, + target=administrative_area, + type=RelationshipType.LOCATED_AT, + **self.common_props, + ), + ] + ) diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/network.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/network.py new file mode 100644 index 00000000000..e5928491f0b --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/network.py @@ -0,0 +1,116 @@ +import ipaddress + +from censys_enrichmentapis.builders.base import AreaStixBuilder +from censys_platform import HostDNS +from connectors_sdk.models import ( + AutonomousSystem, + Country, + Hostname, + IPV4Address, + IPV6Address, + Organization, + Reference, + Relationship, +) +from connectors_sdk.models.enums import RelationshipType + + +class NetworkStixBuilder(AreaStixBuilder): + def add_hostnames(self, observable: Reference, dns: HostDNS | None) -> None: + if not dns: + return + + for name in dns.names or []: + host_name = Hostname(value=name, **self.common_props) + self.bundle.extend( + [ + host_name, + Relationship( + source=host_name, + target=observable, + type=RelationshipType.RESOLVES_TO, + **self.common_props, + ), + ] + ) + + def add_organization( + self, + observable: Reference, + name: str | None, + ) -> Organization | None: + if not name: + return None + + organization = Organization(name=name, **self.common_props) + self.bundle.extend( + [ + organization, + Relationship( + source=observable, + target=organization, + type=RelationshipType.RELATED_TO, + **self.common_props, + ), + ] + ) + return organization + + def add_autonomous_system( + self, + observable: Reference, + number: int | None, + name: str | None, + description: str | None, + *, + organization: Organization | None = None, + country: Country | None = None, + ) -> AutonomousSystem | None: + if not number: + return None + + autonomous_system = AutonomousSystem( + name=name, + description=description, + number=number, + **self.common_props, + ) + self.bundle.extend( + [ + autonomous_system, + Relationship( + source=observable, + target=autonomous_system, + type=RelationshipType.BELONGS_TO, + **self.common_props, + ), + ] + ) + if organization: + self.add_relationship( + autonomous_system, organization, RelationshipType.RELATED_TO + ) + if country: + self.add_relationship( + autonomous_system, country, RelationshipType.RELATED_TO + ) + return autonomous_system + + def add_ip(self, observable: Reference, ip: str) -> IPV4Address | IPV6Address: + ip_version = ipaddress.ip_network(ip, strict=False).version + if ip_version == 4: + ip_address = IPV4Address(value=ip, **self.common_props) + else: + ip_address = IPV6Address(value=ip, **self.common_props) + self.bundle.extend( + [ + ip_address, + Relationship( + source=observable, + target=ip_address, + type=RelationshipType.RELATED_TO, + **self.common_props, + ), + ] + ) + return ip_address diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/service.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/service.py new file mode 100644 index 00000000000..7e2fa0691fd --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/builders/service.py @@ -0,0 +1,611 @@ +import datetime +from collections.abc import Sequence +from typing import Any +from urllib.parse import quote + +from censys_enrichmentapis.builders.base import AreaStixBuilder, StixBuildContext +from censys_platform import HostEnrichmentService, Reputation, Service +from connectors_sdk.models import ( + AttackPattern, + ExternalReference, + Malware, + Note, + Reference, + Relationship, + Software, + Vulnerability, +) +from connectors_sdk.models.enums import ( + CvssSeverity, + MalwareType, + NoteType, + RelationshipType, +) + + +HostService = HostEnrichmentService | Service + + +class ServiceStixBuilder(AreaStixBuilder): + def __init__(self, context: StixBuildContext) -> None: + super().__init__(context) + self._vulnerabilities_by_identifier: dict[str, Vulnerability] = {} + self._vulnerability_relationships: set[tuple[str, str]] = set() + + def reset(self) -> None: + self._vulnerabilities_by_identifier.clear() + self._vulnerability_relationships.clear() + + def add_software( + self, + observable: Reference, + name: str | None, + vendor: str | None, + cpe: str | None, + version: str | None = None, + ) -> Software | None: + if not name: + return None + + software = Software( + name=name, + vendor=vendor, + cpe=cpe, + version=version, + **self.common_props, + ) + self.bundle.extend( + [ + software, + Relationship( + source=observable, + target=software, + type=RelationshipType.RELATED_TO, + **self.common_props, + ), + ] + ) + return software + + def add_vulnerability( + self, software: Software, vulnerability: object + ) -> Vulnerability | None: + identifier = self._get_value(vulnerability, "id") or self._get_value( + vulnerability, "name" + ) + if not isinstance(identifier, str) or not identifier.startswith("CVE-"): + return None + + vulnerability_entity = self._vulnerabilities_by_identifier.get(identifier) + if vulnerability_entity is None: + vulnerability_entity = self._create_vulnerability( + identifier=identifier, + vulnerability=vulnerability, + ) + self._vulnerabilities_by_identifier[identifier] = vulnerability_entity + self.bundle.append(vulnerability_entity) + + relationship_key = (str(software.id), str(vulnerability_entity.id)) + if relationship_key not in self._vulnerability_relationships: + self.bundle.append( + Relationship( + source=software, + target=vulnerability_entity, + type=RelationshipType.HAS, + **self.common_props, + ) + ) + self._vulnerability_relationships.add(relationship_key) + return vulnerability_entity + + def _create_vulnerability( + self, identifier: str, vulnerability: object + ) -> Vulnerability: + metrics = self._get_value(vulnerability, "metrics") or {} + cvss = self._get_value(metrics, "cvss_v31") or {} + epss = self._get_value(metrics, "epss") or {} + components = self._get_value(cvss, "components") or {} + severity = self._get_value(vulnerability, "severity") + + return Vulnerability( + name=identifier, + cwe_ids=self._string_values(self._get_value(vulnerability, "cwes")), + epss_score=self._get_value(epss, "score"), + epss_percentile=self._get_value(epss, "percentile"), + is_cisa_kev=bool(self._get_value(vulnerability, "kev")), + cvss_v3_vector_string=self._get_value(cvss, "vector"), + cvss_v3_base_score=self._get_value(cvss, "score"), + cvss_v3_base_severity=self._cvss_severity(severity), + cvss_v3_attack_vector=self._get_value(components, "attack_vector"), + cvss_v3_attack_complexity=self._get_value( + components, "attack_complexity" + ), + cvss_v3_privileges_required=self._get_value( + components, "privileges_required" + ), + cvss_v3_user_interaction=self._get_value( + components, "user_interaction" + ), + cvss_v3_scope=self._get_value(components, "scope"), + cvss_v3_confidentiality_impact=self._get_value( + components, "confidentiality" + ), + cvss_v3_integrity_impact=self._get_value(components, "integrity"), + cvss_v3_availability_impact=self._get_value( + components, "availability" + ), + external_references=[ + ExternalReference( + source_name="CVE", + external_id=identifier, + url=f"https://nvd.nist.gov/vuln/detail/{identifier}", + ) + ], + **self.common_props, + ) + + def add_service_vulnerabilities( + self, + observable: Reference, + services: Sequence[HostService] | None, + ) -> None: + """Create the IP -> Software -> Vulnerability path for each service.""" + for service in services or []: + software_by_cpe: dict[str, Software] = {} + for software_data in self._get_value(service, "software") or []: + software = self.add_software( + observable=observable, + name=self._get_value(software_data, "product"), + vendor=self._get_value(software_data, "vendor"), + cpe=self._get_value(software_data, "cpe"), + version=self._get_value(software_data, "version"), + ) + cpe = self._get_value(software_data, "cpe") + if software and isinstance(cpe, str): + software_by_cpe[cpe] = software + + for vulnerability in self._get_value(service, "vulns") or []: + for cpe in self._vulnerability_cpes(vulnerability): + software = software_by_cpe.get(cpe) + if software is None: + software = self._add_software_from_cpe(observable, cpe) + if software is not None: + software_by_cpe[cpe] = software + if software is not None: + self.add_vulnerability(software, vulnerability) + + def _add_software_from_cpe( + self, observable: Reference, cpe: str + ) -> Software | None: + parts = cpe.split(":") + if len(parts) < 6 or parts[0:2] != ["cpe", "2.3"]: + return None + return self.add_software( + observable=observable, + vendor=parts[3], + name=parts[4], + version=parts[5] if parts[5] != "*" else None, + cpe=cpe, + ) + + def _vulnerability_cpes(self, vulnerability: object) -> set[str]: + cpes = set() + for evidence in self._get_value(vulnerability, "evidence") or []: + cpe = self._get_value(evidence, "found_value") + if isinstance(cpe, str) and cpe.startswith("cpe:2.3:"): + cpes.add(cpe) + return cpes + + def add_note( + self, + observable: Reference, + content: str | None, + publication_date: str | None, + port: int | None, + ) -> None: + if not (content and publication_date and port): + return + + self.bundle.append( + Note( + abstract=f"Service banner on port {port}", + content=content, + created=datetime.datetime.fromisoformat(publication_date), + authors=[self._context.author.name], + objects=[observable], + **self.common_props, + ) + ) + + def add_reputation_note( + self, + observable: Reference, + observable_value: str, + reputation: Reputation | None, + ) -> None: + if not reputation: + return + + content_parts = [] + censys_url = ( "https://platform.censys.io/hosts/" f"{quote(observable_value, safe='')}" ) + content_parts.append( + f"\n[View this host {observable_value} on Censys Platform]({censys_url})\n\n" + ) + + + score_label = reputation.label + if reputation.score is not None: + value = int(reputation.score * 100) + content_parts.append(f"- Score: {value}") + if score_label: + content_parts.append(f"- Label: {score_label}") + if reputation.model_version: + content_parts.append(f"- Model version: {reputation.model_version}") + + self._add_reputation_evidence_features(reputation, content_parts) + if not content_parts: + return + + + self.bundle.append( + Note( + abstract="Censys host reputation", + content="\n".join(content_parts), + note_types=[NoteType.EXTERNAL], + labels=[score_label] if score_label else None, + authors=[self._context.author.name], + objects=[observable], + **self.common_props, + ) + ) + + def _add_reputation_evidence_features( + self, reputation: Reputation, content_parts: list[str] + ) -> None: + if not reputation.evidence or not isinstance(reputation.evidence, list): + return + + content_parts.append("\n**Evidence Features:**") + rows = [ + "| Feature | Value | Contribution | Category |", + "|---|---:|---:|---|", + ] + for evidence in reputation.evidence: + if evidence.feature: + feature = evidence.feature + name = str(feature.name) if feature.name else "Unknown" + value = str(feature.value) if feature.value else "Unknown" + category = str(feature.category) if feature.category else "Unknown" + contribution = "Unknown" + if feature.contribution is not None: + if isinstance(feature.contribution, (int, float)): + contribution = f"{feature.contribution:+.2f}%" + else: + contribution = str(feature.contribution) + rows.append( + f"| {name} | {value} | {contribution} | {category} |" + ) + + if len(rows) > 2: + content_parts.append("\n".join(rows)) + + def _markdown_cell(self, value: Any) -> str: + """Make a value safe for use inside a Markdown table cell.""" + if value is None: + return "—" + + if isinstance(value, bool): + value = str(value).lower() + + return ( + str(value) + .replace("|", r"\|") + .replace("\n", "
") + ) + + + def _build_service_content(self, service: HostService) -> str: + content_parts = [] + protocol = self._get_value(service, "protocol") + scan_time = self._get_value(service, "scan_time") + if protocol: + content_parts.append(f"- Protocol: {protocol}") + if scan_time: + content_parts.append(f"- Scan Time: {scan_time}") + + labels = [ + label_value + for label in self._get_value(service, "labels") or [] + if (label_value := self._get_value(label, "value")) + ] + if labels: + if content_parts: + content_parts.append("") + content_parts.append("- Labels") + content_parts.extend(f" - {label}" for label in labels) + + threats_info = [] + for threat in self._get_value(service, "threats") or []: + threat_details = [] + if self._get_value(threat, "name"): + threat_details.append(self._get_value(threat, "name")) + if self._get_value(threat, "severity"): + threat_details.append( + f"Severity: {self._get_value(threat, 'severity')}" + ) + if threat_details: + threats_info.append("- " + " | ".join(threat_details)) + + if threats_info: + if content_parts: + content_parts.append("") + content_parts.append("### Threats") + content_parts.extend(threats_info) + return "\n".join(content_parts) + + def add_service_notes( + self, + observable: Reference, + services: Sequence[HostService] | None, + ) -> None: + for service in services or []: + scan_time = self._get_value(service, "scan_time") + port = self._get_value(service, "port") + protocol = self._get_value(service, "protocol") + if not (scan_time and port): + continue + + content = self._build_service_content(service) + if not content: + continue + self.bundle.append( + Note( + abstract=( + f"Service information on port {port} " + f"({protocol or 'Unknown'})" + ), + content=content, + note_types=[NoteType.EXTERNAL], + created=datetime.datetime.fromisoformat( + scan_time + ), + authors=[self._context.author.name], + objects=[observable], + **self.common_props, + ) + ) + + def add_service_threats( + self, + observable: Reference, + observable_value: str, + services: Sequence[HostService] | None, + ) -> None: + for service in services or []: + self._add_threats( + observable=observable, + observable_value=observable_value, + threats=self._get_value(service, "threats"), + port=self._get_value(service, "port"), + protocol=self._get_value(service, "protocol"), + ) + + def _add_threats( + self, + observable: Reference, + observable_value: str, + threats: object | None, + port: int | None, + protocol: str | None, + ) -> None: + for threat in threats or []: + threat_name = self._get_value(threat, "name") + threat_id = self._get_value(threat, "id") + if not threat_name or not threat_id: + continue + + malware = self._add_threat_malware(threat) + if malware: + self.bundle.extend( + [ + malware, + Relationship( + source=observable, + target=malware, + type=RelationshipType.RELATED_TO, + **self.common_props, + ), + ] + ) + + for tactic in self._get_value(threat, "tactic") or []: + attack_pattern = self._add_threat_attack_pattern(tactic) + if attack_pattern: + self.bundle.extend( + [ + attack_pattern, + Relationship( + source=observable, + target=attack_pattern, + type=RelationshipType.RELATED_TO, + **self.common_props, + ), + ] + ) + + threat_note = self._build_threat_note( + observable=observable, + observable_value=observable_value, + threat=threat, + port=port, + protocol=protocol, + ) + if threat_note: + self.bundle.append(threat_note) + + def _add_threat_malware(self, threat: object) -> Malware | None: + malware_data = self._get_value(threat, "malware") + if not isinstance(malware_data, dict): + return None + primary_name = malware_data.get("primary_name") + if not primary_name: + return None + + malware_type_enums = [] + for threat_type in self._get_value(threat, "type") or []: + if isinstance(threat_type, str): + try: + normalized = threat_type.lower().replace("_", "-") + malware_type_enums.append(MalwareType(normalized)) + except (ValueError, KeyError): + pass + + return Malware( + name=primary_name, + is_family=False, + aliases=malware_data.get("all_names", []), + types=malware_type_enums or None, + description=( + f"{self._get_value(threat, 'id')}: " + f"{self._get_value(threat, 'name')}" + ), + **self.common_props, + ) + + def _add_threat_attack_pattern(self, tactic: str) -> AttackPattern | None: + if not isinstance(tactic, str) or not tactic.strip(): + return None + + tactic_name = tactic.upper().replace("_", " ") + mitre_id = self._get_mitre_tactic_id(tactic) + external_refs = [] + if mitre_id: + external_refs.append( + ExternalReference( + source_name="mitre-attack", + external_id=mitre_id, + url=f"https://attack.mitre.org/tactics/{mitre_id}/", + ) + ) + return AttackPattern( + name=tactic_name, + external_references=external_refs or None, + **self.common_props, + ) + + def _build_threat_note( + self, + observable: Reference, + observable_value: str, + threat: object, + port: int | None, + protocol: str | None, + ) -> Note | None: + threat_name = self._get_value(threat, "name") + threat_id = self._get_value(threat, "id") + if not threat_name: + return None + + censys_url = ( "https://platform.censys.io/hosts/" f"{quote(observable_value, safe='')}" ) + content_parts = [ + f"\n[View this host {observable_value} on Censys Platform]({censys_url})\n\n" + ] + + + rows = [ + "| Key | Value |", + "|---|---|", + f"| Threat ID | {self._markdown_cell(threat_id)} |", + f"| Name | {self._markdown_cell(threat_name)} |", + ] + + threat_types = self._get_value(threat, "type") or [] + types_str = ", ".join( + item.replace("_", " ") + for item in threat_types + if isinstance(item, str) + ) + if types_str: + rows.append(f"| Threat Types | {self._markdown_cell(types_str)} |") + + tactics = self._get_value(threat, "tactic") or [] + tactics_str = ", ".join( + item.replace("_", " ").title() + for item in tactics + if isinstance(item, str) + ) + if tactics_str: + rows.append(f"| Tactics | {self._markdown_cell(tactics_str)} |") + + content_parts.append("\n".join(rows)) + + if evidence := self._get_value(threat, "evidence"): + if isinstance(evidence, list) and evidence: + content_parts.append("\n\n**Evidence:**") + for item in evidence: + if isinstance(item, dict): + data_path = item.get("data_path", "unknown") + found_value = item.get("found_value", "") + content_parts.append(f"- {data_path}: {found_value}") + + if malware_data := self._get_value(threat, "malware"): + if isinstance(malware_data, dict) and malware_data.get("primary_name"): + content_parts.append( + f"\n- **Malware:** {malware_data['primary_name']}" + ) + if aliases := malware_data.get("all_names"): + content_parts.append(f"- **Aliases:** {', '.join(aliases)}") + if updated := malware_data.get("last_updated_at"): + content_parts.append(f"- **Last Updated:** {updated}") + + return Note( + abstract=f"Service Threat: {threat_name}" + + (f" (Port {port}/{protocol})" if port and protocol else ""), + content="\n".join(content_parts), + note_types=[NoteType.EXTERNAL], + labels=[ + item.lower().replace("_", "-") + for item in threat_types + if isinstance(item, str) + ] + or None, + authors=[self._context.author.name], + objects=[observable], + **self.common_props, + ) + + def _get_mitre_tactic_id(self, tactic: str) -> str | None: + mitre_map = { + "persistence": "TA0003", + "execution": "TA0002", + "discovery": "TA0007", + "lateral_movement": "TA0008", + "collection": "TA0009", + "exfiltration": "TA0010", + "command_and_control": "TA0011", + "impact": "TA0040", + "initial_access": "TA0001", + "privilege_escalation": "TA0004", + "defense_evasion": "TA0005", + "credential_access": "TA0006", + } + return mitre_map.get(tactic.lower()) + + def _get_value(self, value: object, field: str) -> object | None: + if isinstance(value, dict): + return value.get(field) + return getattr(value, field, None) + + def _string_values(self, value: object | None) -> list[str] | None: + if not isinstance(value, list): + return None + values = [item for item in value if isinstance(item, str)] + return values or None + + def _cvss_severity(self, value: object | None) -> CvssSeverity | None: + if not isinstance(value, str): + return None + try: + return CvssSeverity(value.upper()) + except ValueError: + return None diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/client.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/client.py new file mode 100644 index 00000000000..4ec7f22d134 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/client.py @@ -0,0 +1,153 @@ +from typing import Any, Dict, Generator + +import httpx + +from censys_enrichmentapis.errors import EntityHasNoUsableHashError +from censys_platform import ( + SDK, + Certificate, + Host, + HostEnrichment, + SearchQueryInputBody, + V3GlobaldataSearchQueryResponse, +) + + +class Client: + def __init__(self, organisation_id: str, token: str): + self.organisation_id = organisation_id + self.token = token + + def fetch_ip(self, ip: str) -> HostEnrichment: + """Fetch host enrichment data for a given IP address from Censys. + Args: + ip (str): The IP address to fetch data for. + Returns: + HostEnrichment: The host enrichment data retrieved from Censys. + Raises: + ValueError: If no data is found for the given IP address. + """ + raw_response: dict[str, Any] = {} + + def preserve_response(response: httpx.Response) -> None: + # The generated Censys SDK currently omits service ``software`` + # and ``vulns`` from HostEnrichmentService. Read the response once + # in a hook so the fields can be restored after SDK deserialization. + try: + response.read() + raw_response.update(response.json()) + except (ValueError, TypeError): + pass + + with httpx.Client( + follow_redirects=True, event_hooks={"response": [preserve_response]} + ) as http_client, SDK( + organization_id=self.organisation_id, + personal_access_token=self.token, + client=http_client, + ) as sdk: + res = sdk.global_data.get_host_enrichment(host_ip=ip) + if host_asset := res.result.result: + self._restore_service_fields(host_asset.resource, raw_response) + return host_asset.resource + raise ValueError(f"No data found for IP {ip}") + + @staticmethod + def _restore_service_fields( + host: HostEnrichment, response: dict[str, Any] + ) -> None: + """Restore service fields not yet represented by censys-platform 0.16.""" + resource = response.get("result", {}).get("result", {}).get("resource", {}) + raw_services = resource.get("services", []) if isinstance(resource, dict) else [] + for service, raw_service in zip(host.services or [], raw_services, strict=False): + if not isinstance(raw_service, dict): + continue + for field in ("software", "vulns"): + if field in raw_service: + # Generated Pydantic models ignore unknown API fields, but + # their instances remain safely extensible for conversion. + service.__dict__[field] = raw_service[field] + + def fetch_certs(self, hashes: Dict[str, str]) -> Generator[Certificate, None, None]: + """Fetch certificates by their hashes + + Args: + hashes (Dict[str, str]): A dictionary containing one or more of the following keys + with their corresponding hash values: + - "MD5" + - "SHA-1" + - "SHA-256" + Yields: + Certificate: Censys Certificate objects matching the provided hashes. + Raises: + EntityHasNoUsableHashError: If none of the required hashes are provided. + """ + if not any(h in hashes for h in ("MD5", "SHA-1", "SHA-256")): + raise EntityHasNoUsableHashError( + "At least one hash (MD5, SHA1, SHA256) must be provided." + ) + parts = [] + if "MD5" in hashes: + parts.append(f'cert.fingerprint_md5 = "{hashes["MD5"]}"') + if "SHA-1" in hashes: + parts.append(f'cert.fingerprint_sha1 = "{hashes["SHA-1"]}"') + if "SHA-256" in hashes: + parts.append(f'cert.fingerprint_sha256 = "{hashes["SHA-256"]}"') + query = " or ".join(parts) + search_query = SearchQueryInputBody(query=query) + with SDK( + organization_id=self.organisation_id, + personal_access_token=self.token, + ) as sdk: + ## TODO: change to use get_property instead of search on port 443 + res: V3GlobaldataSearchQueryResponse = sdk.global_data.search( + search_query_input_body=search_query + ) + if res.result.result: + for hit in res.result.result.hits: + if hit.certificate_v1: + yield hit.certificate_v1.resource + + def fetch_hosts(self, hostname: str) -> Generator[Host, None, None]: + """Fetch hosts by hostname + Args: + hostname (str): The hostname to search for. + Yields: + Generator[Host, None, None]: Yields Host objects matching the hostname. + """ + with SDK( + organization_id=self.organisation_id, + personal_access_token=self.token, + ) as sdk: + query = f"host.dns.names = '{hostname}'" + search_query = SearchQueryInputBody(query=query) + res: V3GlobaldataSearchQueryResponse = sdk.global_data.search( + search_query_input_body=search_query + ) + if res.result.result: + for hit in res.result.result.hits: + if hit.host_v1: + yield hit.host_v1.resource + + def fetch_certs_by_domain(self, domain: str) -> Generator[Certificate, None, None]: + """Fetch certificates that reference a domain in their names + + Args: + domain (str): The domain name to search for. + + Yields: + Generator[Certificate, None, None]: Yields Certificate objects matching the domain. + """ + with SDK( + organization_id=self.organisation_id, + personal_access_token=self.token, + ) as sdk: + query = f"cert.names = '{domain}'" + search_query = SearchQueryInputBody(query=query) + res: V3GlobaldataSearchQueryResponse = sdk.global_data.search( + search_query_input_body=search_query + ) + if res.result.result: + for hit in res.result.result.hits: + if hit.certificate_v1: + yield hit.certificate_v1.resource diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/connector.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/connector.py new file mode 100644 index 00000000000..e36dff96e66 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/connector.py @@ -0,0 +1,147 @@ +from typing import Any, Iterator + +from censys_enrichmentapis.client import Client +from censys_enrichmentapis.converters import get_converter +from censys_enrichmentapis.converters.base import CensysConverter +from censys_enrichmentapis.errors import ( + EntityNotInScopeError, + MaxTlpError, +) +from censys_enrichmentapis.settings import ConfigLoader +from connectors_sdk.models import BaseObject +from pycti import OpenCTIConnectorHelper + + +class Connector: + """Censys connector""" + + def __init__( + self, + config: ConfigLoader, + helper: OpenCTIConnectorHelper, + client: Client, + ) -> None: + self.config = config + self.helper = helper + self.client = client + + def _send_bundle(self, stix_objects: list[dict[str, Any]]) -> str: + bundle = self.helper.stix2_create_bundle(items=stix_objects) + bundles_sent = self.helper.send_stix2_bundle(bundle=bundle) + return f"Sending {len(bundles_sent)} stix bundle(s) for worker import" + + def _is_entity_in_scope(self, entity_type: str) -> bool: + """Return True if the entity type is supported by the connector scope.""" + return entity_type in self.config.connector.scope + + def _extract_tlp(self, markings: list[dict[str, Any]]) -> str | None: + """Return the first TLP string (e.g., 'TLP:AMBER'), or None if not present.""" + return next( + ( + marking["definition"] + for marking in markings + if marking["definition_type"] == "TLP" + ), + None, + ) + + def _is_entity_tlp_allowed(self, markings: list[dict[str, Any]]) -> bool: + """Return True if the entity's TLP is <= configured max TLP.""" + return self.helper.check_max_tlp( + tlp=self._extract_tlp(markings=markings), + max_tlp=self.config.censys_enrichment.max_tlp, + ) + + def _generate_octi_objects( + self, + stix_entity: dict[str, Any], + primary_observable_labels: list[str] | None = None, + ) -> Iterator[BaseObject]: + # Annotate ``Iterator`` (not ``Generator``) so the type + # matches the ``list_iterator`` returned by + # ``iter(converter.to_stix(...))``. Keeping ``return + # iter(...)`` instead of rewriting as a real ``yield from`` + # generator is deliberate: the converter dispatch + # (``_get_converter`` → ``get_converter`` → + # ``EntityTypeNotSupportedError``) must run eagerly so + # misconfigured entity types surface at call time rather + # than only when something starts iterating the returned + # object — the test suite (and the ``_message_callback`` + # error path that wraps this) both rely on the eager + # behaviour. + converter = self._get_converter(entity_type=stix_entity["type"]) + stix_objects = converter.to_stix(observable=stix_entity) + if primary_observable_labels is not None: + primary_observable_labels.extend(converter.primary_observable_labels) + return iter(stix_objects) + + def _get_converter(self, entity_type: str) -> CensysConverter: + converter = get_converter(entity_type=entity_type) + converter.client = self.client + return converter + + @staticmethod + def _merge_primary_observable_labels( + stix_objects: list[dict[str, Any]], + stix_id: str, + labels: list[str], + ) -> None: + if not labels: + return + + for stix_object in stix_objects: + if stix_object.get("id") != stix_id: + continue + existing_labels = stix_object.get("x_opencti_labels", []) + stix_object["x_opencti_labels"] = list( + dict.fromkeys([*existing_labels, *labels]) + ) + return + + def _process( + self, + observable: dict[str, Any], + stix_entity: dict[str, Any], + original_stix_objects: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + if not self._is_entity_in_scope(entity_type=observable["entity_type"]): + raise EntityNotInScopeError( + f"Unsupported entity type: {observable['entity_type']}" + ) + if not self._is_entity_tlp_allowed(markings=observable["objectMarking"]): + raise MaxTlpError( + f"TLP {observable['objectMarking']} of observable exceeds MAX TLP" + ) + primary_observable_labels: list[str] = [] + generated_stix_objects = [ + octi_object.to_stix2_object() + for octi_object in self._generate_octi_objects( + stix_entity=stix_entity, + primary_observable_labels=primary_observable_labels, + ) + ] + self._merge_primary_observable_labels( + stix_objects=original_stix_objects, + stix_id=stix_entity["id"], + labels=primary_observable_labels, + ) + return original_stix_objects + generated_stix_objects + + def _message_callback(self, data: dict[str, Any]) -> str: + try: + stix_objects = self._process( + observable=data["enrichment_entity"], + stix_entity=data["stix_entity"], + original_stix_objects=data["stix_objects"], + ) + return self._send_bundle(stix_objects=stix_objects) + except Exception as e: + self.helper.connector_logger.error(e) + is_in_playbook_context = not bool(data.get("event_type")) + if is_in_playbook_context: + # If it's in a playbook context, we send the original bundle unchanged + return self._send_bundle(stix_objects=data["stix_objects"]) + raise e + + def run(self) -> None: + self.helper.listen(message_callback=self._message_callback) diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/__init__.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/__init__.py new file mode 100644 index 00000000000..88c7c327dae --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/__init__.py @@ -0,0 +1,34 @@ +from censys_enrichmentapis.converters.base import CensysConverter +from censys_enrichmentapis.converters.certificate import CertificateConverter +from censys_enrichmentapis.converters.domain import DomainConverter +from censys_enrichmentapis.converters.host import HostConverter +from censys_enrichmentapis.errors import EntityTypeNotSupportedError + +_CONVERTER_MAP: dict[str, type[CensysConverter]] = { + "IPv4-Addr": HostConverter, + "IPv6-Addr": HostConverter, + "X509-Certificate": CertificateConverter, + "Domain-Name": DomainConverter, + "ipv4-addr": HostConverter, + "ipv6-addr": HostConverter, + "x509-certificate": CertificateConverter, + "domain-name": DomainConverter, +} + + +def get_converter(entity_type: str) -> CensysConverter: + cls = _CONVERTER_MAP.get(entity_type) + if cls is None: + raise EntityTypeNotSupportedError( + f"Observable type {entity_type} not supported" + ) + return cls() + + +__all__ = [ + "CensysConverter", + "CertificateConverter", + "DomainConverter", + "HostConverter", + "get_converter", +] diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/base.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/base.py new file mode 100644 index 00000000000..806c56362ab --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/base.py @@ -0,0 +1,55 @@ +from abc import ABC, abstractmethod +from typing import Any, Mapping + +from censys_enrichmentapis.builder import CensysStixBuilder +from censys_enrichmentapis.client import Client +from connectors_sdk.models import BaseObject + +# ``observable`` arrives at the converters in two shapes: a plain +# ``dict`` from the OpenCTI enrichment payload (see +# ``Connector._process``) AND a ``stix2`` object (e.g. +# ``stix2.IPv4Address``) when ``DomainConverter._append_hosts`` +# composes ``HostConverter._convert(observable=ip_stix.to_stix2_object(), ...)`` +# or when a test passes a ``stix2`` instance directly. Both shapes +# expose the read-only ``observable["..."]`` / ``.get(...)`` access +# pattern the converters rely on, so the contract is "any +# string-keyed mapping" rather than ``dict`` specifically — using +# ``Mapping[str, Any]`` lets static type checkers (mypy, pyright) +# accept the ``stix2`` callers without unsafe casts and also makes +# the read-only intent explicit at the API surface. +ObservableLike = Mapping[str, Any] + + +class CensysConverter(ABC): + def __init__(self) -> None: + self.builder = CensysStixBuilder() + self.client: Client | None = None + self.primary_observable_labels: list[str] = [] + + def to_stix( + self, observable: ObservableLike, data: Any | None = None + ) -> list[BaseObject]: + """Return the STIX bundle for *observable*. + + If *data* is provided, skip the API fetch and convert it directly — + useful for tests and for callers that already have the payload. + """ + self.builder.reset() + self.primary_observable_labels = [] + if data is None: + data = self._fetch_data(observable=observable) + self._convert(observable=observable, data=data) + return self.builder.bundle + + def _require_client(self) -> Client: + if self.client is None: + raise ValueError("Client is required") + return self.client + + @abstractmethod + def _fetch_data(self, observable: ObservableLike) -> Any: + """Fetch data required for STIX conversion.""" + + @abstractmethod + def _convert(self, observable: ObservableLike, data: Any) -> None: + """Convert fetched data to STIX objects.""" diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/certificate.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/certificate.py new file mode 100644 index 00000000000..3a10c0b5532 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/certificate.py @@ -0,0 +1,12 @@ +from censys_enrichmentapis.converters.base import CensysConverter, ObservableLike +from censys_platform import Certificate + + +class CertificateConverter(CensysConverter): + def _fetch_data(self, observable: ObservableLike) -> list[Certificate]: + return list(self._require_client().fetch_certs(hashes=observable["hashes"])) + + def _convert(self, observable: ObservableLike, data: list[Certificate]) -> None: + self.builder.add_author_and_marking() + for cert in data: + self.builder.certificates.add_certificate(cert=cert) diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/domain.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/domain.py new file mode 100644 index 00000000000..debfc65665d --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/domain.py @@ -0,0 +1,53 @@ +from typing import Any + +from censys_enrichmentapis.converters.base import CensysConverter, ObservableLike +from censys_enrichmentapis.converters.host import HostConverter +from censys_platform import Certificate, Host +from connectors_sdk.models import Reference + + +class DomainConverter(CensysConverter): + def _fetch_data(self, observable: ObservableLike) -> dict[str, list[Any]]: + client = self._require_client() + return { + "hosts": list(client.fetch_hosts(observable["value"])), + "certs": list(client.fetch_certs_by_domain(observable["value"])), + } + + def _convert(self, observable: ObservableLike, data: dict[str, list[Any]]) -> None: + self._append_hosts(stix_entity=observable, hosts=data["hosts"]) + self._append_domain_certs(stix_entity=observable, certs=data["certs"]) + + def _append_hosts(self, stix_entity: ObservableLike, hosts: list[Host]) -> None: + host_converter = HostConverter() + host_converter.builder = self.builder + + for host in hosts: + ip_stix = self.builder.network.add_ip( + observable=Reference(id=stix_entity.get("id")), + ip=host.ip, + ) + host_converter._convert(observable=ip_stix.to_stix2_object(), data=host) + + def _append_domain_certs( + self, stix_entity: ObservableLike, certs: list[Certificate] + ) -> None: + """Append certificate STIX objects and domain relationships to the bundle. + + Args: + stix_entity: The domain STIX entity + certs: List of Certificate objects from Censys + + Side effects: + Appends STIX objects (and a related-to relationship per certificate) to + self.builder.bundle. + """ + observable = Reference(id=stix_entity.get("id")) + + self.builder.add_author_and_marking() + + for cert in certs: + self.builder.certificates.add_certificate( + cert=cert, + related_observable=observable, + ) diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/host.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/host.py new file mode 100644 index 00000000000..e2a75d74f3f --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/converters/host.py @@ -0,0 +1,120 @@ +import re + +from censys_enrichmentapis.converters.base import CensysConverter, ObservableLike +from censys_platform import Host, HostEnrichment, Reputation +from connectors_sdk.models import Reference + + +class HostConverter(CensysConverter): + def _fetch_data(self, observable: ObservableLike) -> HostEnrichment: + return self._require_client().fetch_ip(observable["value"]) + + def _convert_labels(self, data: Host | HostEnrichment) -> list[str]: + label_values = [ + label_value + for label in self._value(data, "labels") or [] + if isinstance((label_value := self._value(label, "value")), str) + ] + label_values.extend( + label_value + for service in self._value(data, "services") or [] + for label in self._value(service, "labels") or [] + if isinstance((label_value := self._value(label, "value")), str) + ) + threat_names = [ + self._value(threat, "name") + for service in self._value(data, "services") or [] + for threat in self._value(service, "threats") or [] + ] + label_values.extend(name for name in threat_names if isinstance(name, str)) + reputation = self._value(data, "reputation") + reputation_label = self._value(reputation, "label") + if isinstance(reputation_label, str): + label_values.append(reputation_label) + return list( + dict.fromkeys( + f"Censys_{self._to_snake_case(label_value.strip())}" + for label_value in label_values + if label_value and label_value.strip() + ) + ) + + def _convert( + self, observable: ObservableLike, data: Host | HostEnrichment + ) -> None: + stix_entity = observable + observable = Reference(id=stix_entity.get("id")) + self.primary_observable_labels = self._convert_labels(data) + + self.builder.add_author_and_marking() + self.builder.geography.add_city( + observable=observable, + name=data.location.city if data.location else None, + ) + self.builder.geography.add_region( + observable=observable, + name=data.location.continent if data.location else None, + ) + self.builder.geography.add_administrative_area( + observable=observable, + name=data.location.province if data.location else None, + coordinates=data.location.coordinates if data.location else None, + ) + + country = self.builder.geography.add_country( + observable=observable, + name=data.location.country if data.location else None, + ) + organization = self.builder.network.add_organization( + observable=observable, + name=data.autonomous_system.name if data.autonomous_system else None, + ) + self.builder.network.add_autonomous_system( + observable=observable, + name=data.autonomous_system.name if data.autonomous_system else None, + description=( + data.autonomous_system.description if data.autonomous_system else None + ), + number=data.autonomous_system.asn if data.autonomous_system else None, + organization=organization, + country=country, + ) + + self.builder.network.add_hostnames( + observable=observable, + dns=data.dns, + ) + + self.builder.services.add_service_notes( + observable=observable, + services=data.services, + ) + + reputation = self._value(data, "reputation") + self.builder.services.add_reputation_note( + observable=observable, + observable_value=stix_entity.get("value"), + reputation=reputation if isinstance(reputation, Reputation) else None, + ) + self.builder.services.add_service_vulnerabilities( + observable=observable, + services=data.services, + ) + self.builder.services.add_service_threats( + observable=observable, + observable_value=stix_entity.get("value"), + services=data.services, + ) + + @staticmethod + def _to_snake_case(text: str) -> str: + """Convert text to snake_case format.""" + # Replace spaces and hyphens with underscores + text = re.sub(r'[\s\-]+', '_', text) + return text + + @staticmethod + def _value(value: object, field: str) -> object | None: + if isinstance(value, dict): + return value.get(field) + return getattr(value, field, None) diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/errors.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/errors.py new file mode 100644 index 00000000000..6dc1e1c709d --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/errors.py @@ -0,0 +1,14 @@ +class EntityNotInScopeError(Exception): + """Custom exception for entity not in scope""" + + +class MaxTlpError(Exception): + """Custom exception for exceeding maximum TLP level""" + + +class EntityTypeNotSupportedError(Exception): + """Custom exception for unsupported entity type""" + + +class EntityHasNoUsableHashError(Exception): + """Custom exception for entity having no usable hash""" diff --git a/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/settings.py b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/settings.py new file mode 100644 index 00000000000..b8a3b6b1173 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/censys_enrichmentapis/settings.py @@ -0,0 +1,103 @@ +from typing import Literal + +from connectors_sdk import ( + BaseConfigModel, + BaseConnectorSettings, + BaseInternalEnrichmentConnectorConfig, + ListFromString, +) +from pydantic import Field, SecretStr, field_validator + +# Entity types this connector can actually enrich. These are the +# capitalised OpenCTI ``entity_type`` values, i.e. the capitalised +# subset of the keys of ``censys_enrichmentapis.converters._CONVERTER_MAP`` +# (that map also holds the STIX-lowercase aliases — e.g. ``ipv4-addr`` — +# used for converter dispatch by ``stix_entity["type"]``; this constant +# tracks only the capitalised forms on purpose, because +# ``_is_entity_in_scope`` in ``connector.py`` matches +# ``observable["entity_type"]`` against the capitalised form). Defined +# here as a module-level constant rather than imported from +# ``converters/`` to avoid an import cycle when the SDK loads the +# settings before the connector module is wired up. +SUPPORTED_SCOPE_ENTITY_TYPES: frozenset[str] = frozenset( + {"IPv4-Addr", "IPv6-Addr", "X509-Certificate", "Domain-Name"} +) + + +class _ConnectorConfig(BaseInternalEnrichmentConnectorConfig): + id: str = Field( + default="censys-enrichmentapis--674403d0-4723-40cd-b03c-42fb959d5469", + description="A UUID v4 to identify the connector in OpenCTI.", + ) + name: str = Field( + default="Censys EnrichmentAPIs", + description="The name of the connector.", + ) + scope: ListFromString = Field( + default=["IPv4-Addr", "IPv6-Addr", "X509-Certificate", "Domain-Name"], + description=( + "The scope of the connector. Must be a subset of: " + f"{sorted(SUPPORTED_SCOPE_ENTITY_TYPES)}." + ), + ) + log_level: Literal["debug", "info", "warn", "warning", "error"] = Field( + default="error", + description="The minimum level of logs to display.", + ) + + @field_validator("scope") + @classmethod + def _scope_must_be_supported(cls, value: list[str]) -> list[str]: + """Reject ``CONNECTOR_SCOPE`` entries the connector cannot handle. + + Without this check, a misconfigured ``CONNECTOR_SCOPE`` (e.g. + a typo like ``Domain-name`` or an entity type the connector + does not implement, like ``Url``) would silently fall through + the scope gate in ``Connector._is_entity_in_scope`` AND then + explode much later at dispatch time with + ``EntityTypeNotSupportedError``, after a work has already been + accepted off the queue. Validating the scope at startup turns + the silent dispatch-time failure into a clear configuration + error the operator sees before the connector ever registers + with OpenCTI. + """ + unsupported = [v for v in value if v not in SUPPORTED_SCOPE_ENTITY_TYPES] + if unsupported: + raise ValueError( + f"Unsupported scope entries: {unsupported}. " + f"CONNECTOR_SCOPE must be a subset of " + f"{sorted(SUPPORTED_SCOPE_ENTITY_TYPES)}." + ) + return value + + +class _CensysEnrichmentConfig(BaseConfigModel): + max_tlp: Literal[ + "TLP:WHITE", + "TLP:CLEAR", + "TLP:GREEN", + "TLP:AMBER", + "TLP:AMBER+STRICT", + "TLP:RED", + ] = Field( + default="TLP:AMBER", + description="The maximum TLP level allowed for enrichment.", + ) + + organisation_id: SecretStr = Field( + description="Censys organisation ID.", + ) + token: SecretStr = Field( + description="Censys API token.", + ) + + +class ConfigLoader(BaseConnectorSettings): + connector: _ConnectorConfig = Field( + default_factory=_ConnectorConfig, + description="Internal Enrichment Connector configurations.", + ) + censys_enrichment: _CensysEnrichmentConfig = Field( + default_factory=_CensysEnrichmentConfig, + description="Censys EnrichmentAPIs configurations.", + ) diff --git a/internal-enrichment/censys-enrichmentapis/src/config.yml.sample b/internal-enrichment/censys-enrichmentapis/src/config.yml.sample new file mode 100644 index 00000000000..c896317bfd2 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/config.yml.sample @@ -0,0 +1,15 @@ +opencti: + url: 'http://localhost' + token: 'ChangeMe' + +#connector: +# type: 'INTERNAL_ENRICHMENT' +# id: "ChangeMe" +# name: "Censys EnrichmentAPIs" +# scope: "IPv4-Addr,IPv6-Addr,X509-Certificate,Domain-Name" +# log_level: "error" + +censys_enrichment: + organisation_id: "ChangeMe" + token: "ChangeMe" +# max_tlp: "TLP:AMBER" diff --git a/internal-enrichment/censys-enrichmentapis/src/main.py b/internal-enrichment/censys-enrichmentapis/src/main.py new file mode 100644 index 00000000000..8ffec42847c --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/main.py @@ -0,0 +1,38 @@ +""" +Entry point of the script + +- traceback.print_exc(): This function prints the traceback of the exception to the standard error (stderr). +The traceback includes information about the point in the program where the exception occurred, +which is very useful for debugging purposes. +- exit(1): effective way to terminate a Python program when an error is encountered. +It signals to the operating system and any calling processes that the program did not complete successfully. +""" + +import sys +import traceback + +from censys_enrichmentapis.client import Client +from censys_enrichmentapis.connector import Connector +from censys_enrichmentapis.settings import ConfigLoader +from pycti import OpenCTIConnectorHelper + +if __name__ == "__main__": + try: + config = ConfigLoader() + helper = OpenCTIConnectorHelper( + config=config.to_helper_config(), + playbook_compatible=True, + ) + client = Client( + organisation_id=config.censys_enrichment.organisation_id.get_secret_value(), + token=config.censys_enrichment.token.get_secret_value(), + ) + connector = Connector( + config=config, + helper=helper, + client=client, + ) + connector.run() + except Exception: + traceback.print_exc() + sys.exit(1) diff --git a/internal-enrichment/censys-enrichmentapis/src/requirements.txt b/internal-enrichment/censys-enrichmentapis/src/requirements.txt new file mode 100644 index 00000000000..c8f96932105 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/src/requirements.txt @@ -0,0 +1,6 @@ +pycti==7.260910.0 +pydantic~=2.13.5 +requests~=2.33.0 +validators==0.35.0 +connectors-sdk @ git+https://github.com/OpenCTI-Platform/connectors.git@master#subdirectory=connectors-sdk +censys-platform~=0.16.1 diff --git a/internal-enrichment/censys-enrichmentapis/tests/__init__.py b/internal-enrichment/censys-enrichmentapis/tests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/__init__.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/conftest.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/conftest.py new file mode 100644 index 00000000000..f96f7e0f1d8 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/conftest.py @@ -0,0 +1,175 @@ +import os +import sys +from dataclasses import asdict +from unittest.mock import MagicMock, Mock, patch + +import pytest +from censys_platform import ( + BasicConstraints, + Coordinates, + ExtendedKeyUsage, + Host, + HostAsset, + HostAssetWithMatchedServices, + HostDNS, + HostEnrichment, + HostEnrichmentService, + KeyAlgorithm, + KeyUsage, + Label, + Location, + ResponseEnvelopeHostAsset, + ResponseEnvelopeSearchQueryResponse, + Routing, + SearchQueryHit, + SearchQueryResponse, + Signature, + SubjectKeyInfo, + V3GlobaldataAssetHostResponse, + V3GlobaldataSearchQueryResponse, + ValidityPeriod, +) +from pycti import OpenCTIConnectorHelper +from pytest_mock import MockerFixture + +from .factories import DomainNameEnrichmentFactory, HostFactory, Ipv4EnrichmentFactory + +sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src")) + + +@pytest.fixture(name="mock_config") +def fixture_mock_config(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OPENCTI_URL", "http://test") + monkeypatch.setenv("OPENCTI_TOKEN", "opencti-token") + monkeypatch.setenv("CENSYS_ENRICHMENT_ORGANISATION_ID", "censys-organisation_id") + monkeypatch.setenv("CENSYS_ENRICHMENT_TOKEN", "censys-token") + + +@pytest.fixture(name="mocked_helper") +def fixture_mocked_helper(mocker: MockerFixture) -> Mock: + mocked_helper = mocker.patch("pycti.OpenCTIConnectorHelper") + mocked_helper.stix2_create_bundle = MagicMock( + side_effect=OpenCTIConnectorHelper.stix2_create_bundle + ) + mocked_helper.check_max_tlp = OpenCTIConnectorHelper.check_max_tlp + return mocked_helper + + +@pytest.fixture(name="host_ipv4") +def fixture_host_ipv4() -> HostEnrichment: + return HostEnrichment( + ip="1.1.1.1", + location=Location( + city="Brisbane", + continent="Oceania", + coordinates=Coordinates(latitude=-27.47, longitude=153.02), + country="Australia", + province="Queensland", + ), + dns=HostDNS( + names=["guestcontroller.sa.gov.au", "matrix.cyops.cloud"], + ), + autonomous_system=Routing( + asn=13335, + bgp_prefix="1.1.1.0/24", + country_code="US", + description="CLOUDFLARENET", + name="CLOUDFLARENET", + ), + labels=[ + Label(value="BULLETPROOF"), + Label(value="BULLETPROOF"), + Label(value=""), + ], + services=[ + HostEnrichmentService( + port=443, + scan_time="2025-11-03T12:35:48Z", + labels=[Label(value="REMOTE_ACCESS")], + ) + ], + ) + + +@pytest.fixture +def get_host(): + with patch("censys_platform.global_data.GlobalData.get_host_enrichment") as mock_get_host_enrichment: + host = HostEnrichment( + ip="1.1.1.1", + location=Location( + city="Brisbane", + continent="Oceania", + coordinates=Coordinates(latitude=-27.47, longitude=153.02), + country="Australia", + province="Queensland", + ), + dns=HostDNS( + names=["guestcontroller.sa.gov.au", "matrix.cyops.cloud"], + ), + autonomous_system=Routing( + asn=13335, + bgp_prefix="1.1.1.0/24", + country_code="US", + description="CLOUDFLARENET", + name="CLOUDFLARENET", + ), + labels=[ + Label(value="BULLETPROOF"), + Label(value="BULLETPROOF"), + Label(value=""), + ], + services=[ + HostEnrichmentService( + port=443, + scan_time="2025-11-03T12:35:48Z", + labels=[Label(value="REMOTE_ACCESS")], + ) + ], + ) + mock_result = MagicMock() + mock_result.result.result.resource = host + mock_get_host_enrichment.return_value = mock_result + yield host + + +@pytest.fixture +def ipv4_enrichment_message(): + yield asdict(Ipv4EnrichmentFactory()) + + +@pytest.fixture +def fetch_hosts(): + with patch("censys_platform.global_data.GlobalData.search") as mock_fetch_hosts: + hosts = HostFactory.create_batch(2) + result = V3GlobaldataSearchQueryResponse( + headers={}, + result=ResponseEnvelopeSearchQueryResponse( + result=SearchQueryResponse( + hits=[ + SearchQueryHit( + host_v1=HostAssetWithMatchedServices( + extensions={}, + resource=hosts[0], + ) + ), + SearchQueryHit( + host_v1=HostAssetWithMatchedServices( + extensions={}, + resource=hosts[1], + ) + ), + ], + total_hits=2, + next_page_token="", + query_duration_millis=123, + previous_page_token="", + ), + ), + ) + mock_fetch_hosts.return_value = result + yield hosts + + +@pytest.fixture +def domain_name_enrichment_message(): + yield asdict(DomainNameEnrichmentFactory()) diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/factories.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/factories.py new file mode 100644 index 00000000000..540077d2f0f --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/factories.py @@ -0,0 +1,536 @@ +from dataclasses import dataclass +from datetime import timezone + +from censys_platform import ( + Attribute, + BasicConstraints, + Certificate, + CertificateExtensions, + CertificateParsed, + CertificatePolicy, + Coordinates, + ExtendedKeyUsage, + Host, + HostDNS, + KeyAlgorithm, + KeyUsage, + Location, + Routing, + Service, + Signature, + SubjectKeyInfo, + ValidityPeriod, +) +from factory import ( + Factory, + Faker, + LazyAttribute, + List, + SelfAttribute, + Sequence, + SubFactory, + fuzzy, +) + + +class CoordinatesFactory(Factory): + class Meta: + model = Coordinates + + latitude = Faker("latitude") + longitude = Faker("longitude") + + +class LocationFactory(Factory): + class Meta: + model = Location + + city = Faker("city") + continent = fuzzy.FuzzyChoice( + [ + "Africa", + "Antarctica", + "Asia", + "Europe", + "North America", + "Oceania", + "South America", + ] + ) + coordinates = SubFactory(CoordinatesFactory) + country = Faker("country") + province = Faker("state") + + +class HostDNSFactory(Factory): + class Meta: + model = HostDNS + + names = List([Faker("domain_name"), Faker("domain_name")]) + + +class RoutingFactory(Factory): + class Meta: + model = Routing + + asn = Faker("random_int", min=1, max=65535) + bgp_prefix = Faker("ipv4") + country_code = Faker("country_code") + description = Faker("company") + name = Faker("company") + + +class KeyAlgorithmFactory(Factory): + class Meta: + model = KeyAlgorithm + + name = Faker("word") + + +class SignatureFactory(Factory): + class Meta: + model = Signature + + signature_algorithm = SubFactory(KeyAlgorithmFactory) + + +class ValidityPeriodFactory(Factory): + class Meta: + model = ValidityPeriod + + not_before = Faker("iso8601", tzinfo=timezone.utc) + not_after = Faker("iso8601", tzinfo=timezone.utc) + + +class SubjectKeyInfoFactory(Factory): + class Meta: + model = SubjectKeyInfo + + key_algorithm = SubFactory(KeyAlgorithmFactory) + + +class CertificatePolicyFactory(Factory): + class Meta: + model = CertificatePolicy + + cps = List([Faker("uri"), Faker("uri")]) + id = Faker("bothify", text="2.23.140.1.2.?") + + +class KeyUsageFactory(Factory): + class Meta: + model = KeyUsage + + +class BasicConstraintsFactory(Factory): + class Meta: + model = BasicConstraints + + +class ExtendedKeyUsageFactory(Factory): + class Meta: + model = ExtendedKeyUsage + + +class CertificateExtensionsFactory(Factory): + class Meta: + model = CertificateExtensions + + key_usage = SubFactory(KeyUsageFactory) + basic_constraints = SubFactory(BasicConstraintsFactory) + crl_distribution_points = List([Faker("uri"), Faker("uri")]) + authority_key_id = Faker("sha1") + extended_key_usage = SubFactory(ExtendedKeyUsageFactory) + certificate_policies = List([SubFactory(CertificatePolicyFactory)]) + + +class CertificateParsedFactory(Factory): + class Meta: + model = CertificateParsed + + serial_number = Sequence(lambda n: str(100000000 + n)) + issuer_dn = Faker("sentence", nb_words=6) + subject_dn = Faker("sentence", nb_words=3) + signature = SubFactory(SignatureFactory) + validity_period = SubFactory(ValidityPeriodFactory) + subject_key_info = SubFactory(SubjectKeyInfoFactory) + extensions = SubFactory(CertificateExtensionsFactory) + + +class CertificateFactory(Factory): + class Meta: + model = Certificate + + fingerprint_md5 = Faker("md5") + fingerprint_sha1 = Faker("sha1") + fingerprint_sha256 = Faker("sha256") + parsed = SubFactory(CertificateParsedFactory) + + +class AttributeFactory(Factory): + class Meta: + model = Attribute + + product = Faker("word") + vendor = Faker("company") + cpe = Faker("bothify", text="cpe:2.3:a:?????:*:*:*:*:*:*:*:*") + + +class ServiceFactory(Factory): + class Meta: + model = Service + + banner = Faker("sentence", nb_words=4) + cert = SubFactory(CertificateFactory) + port = Faker("random_int", min=1, max=65535) + scan_time = Faker("iso8601", tzinfo=timezone.utc) + software = List([SubFactory(AttributeFactory)]) + + +class HostFactory(Factory): + def __new__(cls, *args, **kwargs) -> Host: + return super().__new__(*args, **kwargs) + + class Meta: + model = Host + + ip = Faker("ipv4") + location = SubFactory(LocationFactory) + dns = SubFactory(HostDNSFactory) + autonomous_system = SubFactory(RoutingFactory) + services = List([ServiceFactory(), ServiceFactory()]) + + +@dataclass +class StixExternalReference: + description: str + external_id: str + source_name: str + url: str + + +@dataclass +class StixIpv4Entity: + id: str + x_opencti_score: int + x_opencti_description: str + value: str + x_opencti_id: str + x_opencti_type: str + type: str + external_references: list[StixExternalReference] + x_opencti_labels: list[str] + spec_version: str = "2.1" + + +class StixExternalReferenceFactory(Factory): + class Meta: + model = StixExternalReference + + description = Faker("sentence") + external_id = Faker("uuid4") + source_name = "MISP" + url = Faker("url") + + +class StixIpv4EntityFactory(Factory): + class Meta: + model = StixIpv4Entity + + id = Faker("uuid4") + x_opencti_score = Faker("random_int", min=0, max=100) + x_opencti_description = Faker("sentence") + value = Faker("ipv4") + x_opencti_id = SelfAttribute("id") + x_opencti_type = "IPv4-Addr" + type = "ipv4-addr" + external_references = List([SubFactory(StixExternalReferenceFactory)]) + x_opencti_labels = List([Faker("word")]) + + +@dataclass +class Creator: + id: str + name: str + + +class CreatorFactory(Factory): + class Meta: + model = Creator + + id = Faker("uuid4") + name = Faker("name") + + +@dataclass +class Ipv4EnrichmentEntity: + created_at: str + creators: list[Creator] + entity_type: str + id: str + indicators: list[dict] + indicatorsIds: list[str] + objectLabel: list + objectLabelIds: list[str] + objectMarking: list + objectMarkingIds: list[str] + objectOrganization: list[str] + observable_value: str + parent_types: list[str] + spec_version: str + standard_id: str + updated_at: str + value: str + x_opencti_score: int + createdBy: dict = None + createdById: str | None = None + x_opencti_description: str | None = None + + +class Ipv4EnrichmentEntityFactory(Factory): + class Meta: + model = Ipv4EnrichmentEntity + + created_at = Faker("iso8601", tzinfo=timezone.utc) + creators = List([SubFactory(CreatorFactory)]) + entity_type = "IPv4-Addr" + id = Faker("uuid4") + indicators = [] + indicatorsIds = [] + objectLabel = [] + objectLabelIds = [] + objectMarking = [] + objectMarkingIds = LazyAttribute(lambda o: [m.id for m in o.objectMarking]) + objectOrganization = [] + observable_value = Faker("ipv4") + parent_types = [ + "Basic-Object", + "Stix-Object", + "Stix-Core-Object", + "Stix-Cyber-Observable", + ] + spec_version = "2.1" + standard_id = LazyAttribute(lambda o: f"ipv4-addr--{o.id}") + updated_at = Faker("iso8601", tzinfo=timezone.utc) + value = SelfAttribute("observable_value") + x_opencti_score = Faker("random_int", min=0, max=100) + createdBy = None + createdById = LazyAttribute( + lambda o: o.createdBy.x_opencti_id if o.createdBy else None + ) + x_opencti_description = Faker("sentence") + + +@dataclass +class EnrichmentMessage: + id: str + entity_id: str + event_type: str + entity_type: str + enrichment_entity: Ipv4EnrichmentEntity + stix_entity: StixIpv4Entity + stix_objects: list[StixIpv4Entity] + + +class Ipv4EnrichmentFactory(Factory): + def __new__(cls, *args, **kwargs) -> EnrichmentMessage: + return super().__new__(*args, **kwargs) + + class Meta: + model = EnrichmentMessage + + id = Faker("uuid4") + entity_id = LazyAttribute(lambda o: f"ipv4-addr--{o.id}") + entity_type = "IPv4-Addr" + event_type = "INTERNAL_ENRICHMENT" + enrichment_entity = SubFactory(Ipv4EnrichmentEntityFactory) + stix_entity = SubFactory(StixIpv4EntityFactory, id=SelfAttribute("..entity_id")) + stix_objects = LazyAttribute(lambda o: [o.stix_entity]) + + +@dataclass +class MetaData: + mimetype: str + version: str + + +class MetaDataFactory(Factory): + class Meta: + model = MetaData + + mimetype = "application/json" + version = Faker("iso8601") + + +@dataclass +class ImportFile: + id: str + name: str + size: int + metaData: MetaData + createdById: str | None = None + + +@dataclass +class ExternalReference: + id: str + source_name: str + url: str + entity_type: str + external_id: str | None = None + description: str | None = None + created: str | None = None + modified: str | None = None + createdById: str | None = None + hash: str | None = None + importFiles: list[ImportFile] | None = None + importFilesIds: list[str] | None = None + standard_id: str | None = None + + +@dataclass +class DomainNameEnrichmentEntity: + created_at: str + creators: list[Creator] + entity_type: str + externalReferences: list[ExternalReference] + externalReferencesIds: list[str] + id: str + importFiles: list[ImportFile] + importFilesIds: list[str] + indicators: list[dict] + indicatorsIds: list[str] + objectLabel: list + objectLabelIds: list[str] + objectMarking: list + objectMarkingIds: list[str] + objectOrganization: list[str] + observable_value: str + parent_types: list[str] + spec_version: str + standard_id: str + updated_at: str + value: str + x_opencti_score: int + createdBy: dict = None + createdById: str | None = None + x_opencti_description: str | None = None + + +class ImportFileFactory(Factory): + class Meta: + model = ImportFile + + id = Faker("file_path") + name = Faker("file_name") + size = Faker("random_int", min=100, max=10000) + metaData = SubFactory(MetaDataFactory) + createdById = None + + +class ExternalReferenceFactory(Factory): + class Meta: + model = ExternalReference + + id = Faker("uuid4") + source_name = "MISP" + url = Faker("url") + entity_type = "External-Reference" + external_id = Faker("uuid4") + description = Faker("sentence") + created = Faker("iso8601") + modified = Faker("iso8601") + createdById = None + hash = None + importFiles = List([SubFactory(ImportFileFactory)]) + importFilesIds = LazyAttribute( + lambda o: [f.id for f in o.importFiles] if o.importFiles else [] + ) + standard_id = LazyAttribute(lambda o: f"external-reference--{o.id}") + + +class DomainNameEnrichmentEntityFactory(Factory): + class Meta: + model = DomainNameEnrichmentEntity + + created_at = Faker("iso8601") + creators = List([SubFactory(CreatorFactory)]) + entity_type = "Domain-Name" + externalReferences = List([SubFactory(ExternalReferenceFactory)]) + externalReferencesIds = LazyAttribute( + lambda o: [r.id for r in o.externalReferences] + ) + id = Faker("uuid4") + importFiles = [] + importFilesIds = [] + indicators = [] + indicatorsIds = [] + objectLabel = [] + objectLabelIds = [] + objectMarking = [] + objectMarkingIds = LazyAttribute(lambda o: [m.id for m in o.objectMarking]) + objectOrganization = [] + observable_value = Faker("domain_name") + parent_types = [ + "Basic-Object", + "Stix-Object", + "Stix-Core-Object", + "Stix-Cyber-Observable", + ] + spec_version = "2.1" + standard_id = LazyAttribute(lambda o: f"domain-name--{o.id}") + updated_at = Faker("iso8601") + value = SelfAttribute("observable_value") + x_opencti_score = Faker("random_int", min=0, max=100) + createdBy = None + createdById = LazyAttribute( + lambda o: o.createdBy.x_opencti_id if o.createdBy else None + ) + x_opencti_description = Faker("sentence") + + +@dataclass +class StixDomainNameEntity: + id: str + x_opencti_score: int + x_opencti_description: str + value: str + x_opencti_id: str + x_opencti_type: str + type: str + external_references: list[StixExternalReference] + x_opencti_labels: list[str] + spec_version: str = "2.1" + + +class StixDomainNameEntityFactory(Factory): + class Meta: + model = StixDomainNameEntity + + id = Faker("uuid4") + x_opencti_score = Faker("random_int", min=0, max=100) + x_opencti_description = Faker("sentence") + value = Faker("domain_name") + x_opencti_id = SelfAttribute("id") + x_opencti_type = "Domain-Name" + type = "domain-name" + external_references = List([SubFactory(StixExternalReferenceFactory)]) + x_opencti_labels = List([Faker("word")]) + + +class DomainNameEnrichmentFactory(Factory): + def __new__(cls, *args, **kwargs) -> EnrichmentMessage: + return super().__new__(*args, **kwargs) + + class Meta: + model = EnrichmentMessage + + id = Faker("uuid4") + entity_id = LazyAttribute(lambda o: f"domain-name--{o.id}") + entity_type = "Domain-Name" + event_type = "INTERNAL_ENRICHMENT" + enrichment_entity = SubFactory(DomainNameEnrichmentEntityFactory) + stix_entity = SubFactory( + StixDomainNameEntityFactory, id=SelfAttribute("..entity_id") + ) + stix_objects = LazyAttribute(lambda o: [o.stix_entity]) diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_builder.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_builder.py new file mode 100644 index 00000000000..fe952f0144b --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_builder.py @@ -0,0 +1,131 @@ +from censys_enrichmentapis.builder import CensysStixBuilder +from censys_platform import Certificate, CertificateParsed +from connectors_sdk.models import City, IPV4Address, IPV6Address, Reference, Vulnerability +from connectors_sdk.models.enums import HashAlgorithm + +SHA256 = "73b8ed5becf1ba6493d2e2215a42dfdc7877e91e311ff5e59fb43d094871e699" +OBSERVABLE = Reference(id="ipv4-addr--cbd67181-b9f8-595b-8bc3-3971e34fa1cc") + + +def test_area_builders_share_context_and_reset_replaces_bundle() -> None: + builder = CensysStixBuilder() + builder.add_author_and_marking() + original_bundle = builder.bundle + + builder.reset() + + assert len(original_bundle) == 2 + assert builder.bundle == [] + assert builder.bundle is not original_bundle + + +def test_geography_builder_adds_to_shared_bundle() -> None: + builder = CensysStixBuilder() + + builder.geography.add_city(observable=OBSERVABLE, name="Paris") + + assert isinstance(builder.bundle[0], City) + assert builder.bundle[0].name == "Paris" + assert len(builder.bundle) == 2 + + +def test_network_builder_selects_ip_version() -> None: + builder = CensysStixBuilder() + + ipv4 = builder.network.add_ip(OBSERVABLE, "192.0.2.1") + ipv6 = builder.network.add_ip(OBSERVABLE, "2001:db8::1") + + assert isinstance(ipv4, IPV4Address) + assert isinstance(ipv6, IPV6Address) + + +def test_service_builder_skips_invalid_vulnerability() -> None: + builder = CensysStixBuilder() + software = builder.services.add_software( + observable=OBSERVABLE, + name="nginx", + vendor="nginx", + cpe="cpe:2.3:a:nginx:nginx:1.0:*:*:*:*:*:*:*", + ) + assert software is not None + bundle_size = len(builder.bundle) + + vulnerability = builder.services.add_vulnerability( + software, {"id": "not-a-cve"} + ) + + assert vulnerability is None + assert len(builder.bundle) == bundle_size + + +def test_builder_reset_clears_vulnerability_cache() -> None: + builder = CensysStixBuilder() + software = builder.services.add_software( + observable=OBSERVABLE, + name="nginx", + vendor="nginx", + cpe="cpe:2.3:a:nginx:nginx:1.0:*:*:*:*:*:*:*", + ) + assert software is not None + first_vulnerability = builder.services.add_vulnerability( + software, {"id": "CVE-2026-12345"} + ) + + builder.reset() + software = builder.services.add_software( + observable=OBSERVABLE, + name="nginx", + vendor="nginx", + cpe="cpe:2.3:a:nginx:nginx:1.0:*:*:*:*:*:*:*", + ) + assert software is not None + second_vulnerability = builder.services.add_vulnerability( + software, {"id": "CVE-2026-12345"} + ) + + assert second_vulnerability is not first_vulnerability + assert len( + [obj for obj in builder.bundle if isinstance(obj, Vulnerability)] + ) == 1 + + +def test_add_certificate_filters_missing_fingerprints() -> None: + # A certificate that only carries a SHA-256 fingerprint must not leak + # ``None`` values into ``hashes`` (the SDK model rejects them); only the + # present fingerprint is kept and the object must serialize cleanly. + builder = CensysStixBuilder() + + certificate = builder.certificates.add_certificate( + cert=Certificate(fingerprint_sha256=SHA256) + ) + + assert certificate is not None + assert certificate.hashes == {HashAlgorithm.SHA256: SHA256} + # Would raise before the fix (None hash values fail validation). + certificate.to_stix2_object() + + +def test_add_certificate_without_fingerprints_is_skipped() -> None: + # A certificate with parsed metadata but no fingerprint cannot be + # serialized (empty hashes are rejected by stix2), so it is skipped. + builder = CensysStixBuilder() + + certificate = builder.certificates.add_certificate( + cert=Certificate( + parsed=CertificateParsed( + serial_number="123456789", + issuer_dn="C=US, O=Example", + subject_dn="CN=example.com", + ), + ) + ) + + assert certificate is None + assert builder.bundle == [] + + +def test_add_certificate_returns_none_for_empty_certificate() -> None: + builder = CensysStixBuilder() + + assert builder.certificates.add_certificate(cert=Certificate()) is None + assert builder.certificates.add_certificate(cert=None) is None diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_config.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_config.py new file mode 100644 index 00000000000..d5e28d747ee --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_config.py @@ -0,0 +1,93 @@ +import pytest +from censys_enrichmentapis.settings import ConfigLoader +from connectors_sdk import ConfigValidationError +from pydantic import HttpUrl + + +@pytest.mark.usefixtures("mock_config") +def test_config() -> None: + config = ConfigLoader() + + # Test config from env + assert config.opencti.url == HttpUrl("http://test") + assert config.opencti.token.get_secret_value() == "opencti-token" + + assert ( + config.censys_enrichment.organisation_id.get_secret_value() + == "censys-organisation_id" + ) + assert config.censys_enrichment.token.get_secret_value() == "censys-token" + + # Test defaults + assert ( + config.connector.id == "censys-enrichmentapis--674403d0-4723-40cd-b03c-42fb959d5469" + ) + assert config.connector.type == "INTERNAL_ENRICHMENT" + assert config.connector.name == "Censys EnrichmentAPIs" + assert config.connector.scope == [ + "IPv4-Addr", + "IPv6-Addr", + "X509-Certificate", + "Domain-Name", + ] + assert config.connector.log_level == "error" + assert config.connector.auto is False + + assert config.censys_enrichment.max_tlp == "TLP:AMBER" + + +@pytest.mark.usefixtures("mock_config") +def test_missing_values(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("OPENCTI_URL") + with pytest.raises(ConfigValidationError) as exc_info: + ConfigLoader() + assert exc_info.value.args == ("Error validating configuration.",) + + +# --------------------------------------------------------------------------- +# Scope field validator +# --------------------------------------------------------------------------- +# +# Without ``@field_validator("scope")``, a misconfigured +# ``CONNECTOR_SCOPE`` (e.g. a typo like ``Domain-name`` or an entity +# type the connector does not implement, like ``Url``) silently +# falls through ``Connector._is_entity_in_scope`` and only blows up +# at dispatch time inside ``_generate_octi_objects`` with +# ``EntityTypeNotSupportedError`` — after a work has already been +# accepted off the queue. These tests pin the new fail-at-startup +# contract so a future contributor cannot regress the validator +# back to a no-op. + + +@pytest.mark.usefixtures("mock_config") +def test_scope_accepts_subset_of_supported_types( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("CONNECTOR_SCOPE", "IPv4-Addr,Domain-Name") + config = ConfigLoader() + assert config.connector.scope == ["IPv4-Addr", "Domain-Name"] + + +@pytest.mark.usefixtures("mock_config") +def test_scope_rejects_unsupported_entity_type( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # ``Url`` is a legal OpenCTI entity_type but the Censys + # connector does not implement a converter for it. The + # validator must surface this at startup rather than letting it + # slip through to dispatch time. + monkeypatch.setenv("CONNECTOR_SCOPE", "IPv4-Addr,Url") + with pytest.raises(ConfigValidationError): + ConfigLoader() + + +@pytest.mark.usefixtures("mock_config") +def test_scope_rejects_case_typo(monkeypatch: pytest.MonkeyPatch) -> None: + # OpenCTI uses ``Domain-Name`` (capitalised D, capitalised N). + # A typo like ``Domain-name`` would match no converter at + # dispatch time and produce a confusing + # ``EntityTypeNotSupportedError`` after the fact — the + # validator must reject it now instead. + monkeypatch.setenv("CONNECTOR_SCOPE", "Domain-name") + with pytest.raises(ConfigValidationError): + ConfigLoader() diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_connector.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_connector.py new file mode 100644 index 00000000000..33def993b72 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_connector.py @@ -0,0 +1,451 @@ +import json +from typing import Any +from unittest.mock import Mock + +import pytest +from pytest_mock import MockerFixture +from censys_enrichmentapis.client import Client +from censys_enrichmentapis.connector import Connector +from censys_enrichmentapis.errors import ( + EntityNotInScopeError, + EntityTypeNotSupportedError, + MaxTlpError, +) +from censys_enrichmentapis.settings import ConfigLoader + + +def filter_by_key_value(items: list[dict], key: str, value: Any) -> list[dict]: + return [item for item in items if item.get(key) == value] + + +@pytest.mark.usefixtures("mock_config") +def test__send_bundle(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + res = connector._send_bundle([]) + mocked_helper.stix2_create_bundle.assert_called_once_with(items=[]) + mocked_helper.send_stix2_bundle.assert_called_once() + assert res == "Sending 0 stix bundle(s) for worker import" + + +@pytest.mark.usefixtures("mock_config") +def test__is_entity_in_scope(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + assert connector._is_entity_in_scope("IPv4-Addr") + assert not connector._is_entity_in_scope("NotInScope") + + +@pytest.mark.usefixtures("mock_config") +@pytest.mark.parametrize( + "markings, expected_tlp", + [ + ([], None), + ([{"definition_type": "TLP", "definition": "TLP:AMBER"}], "TLP:AMBER"), + ([{"definition_type": "PAP", "definition": "PAP:AMBER"}], None), + ( + [ + {"definition_type": "TLP", "definition": "TLP:AMBER"}, + {"definition_type": "PAP", "definition": "PAP:AMBER"}, + ], + "TLP:AMBER", + ), + ], +) +def test__extract_tlp( + mocked_helper: Mock, markings: list[dict[str, str]], expected_tlp: str | None +) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + assert connector._extract_tlp(markings) == expected_tlp + + +@pytest.mark.usefixtures("mock_config") +@pytest.mark.parametrize( + "markings, expected", + [ + ([], True), + ([{"definition_type": "TLP", "definition": "TLP:AMBER"}], True), + ([{"definition_type": "TLP", "definition": "TLP:RED"}], False), + ([{"definition_type": "PAP", "definition": "PAP:AMBER"}], True), + ], +) +def test__is_entity_tlp_allowed( + mocked_helper: Mock, markings: list[dict[str, str]], expected: bool +) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + + assert connector._is_entity_tlp_allowed(markings) == expected + + +@pytest.mark.usefixtures("mock_config") +def test__generate_octi_objects_wrong_entity_type(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + with pytest.raises(EntityTypeNotSupportedError) as exc_info: + connector._generate_octi_objects({"type": "wrong-type"}) + + assert exc_info.typename == "EntityTypeNotSupportedError" + assert exc_info.value.args == ("Observable type wrong-type not supported",) + + +@pytest.mark.usefixtures("mock_config") +def test__process_entity_not_in_scope_error(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + + with pytest.raises(EntityNotInScopeError) as exc_info: + connector._process( + observable={"entity_type": "wrong-type"}, + stix_entity={}, + original_stix_objects=[], + ) + assert exc_info.typename == "EntityNotInScopeError" + assert exc_info.value.args == ("Unsupported entity type: wrong-type",) + + with pytest.raises(MaxTlpError) as exc_info: + connector._process( + observable={ + "entity_type": "IPv4-Addr", + "objectMarking": [{"definition_type": "TLP", "definition": "TLP:RED"}], + }, + stix_entity={}, + original_stix_objects=[], + ) + assert exc_info.typename == "MaxTlpError" + assert exc_info.value.args == ( + "TLP [{'definition_type': 'TLP', 'definition': 'TLP:RED'}] of observable exceeds MAX TLP", + ) + + with pytest.raises(EntityTypeNotSupportedError) as exc_info: + connector._process( + observable={ + "entity_type": "IPv4-Addr", + "objectMarking": [ + {"definition_type": "TLP", "definition": "TLP:AMBER"} + ], + }, + stix_entity={"type": "wrong-type"}, + original_stix_objects=[], + ) + + assert exc_info.typename == "EntityTypeNotSupportedError" + assert exc_info.value.args == ("Observable type wrong-type not supported",) + + +@pytest.mark.usefixtures("mock_config") +def test__process_max_tlp_error(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + + with pytest.raises(MaxTlpError) as exc_info: + connector._process( + observable={ + "entity_type": "IPv4-Addr", + "objectMarking": [{"definition_type": "TLP", "definition": "TLP:RED"}], + }, + stix_entity={}, + original_stix_objects=[], + ) + assert exc_info.typename == "MaxTlpError" + assert exc_info.value.args == ( + "TLP [{'definition_type': 'TLP', 'definition': 'TLP:RED'}] of observable exceeds MAX TLP", + ) + + +@pytest.mark.usefixtures("mock_config") +def test__process_entity_type_not_supported_error(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + + with pytest.raises(EntityTypeNotSupportedError) as exc_info: + connector._process( + observable={ + "entity_type": "IPv4-Addr", + "objectMarking": [ + {"definition_type": "TLP", "definition": "TLP:AMBER"} + ], + }, + stix_entity={"type": "wrong-type"}, + original_stix_objects=[], + ) + + assert exc_info.typename == "EntityTypeNotSupportedError" + assert exc_info.value.args == ("Observable type wrong-type not supported",) + + +@pytest.mark.usefixtures("mock_config") +def test__message_callback_entity_type_not_supported_error(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + + with pytest.raises(EntityTypeNotSupportedError) as exc_info: + connector._message_callback( + { + "event_type": "INTERNAL_ENRICHMENT", + "stix_entity": {"type": "wrong-type"}, + "stix_objects": [], + "enrichment_entity": { + "entity_type": "IPv4-Addr", + "objectMarking": [ + {"definition_type": "TLP", "definition": "TLP:AMBER"} + ], + }, + } + ) + assert exc_info.typename == "EntityTypeNotSupportedError" + assert exc_info.value.args == ("Observable type wrong-type not supported",) + + +@pytest.mark.usefixtures("mock_config") +def test__message_callback_in_playbook(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + + res = connector._message_callback( + { + "stix_objects": [], + "enrichment_entity": { + "entity_type": "wrong-type", + "objectMarking": [ + {"definition_type": "TLP", "definition": "TLP:AMBER"} + ], + }, + } + ) + assert res == "Sending 0 stix bundle(s) for worker import" + + +@pytest.mark.usefixtures("mock_config") +def test__message_callback_not_in_playbook(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + with pytest.raises(KeyError) as exc_info: + connector._message_callback( + { + "event_type": "INTERNAL_ENRICHMENT", # Mean not in playbook + "stix_objects": [], + "enrichment_entity": { + "entity_type": "wrong-type", + "objectMarking": [ + {"definition_type": "TLP", "definition": "TLP:AMBER"} + ], + }, + } + ) + assert exc_info.typename == "KeyError" + assert exc_info.value.args == ("stix_entity",) + + +@pytest.mark.usefixtures("mock_config") +def test_run(mocked_helper: Mock) -> None: + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=Mock(), + ) + connector.run() + + mocked_helper.listen.assert_called_once_with( + message_callback=connector._message_callback + ) + + +@pytest.mark.usefixtures("mock_config") +def test_enrichment(mocked_helper: Mock, get_host, ipv4_enrichment_message): + client = Client( + organisation_id="test-org-id", + token="test-token", + ) + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=client, + ) + sent_bundle = {} + + def capture_sent_bundle(bundle: str, **_): + nonlocal sent_bundle + sent_bundle = json.loads(bundle) + return sent_bundle["objects"] + + connector.helper.send_stix2_bundle = capture_sent_bundle + existing_labels = ipv4_enrichment_message["stix_objects"][0][ + "x_opencti_labels" + ].copy() + connector._message_callback(ipv4_enrichment_message) + + primary_observable = next( + stix_object + for stix_object in sent_bundle["objects"] + if stix_object["id"] == ipv4_enrichment_message["stix_entity"]["id"] + ) + assert primary_observable["x_opencti_labels"] == [ + *existing_labels, + "Censys_BULLETPROOF", + "Censys_REMOTE_ACCESS", + ] + + city_name = filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "City" + )[0]["name"] + assert city_name == get_host.location.city + region_name = filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "Region" + )[0]["name"] + assert region_name == get_host.location.continent + administrative_area_name = filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "Administrative-Area" + )[0]["name"] + assert administrative_area_name == get_host.location.province + country_name = filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "Country" + ) + assert country_name[0]["name"] == get_host.location.country + + hostnames = filter_by_key_value(sent_bundle["objects"], "type", "hostname") + for url in get_host.dns.names: + assert any(hostname_obj["value"] == url for hostname_obj in hostnames) + + autonomous_system = filter_by_key_value( + sent_bundle["objects"], "type", "autonomous-system" + )[0] + assert autonomous_system["number"] == get_host.autonomous_system.asn + assert autonomous_system["name"] == get_host.autonomous_system.name + assert ( + autonomous_system["x_opencti_description"] + == get_host.autonomous_system.description + ) + + +@pytest.mark.usefixtures("mock_config") +def test_domain_name_enrichment( + mocker: MockerFixture, mocked_helper: Mock, fetch_hosts, domain_name_enrichment_message +): + mocker.patch("censys_enrichmentapis.client.Client.fetch_certs_by_domain", return_value=[]) + mock_get_host_enrichment = mocker.patch( + "censys_platform.global_data.GlobalData.get_host_enrichment" + ) + client = Client( + organisation_id="test-org-id", + token="test-token", + ) + connector = Connector( + config=ConfigLoader(), + helper=mocked_helper, + client=client, + ) + sent_bundle = {} + + def capture_sent_bundle(bundle: str, **_): + nonlocal sent_bundle + sent_bundle = json.loads(bundle) + return sent_bundle["objects"] + + connector.helper.send_stix2_bundle = capture_sent_bundle + connector._message_callback(domain_name_enrichment_message) + mock_get_host_enrichment.assert_not_called() + + for host in fetch_hosts: + ipv4_addresses = [ + addr["value"] + for addr in filter_by_key_value(sent_bundle["objects"], "type", "ipv4-addr") + ] + assert host.ip in ipv4_addresses + city_names = [ + city["name"] + for city in filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "City" + ) + ] + assert host.location.city in city_names + region_names = [ + region["name"] + for region in filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "Region" + ) + ] + assert host.location.continent in region_names + administrative_area_names = [ + area["name"] + for area in filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "Administrative-Area" + ) + ] + assert host.location.province in administrative_area_names + country_names = [ + country["name"] + for country in filter_by_key_value( + sent_bundle["objects"], "x_opencti_location_type", "Country" + ) + ] + assert host.location.country in country_names + + hostnames = filter_by_key_value(sent_bundle["objects"], "type", "hostname") + for url in host.dns.names: + assert any(hostname_obj["value"] == url for hostname_obj in hostnames) + + autonomous_system_numbers = [ + asys["number"] + for asys in filter_by_key_value( + sent_bundle["objects"], "type", "autonomous-system" + ) + ] + assert host.autonomous_system.asn in autonomous_system_numbers + autonomous_system_names = [ + asys["name"] + for asys in filter_by_key_value( + sent_bundle["objects"], "type", "autonomous-system" + ) + ] + assert host.autonomous_system.name in autonomous_system_names + autonomous_system_descriptions = [ + asys["x_opencti_description"] + for asys in filter_by_key_value( + sent_bundle["objects"], "type", "autonomous-system" + ) + ] + assert host.autonomous_system.description in autonomous_system_descriptions + + service_notes = filter_by_key_value(sent_bundle["objects"], "type", "note") + for service in host.services: + assert any( + note["abstract"].startswith( + f"Service information on port {service.port} " + ) + for note in service_notes + ) diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_converter.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_converter.py new file mode 100644 index 00000000000..5ced0ff6224 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_converter.py @@ -0,0 +1,113 @@ +import stix2 +from censys_enrichmentapis.converters.host import HostConverter +from censys_platform import HostEnrichment + + +def test_converter_ipv4(host_ipv4: HostEnrichment) -> None: + stix_objects = [ + object_.to_stix2_object() + for object_ in HostConverter().to_stix( + observable=stix2.IPv4Address(value="1.1.1.1"), + data=host_ipv4, + ) + ] + + author_id = "identity--169b39b7-ea64-5a16-bb05-ed1045005079" + marking_id = "marking-definition--613f2e26-407d-48c7-9eca-b8e91df99dc9" + ip_id = "ipv4-addr--cbd67181-b9f8-595b-8bc3-3971e34fa1cc" + + assert len(stix_objects) == 21 + + author = next(object_ for object_ in stix_objects if object_.id == author_id) + assert author.type == "identity" + assert author.name == "Censys EnrichmentAPIs Connector" + assert author.identity_class == "organization" + + marking = next(object_ for object_ in stix_objects if object_.id == marking_id) + assert marking.type == "marking-definition" + assert marking.definition == {"statement": "custom"} + assert marking.definition_type == "statement" + assert marking.x_opencti_definition == "TLP:CLEAR" + assert marking.x_opencti_definition_type == "TLP" + + for object_ in stix_objects[2:]: + assert author_id in { + getattr(object_, "created_by_ref", None), + getattr(object_, "x_opencti_created_by_ref", None), + } + assert object_.object_marking_refs == [marking_id] + + locations = { + object_.id: object_ for object_ in stix_objects if object_.type == "location" + } + assert set(locations) == { + "location--718026de-1217-54e3-9915-ebddd72ffc2b", + "location--834c5189-3715-561b-b68a-e835372d05ff", + "location--50b4cef5-9f48-5ae6-9777-8e1217b8f83d", + "location--6004efb1-d850-551c-af0d-4717244377a8", + } + assert locations["location--718026de-1217-54e3-9915-ebddd72ffc2b"].city == "Brisbane" + assert locations["location--834c5189-3715-561b-b68a-e835372d05ff"].region == "Oceania" + administrative_area = locations["location--50b4cef5-9f48-5ae6-9777-8e1217b8f83d"] + assert administrative_area.administrative_area == "Queensland" + assert (administrative_area.latitude, administrative_area.longitude) == (-27.47, 153.02) + assert locations["location--6004efb1-d850-551c-af0d-4717244377a8"].country == "Australia" + + hostnames = { + object_.id: object_ for object_ in stix_objects if object_.type == "hostname" + } + assert {object_.value for object_ in hostnames.values()} == { + "guestcontroller.sa.gov.au", + "matrix.cyops.cloud", + } + + organization = next( + object_ + for object_ in stix_objects + if object_.id == "identity--a7d63be9-7173-560e-9723-a5040d771c2c" + ) + assert organization.name == "CLOUDFLARENET" + assert organization.identity_class == "organization" + + autonomous_system = next( + object_ + for object_ in stix_objects + if object_.id == "autonomous-system--0204c07d-e4dd-5f14-a3d5-c93cb1c5a9fc" + ) + assert autonomous_system.name == "CLOUDFLARENET" + assert autonomous_system.number == 13335 + assert autonomous_system.x_opencti_description == "CLOUDFLARENET" + + relationships = { + (object_.relationship_type, str(object_.source_ref), str(object_.target_ref)) + for object_ in stix_objects + if object_.type == "relationship" + } + assert relationships == { + ("located-at", ip_id, "location--718026de-1217-54e3-9915-ebddd72ffc2b"), + ("located-at", ip_id, "location--834c5189-3715-561b-b68a-e835372d05ff"), + ("located-at", ip_id, "location--50b4cef5-9f48-5ae6-9777-8e1217b8f83d"), + ("located-at", ip_id, "location--6004efb1-d850-551c-af0d-4717244377a8"), + ("resolves-to", "hostname--2aa1a527-f7f9-59c6-aa42-716270bccb27", ip_id), + ("resolves-to", "hostname--21f6b21c-7cae-55af-b29b-54628a2c56f4", ip_id), + ("related-to", ip_id, "identity--a7d63be9-7173-560e-9723-a5040d771c2c"), + ("belongs-to", ip_id, "autonomous-system--0204c07d-e4dd-5f14-a3d5-c93cb1c5a9fc"), + ( + "related-to", + "autonomous-system--0204c07d-e4dd-5f14-a3d5-c93cb1c5a9fc", + "identity--a7d63be9-7173-560e-9723-a5040d771c2c", + ), + ( + "related-to", + "autonomous-system--0204c07d-e4dd-5f14-a3d5-c93cb1c5a9fc", + "location--6004efb1-d850-551c-af0d-4717244377a8", + ), + } + + note = next(object_ for object_ in stix_objects if object_.type == "note") + assert note.abstract == "Service information on port 443 (Unknown)" + assert note.authors == ["Censys EnrichmentAPIs Connector"] + assert "- Scan Time: 2025-11-03T12:35:48Z" in note.content + assert "- Labels" in note.content + assert " - REMOTE_ACCESS" in note.content + assert note.object_refs == [ip_id] diff --git a/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_host_enrichment.py b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_host_enrichment.py new file mode 100644 index 00000000000..ea090d40a9f --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/censys_enrichmentapis/test_host_enrichment.py @@ -0,0 +1,791 @@ +import stix2 +from censys_enrichmentapis.client import Client +from censys_enrichmentapis.converters.host import HostConverter +from censys_platform import HostEnrichment, HostEnrichmentService, Label, Reputation +from censys_platform.models.reputation_evidence import ReputationEvidence, ReputationEvidenceFeature + + +def _get_host_245_52_sample() -> dict: + """Generate mock sample data for host 193.233.245.52 with OpenSSH vulnerabilities.""" + return { + "ip": "193.233.245.52", + "services": [ + { + "port": 22, + "protocol": "SSH", + "scan_time": "2026-08-29T19:57:34.651481661Z", + "labels": [], + "vulns": [ + { + "id": "CVE-2026-35385", + "name": "CVE-2026-35385", + "severity": "HIGH", + "metrics": { + "cvss_v31": { + "score": 7.5, + "vector": "CVSS:3.1/AV:N/AC:H/PR:N/UI:R/S:U/C:H/I:H/A:H", + "components": { + "attack_vector": "NETWORK", + "attack_complexity": "HIGH", + "privileges_required": "NONE", + "user_interaction": "REQUIRED", + "scope": "UNCHANGED", + "confidentiality": "HIGH", + "integrity": "HIGH", + "availability": "HIGH" + } + }, + "epss": { + "score": 0.006, + "percentile": 0.466 + } + }, + "evidence": [ + { + "found_value": "cpe:2.3:a:openbsd:openssh:10.2p1:*:*:*:*:*:*:*" + } + ] + }, + { + "id": "CVE-2026-35386", + "name": "CVE-2026-35386", + "severity": "LOW", + "metrics": { + "cvss_v31": { + "score": 3.6, + "vector": "CVSS:3.1/AV:L/AC:H/PR:L/UI:N/S:U/C:L/I:L/A:N", + "components": { + "attack_vector": "LOCAL", + "attack_complexity": "HIGH", + "privileges_required": "LOW", + "user_interaction": "NONE", + "scope": "UNCHANGED", + "confidentiality": "LOW", + "integrity": "LOW", + "availability": "NONE" + } + }, + "epss": { + "score": 0.003, + "percentile": 0.242 + } + }, + "evidence": [ + { + "found_value": "cpe:2.3:a:openbsd:openssh:10.2p1:*:*:*:*:*:*:*" + } + ] + }, + { + "id": "CVE-2026-35387", + "name": "CVE-2026-35387", + "severity": "LOW", + "metrics": { + "cvss_v31": { + "score": 3.1, + "vector": "CVSS:3.1/AV:N/AC:H/PR:L/UI:N/S:U/C:N/I:L/A:N", + "components": { + "attack_vector": "NETWORK", + "attack_complexity": "HIGH", + "privileges_required": "LOW", + "user_interaction": "NONE", + "scope": "UNCHANGED", + "confidentiality": "NONE", + "integrity": "LOW", + "availability": "NONE" + } + }, + "epss": { + "score": 0.002, + "percentile": 0.146 + } + }, + "evidence": [ + { + "found_value": "cpe:2.3:a:openbsd:openssh:10.2p1:*:*:*:*:*:*:*" + } + ] + } + ], + "software": [ + { + "cpe": "cpe:2.3:a:openbsd:openssh:10.2p1:*:*:*:*:*:*:*", + "product": "openssh", + "vendor": "openbsd", + "version": "10.2p1" + } + ] + } + ] + } + + +def test_converter_adds_threat_names_to_primary_observable_labels() -> None: + """Verify that threat names are added to primary observable labels with Censys_ prefix in snake_case.""" + service = HostEnrichmentService( + port=22, + protocol="SSH", + scan_time="2026-08-31T13:12:28Z", + ) + service.__dict__["threats"] = [ + { + "id": "THREAT-SSH", + "name": "Exposed SSH Service", + "source": "censys", + "confidence": 1.0, + "type": ["remote_access"], + "tactic": ["initial_access"], + "evidence": [], + "malware": {} + }, + { + "id": "THREAT-WEAK-CREDS", + "name": "Weak Credentials", + "source": "censys", + "confidence": 0.8, + "type": ["credential_access"], + "tactic": ["credential_access"], + "evidence": [], + "malware": {} + } + ] + + host = HostEnrichment(services=[service]) + converter = HostConverter() + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in converter.to_stix( + observable=stix2.IPv4Address(value="1.1.1.1"), + data=host + ) + ] + + # Verify threat names are in primary observable labels with Censys_ prefix + assert "Censys_Exposed_SSH_Service" in converter.primary_observable_labels + assert "Censys_Weak_Credentials" in converter.primary_observable_labels + + +def test_converter_host_enrichment_adds_service_labels_as_note() -> None: + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix( + observable=stix2.IPv4Address(value="1.1.1.1"), + data=HostEnrichment( + services=[ + HostEnrichmentService( + port=22, + scan_time="2025-11-03T12:35:48Z", + labels=[Label(value="REMOTE_ACCESS")], + ) + ] + ), + ) + ] + + notes = [stix_object for stix_object in stix_objects if stix_object.type == "note"] + assert len(notes) == 1 + assert notes[0].abstract == "Service information on port 22 (Unknown)" + assert "- Scan Time: 2025-11-03T12:35:48Z" in notes[0].content + assert "- Labels" in notes[0].content + assert " - REMOTE_ACCESS" in notes[0].content + + +def test_converter_adds_external_reputation_note() -> None: + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix( + observable=stix2.IPv4Address(value="1.1.1.1"), + data=HostEnrichment( + reputation=Reputation( + score=0.42, + label="MEDIUM_RISK", + model_version="0.1.0", + ) + ), + ) + ] + + note = next(stix_object for stix_object in stix_objects if stix_object.type == "note") + assert note.abstract == "Censys host reputation" + # Check the content contains the expected parts (no evidence features since none provided) + assert "- Score: 42" in note.content + assert "- Label: MEDIUM_RISK" in note.content + assert "- Model version: 0.1.0" in note.content + # Should not have evidence features section + assert "**Evidence Features:**" not in note.content + assert note.labels == ["MEDIUM_RISK"] + assert note.note_types == ["external"] + + +def test_converter_adds_reputation_note_with_evidence() -> None: + """Verify that reputation note includes evidence features. + + Uses realistic reputation structure from Censys API with ReputationEvidence objects. + """ + # Create reputation with evidence features matching Censys API structure + host_enrichment = HostEnrichment( + reputation=Reputation( + score=0.704, + label="SUSPICIOUS", + model_version="2.0.0", + ) + ) + + # Inject evidence as ReputationEvidence objects with feature field + # This mirrors the actual API response structure + host_enrichment.reputation.__dict__["evidence"] = [ + ReputationEvidence( + feature=ReputationEvidenceFeature( + id="max_port", + name="Max Port", + value="49093", + contribution=8.708259985239051, + category="service_surface" + ) + ), + ReputationEvidence( + feature=ReputationEvidenceFeature( + id="high_port_ratio", + name="High Port Ratio", + value="0.875", + contribution=5.341544169693149, + category="service_surface" + ) + ), + ReputationEvidence( + feature=ReputationEvidenceFeature( + id="avg_epss_score", + name="Avg EPSS Score", + value="0.0937", + contribution=-3.738838369305972, + category="vulnerability_exposure" + ) + ), + ] + + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix( + observable=stix2.IPv4Address(value="104.168.107.43"), + data=host_enrichment, + ) + ] + + note = next(stix_object for stix_object in stix_objects if stix_object.type == "note") + assert note.abstract == "Censys host reputation" + + # Verify score information + assert "- Score: 70" in note.content + assert "- Label: SUSPICIOUS" in note.content + assert "- Model version: 2.0.0" in note.content + + # Verify evidence features are included with proper formatting + assert "**Evidence Features:**" in note.content + assert "| Feature | Value | Contribution | Category |" in note.content + assert "| Max Port | 49093 | +8.71% | service_surface |" in note.content + assert "| High Port Ratio | 0.875 | +5.34% | service_surface |" in note.content + assert ( + "| Avg EPSS Score | 0.0937 | -3.74% | " + "vulnerability_exposure |" in note.content + ) + + assert note.labels == ["SUSPICIOUS"] + assert note.note_types == ["external"] + + +def test_converter_links_service_cves_through_software() -> None: + sample = _get_host_245_52_sample() + sample["services"][0]["vulns"] = sample["services"][0]["vulns"][:2] + + # Match the generated SDK response: it deserializes recognised fields and + # ignores service.vulns, while Client restores those raw service fields. + host = HostEnrichment.model_validate(sample) + Client._restore_service_fields(host, {"result": {"result": {"resource": sample}}}) + host = HostEnrichment(services=host.services) + + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix( + observable=stix2.IPv4Address(value="193.233.245.52"), data=host + ) + ] + + software = next(obj for obj in stix_objects if obj.type == "software") + vulnerabilities = [obj for obj in stix_objects if obj.type == "vulnerability"] + has_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" and obj.relationship_type == "has" + ] + + assert software.name == "openssh" + assert software.vendor == "openbsd" + assert software.version == "10.2p1" + assert [vulnerability.name for vulnerability in vulnerabilities] == [ + "CVE-2026-35385", + "CVE-2026-35386", + ] + assert all(relationship.source_ref == software.id for relationship in has_relationships) + assert {relationship.target_ref for relationship in has_relationships} == { + vulnerability.id for vulnerability in vulnerabilities + } + assert vulnerabilities[0].x_opencti_cvss_base_score == 7.5 + assert vulnerabilities[0].x_opencti_epss_score == 0.006 + + +def test_converter_creates_complete_vulnerability_chain() -> None: + """Verify the complete STIX relationship path: IP -> Software -> CVE.""" + sample = _get_host_245_52_sample() + sample["services"][0]["vulns"] = sample["services"][0]["vulns"][:2] + host = HostEnrichment.model_validate(sample) + Client._restore_service_fields(host, {"result": {"result": {"resource": sample}}}) + host = HostEnrichment(services=host.services) + + observable = stix2.IPv4Address(value="193.233.245.52") + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix(observable=observable, data=host) + ] + + # Verify that relationships from IP to Software exist + ip_to_software_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" + and obj.relationship_type == "related-to" + and any( + obj_id in str(obj.source_ref) and "software" in str(obj.target_ref) + for obj_id in [observable.id] + ) + ] + + software = next(obj for obj in stix_objects if obj.type == "software") + vulnerabilities = [obj for obj in stix_objects if obj.type == "vulnerability"] + + # Verify complete chain: source (IP reference) -> Software + assert any( + rel.source_ref == observable.id and rel.target_ref == software.id + for rel in stix_objects + if rel.type == "relationship" and rel.relationship_type == "related-to" + ), "No RELATED_TO relationship found from IP observable to Software" + + # Verify the CVE chain: Software -> Vulnerability + has_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" and obj.relationship_type == "has" + ] + assert len(has_relationships) == 2, f"Expected 2 HAS relationships, got {len(has_relationships)}" + assert all( + rel.source_ref == software.id for rel in has_relationships + ), "Not all CVE relationships originate from the same Software" + + +def test_converter_deduplicates_software_across_multiple_cves() -> None: + """Verify that multiple CVEs with the same CPE share a single Software object.""" + sample = _get_host_245_52_sample() + # Use 3 CVEs - all reference the same OpenSSH CPE, ensuring deduplication + sample["services"][0]["vulns"] = sample["services"][0]["vulns"][:3] + host = HostEnrichment.model_validate(sample) + Client._restore_service_fields(host, {"result": {"result": {"resource": sample}}}) + host = HostEnrichment(services=host.services) + + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix( + observable=stix2.IPv4Address(value="193.233.245.52"), data=host + ) + ] + + # Should have exactly one Software object for all 3 CVEs + software_objects = [obj for obj in stix_objects if obj.type == "software"] + assert len(software_objects) == 1, f"Expected 1 Software object, got {len(software_objects)}" + + # All HAS relationships should point to the same Software + has_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" and obj.relationship_type == "has" + ] + assert len(has_relationships) == 3, f"Expected 3 HAS relationships, got {len(has_relationships)}" + assert all( + rel.source_ref == software_objects[0].id for rel in has_relationships + ), "All CVEs should be linked to the same Software object" + + +def test_converter_deduplicates_cve_with_multiple_cpe_evidence() -> None: + """One CVE object should be related to every software named by its evidence.""" + sample = _get_host_245_52_sample() + vulnerability = sample["services"][0]["vulns"][0] + second_cpe = "cpe:2.3:a:example:second_product:2.0:*:*:*:*:*:*:*" + vulnerability["evidence"].append({"found_value": second_cpe}) + sample["services"][0]["vulns"] = [vulnerability] + sample["services"][0]["software"].append( + { + "cpe": second_cpe, + "product": "second_product", + "vendor": "example", + "version": "2.0", + } + ) + + host = HostEnrichment.model_validate(sample) + Client._restore_service_fields(host, {"result": {"result": {"resource": sample}}}) + host = HostEnrichment(services=host.services) + + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix( + observable=stix2.IPv4Address(value="193.233.245.52"), data=host + ) + ] + + software_objects = [obj for obj in stix_objects if obj.type == "software"] + vulnerabilities = [obj for obj in stix_objects if obj.type == "vulnerability"] + has_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" and obj.relationship_type == "has" + ] + + assert len(software_objects) == 2 + assert len(vulnerabilities) == 1 + assert len(has_relationships) == 2 + assert {relationship.source_ref for relationship in has_relationships} == { + software.id for software in software_objects + } + assert {relationship.target_ref for relationship in has_relationships} == { + vulnerabilities[0].id + } + + +def test_converter_creates_software_from_cpe_when_not_in_service() -> None: + """Verify that Software can be created from CPE evidence when not in service.software.""" + # Create a service with CVE evidence but no pre-defined software + service = HostEnrichmentService( + port=443, + protocol="HTTPS", + scan_time="2026-01-01T00:00:00Z", + ) + + # Manually attach vulns as the client would do via _restore_service_fields + service.__dict__["vulns"] = [ + { + "id": "CVE-2026-12345", + "name": "CVE-2026-12345", + "severity": "MEDIUM", + "evidence": [ + { + "found_value": "cpe:2.3:a:vendor:product:1.0:*:*:*:*:*:*:*" + } + ], + "metrics": { + "cvss_v31": {"score": 5.5}, + "epss": {"score": 0.05}, + }, + } + ] + service.__dict__["software"] = [] + + host = HostEnrichment(services=[service]) + + stix_objects = [ + octi_object.to_stix2_object() + for octi_object in HostConverter().to_stix( + observable=stix2.IPv4Address(value="10.0.0.1"), data=host + ) + ] + + # Should create Software from CPE evidence + software_objects = [obj for obj in stix_objects if obj.type == "software"] + assert len(software_objects) == 1, f"Expected 1 Software created from CPE, got {len(software_objects)}" + assert software_objects[0].vendor == "vendor" + assert software_objects[0].name == "product" + assert software_objects[0].version == "1.0" + + # Verify the CVE is linked to this generated Software + vulnerabilities = [obj for obj in stix_objects if obj.type == "vulnerability"] + assert len(vulnerabilities) == 1 + has_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" and obj.relationship_type == "has" + ] + assert len(has_relationships) == 1 + assert has_relationships[0].source_ref == software_objects[0].id + assert has_relationships[0].target_ref == vulnerabilities[0].id + + +def test_converter_creates_malware_from_threats() -> None: + """Verify that Malware objects are created from service threats.""" + service = HostEnrichmentService( + port=4224, + protocol="HTTP", + scan_time="2026-08-31T11:11:35Z", + ) + service.__dict__["threats"] = [ + { + "id": "THREAT-0188", + "name": "ShellInABox", + "source": "censys", + "confidence": 0.5, + "type": ["webshell"], + "tactic": ["persistence"], + "evidence": [ + { + "data_path": "http.html_title", + "found_value": "Shell In A Box" + } + ], + "malware": { + "id": "MALWARE-188", + "primary_name": "ShellInABox", + "all_names": ["ShellInABox"], + "last_updated_at": "2025-05-01T00:00:00Z" + } + } + ] + + host = HostEnrichment(services=[service]) + stix_objects = [ + obj.to_stix2_object() + for obj in HostConverter().to_stix( + observable=stix2.IPv4Address(value="37.187.119.91"), data=host + ) + ] + + # Verify Malware object is created + malware = [obj for obj in stix_objects if obj.type == "malware"] + assert len(malware) == 1 + assert malware[0].name == "ShellInABox" + assert "ShellInABox" in malware[0].aliases + assert "webshell" in malware[0].malware_types + + # Verify relationship from IP to Malware + malware_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" and obj.target_ref == malware[0].id + ] + assert len(malware_relationships) == 1 + assert malware_relationships[0].relationship_type == "related-to" + + +def test_converter_creates_attack_patterns_from_threat_tactics() -> None: + """Verify that Attack-Pattern objects are created for threat tactics.""" + service = HostEnrichmentService( + port=7070, + protocol="FRPS", + scan_time="2026-08-31T02:32:15Z", + ) + service.__dict__["threats"] = [ + { + "id": "THREAT-519", + "name": "FRP", + "source": "censys", + "confidence": 0.75, + "type": ["security_tool"], + "tactic": ["command_and_control"], + "evidence": [ + { + "data_path": "protocol", + "found_value": "FRPS" + } + ], + "malware": {} + } + ] + + host = HostEnrichment(services=[service]) + stix_objects = [ + obj.to_stix2_object() + for obj in HostConverter().to_stix( + observable=stix2.IPv4Address(value="104.168.107.43"), data=host + ) + ] + + # Verify Attack-Pattern object is created + attack_patterns = [obj for obj in stix_objects if obj.type == "attack-pattern"] + assert len(attack_patterns) == 1 + assert "COMMAND AND CONTROL" in attack_patterns[0].name + + # Verify MITRE ATT&CK reference + assert attack_patterns[0].external_references + assert attack_patterns[0].external_references[0].source_name == "mitre-attack" + assert "TA0011" in attack_patterns[0].external_references[0].external_id + + # Verify relationship from IP to Attack-Pattern + pattern_relationships = [ + obj + for obj in stix_objects + if obj.type == "relationship" and obj.target_ref == attack_patterns[0].id + ] + assert len(pattern_relationships) == 1 + + +def test_converter_creates_threat_notes_with_evidence() -> None: + """Verify that threat Notes are created with detailed evidence.""" + service = HostEnrichmentService( + port=4224, + protocol="HTTP", + scan_time="2026-08-31T11:11:35Z", + ) + service.__dict__["threats"] = [ + { + "id": "THREAT-0188", + "name": "ShellInABox", + "source": "censys", + "confidence": 0.5, + "type": ["webshell"], + "tactic": ["persistence"], + "evidence": [ + { + "data_path": "http.html_title", + "found_value": "Shell In A Box" + } + ], + "malware": { + "primary_name": "ShellInABox", + "all_names": ["ShellInABox"], + "last_updated_at": "2025-05-01T00:00:00Z" + } + } + ] + + host = HostEnrichment(services=[service]) + stix_objects = [ + obj.to_stix2_object() + for obj in HostConverter().to_stix( + observable=stix2.IPv4Address(value="37.187.119.91"), data=host + ) + ] + + # Verify Threat Note is created + threat_notes = [ + obj for obj in stix_objects + if obj.type == "note" and "Threat" in obj.abstract + ] + assert len(threat_notes) == 1 + note = threat_notes[0] + + # Verify threat summary is rendered as a key/value table. + assert "| Key | Value |" in note.content + assert "| Threat ID | THREAT-0188 |" in note.content + assert "| Name | ShellInABox |" in note.content + assert "| Threat Types | webshell |" in note.content + assert "| Tactics | Persistence |" in note.content + assert ( + "[View this host 37.187.119.91 on Censys Platform]" + "(https://platform.censys.io/hosts/37.187.119.91)" + in note.content + ) + assert "| Source | censys |" not in note.content + assert "0.5" not in note.content + assert "Shell In A Box" in note.content + assert "2025-05-01" in note.content + + # Verify note labels + assert "webshell" in note.labels + + +def test_converter_handles_multiple_threats_per_service() -> None: + """Verify that multiple threats on one service create multiple objects.""" + service = HostEnrichmentService( + port=9080, + protocol="HTTP", + scan_time="2026-08-31T13:12:28Z", + ) + service.__dict__["threats"] = [ + { + "id": "THREAT-519", + "name": "FRP", + "source": "censys", + "confidence": 0.75, + "type": ["security_tool"], + "tactic": ["command_and_control"], + "evidence": [], + "malware": {} + }, + { + "id": "THREAT-520", + "name": "Reverse Shell Proxy", + "source": "censys", + "confidence": 0.8, + "type": ["proxy"], + "tactic": ["lateral_movement"], + "evidence": [], + "malware": { + "primary_name": "Reverse Shell", + "all_names": ["Reverse Shell", "RevShell"] + } + } + ] + + host = HostEnrichment(services=[service]) + stix_objects = [ + obj.to_stix2_object() + for obj in HostConverter().to_stix( + observable=stix2.IPv4Address(value="104.168.107.43"), data=host + ) + ] + + # Verify 2 threat notes + threat_notes = [ + obj for obj in stix_objects + if obj.type == "note" and "Threat" in obj.abstract + ] + assert len(threat_notes) == 2 + + # Verify 1 malware (only second threat has malware) + malware = [obj for obj in stix_objects if obj.type == "malware"] + assert len(malware) == 1 + assert malware[0].name == "Reverse Shell" + + # Verify 2 attack patterns + attack_patterns = [obj for obj in stix_objects if obj.type == "attack-pattern"] + assert len(attack_patterns) == 2 + pattern_names = {p.name for p in attack_patterns} + assert "COMMAND AND CONTROL" in pattern_names + assert "LATERAL MOVEMENT" in pattern_names + + +def test_converter_handles_service_with_both_vulns_and_threats() -> None: + """Verify that services can have both vulnerabilities and threats.""" + sample = _get_host_245_52_sample() + # Limit to 1 CVE and add a threat + sample["services"][0]["vulns"] = sample["services"][0]["vulns"][:1] + sample["services"][0]["threats"] = [ + { + "id": "THREAT-SSH", + "name": "Exposed SSH", + "source": "censys", + "confidence": 1.0, + "type": ["remote_access"], + "tactic": ["initial_access"], + "evidence": [{"data_path": "protocol", "found_value": "SSH"}], + "malware": {} + } + ] + + host = HostEnrichment.model_validate(sample) + Client._restore_service_fields(host, {"result": {"result": {"resource": sample}}}) + + stix_objects = [ + obj.to_stix2_object() + for obj in HostConverter().to_stix( + observable=stix2.IPv4Address(value="193.233.245.52"), data=host + ) + ] + + # Verify both CVE and threat are present + vulnerabilities = [obj for obj in stix_objects if obj.type == "vulnerability"] + threat_notes = [ + obj for obj in stix_objects + if obj.type == "note" and "Threat" in obj.abstract + ] + + assert len(vulnerabilities) == 1 + assert vulnerabilities[0].name == "CVE-2026-35385" + + assert len(threat_notes) == 1 + assert "Exposed SSH" in threat_notes[0].content diff --git a/internal-enrichment/censys-enrichmentapis/tests/conftest.py b/internal-enrichment/censys-enrichmentapis/tests/conftest.py new file mode 100644 index 00000000000..5ee8fc0e226 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/conftest.py @@ -0,0 +1,4 @@ +import os +import sys + +sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src")) diff --git a/internal-enrichment/censys-enrichmentapis/tests/test-requirements.txt b/internal-enrichment/censys-enrichmentapis/tests/test-requirements.txt new file mode 100644 index 00000000000..8a537078aa0 --- /dev/null +++ b/internal-enrichment/censys-enrichmentapis/tests/test-requirements.txt @@ -0,0 +1,5 @@ +# Main dependencies needs to be installed +-r ../src/requirements.txt +pytest==9.0.3 +pytest-mock==3.15.1 +factory-boy==3.3.3