From 8cc49aab59949e70a8d94a612ff3a1c06f3f89cc Mon Sep 17 00:00:00 2001 From: mateen993 Date: Wed, 15 Jul 2026 15:37:38 +0500 Subject: [PATCH 1/2] feat(ipgeolocation): add IPGeolocation.io internal enrichment connector --- internal-enrichment/ipgeolocation/Dockerfile | 29 + internal-enrichment/ipgeolocation/README.md | 194 ++++++ .../ipgeolocation/config.yml.sample | 47 ++ .../ipgeolocation/docker-compose.yml | 51 ++ .../ipgeolocation/docs/ARCHITECTURE.md | 91 +++ .../ipgeolocation/docs/RISK_SCORING.md | 82 +++ .../ipgeolocation/docs/STIX_MAPPING.md | 72 ++ .../ipgeolocation/docs/TESTING_GUIDE.md | 314 +++++++++ .../ipgeolocation/entrypoint.sh | 5 + .../ipgeolocation/requirements.txt | 5 + .../ipgeolocation/src/__init__.py | 3 + .../ipgeolocation/src/api_client.py | 229 +++++++ .../ipgeolocation/src/config.py | 293 ++++++++ .../ipgeolocation/src/connector.py | 201 ++++++ internal-enrichment/ipgeolocation/src/main.py | 34 + .../ipgeolocation/src/markdown_generator.py | 427 ++++++++++++ .../ipgeolocation/src/models.py | 401 +++++++++++ .../ipgeolocation/src/risk_scorer.py | 214 ++++++ .../ipgeolocation/src/stix_mapper.py | 635 ++++++++++++++++++ .../ipgeolocation/tests/__init__.py | 0 .../ipgeolocation/tests/conftest.py | 46 ++ .../ipgeolocation/tests/mock_responses.py | 258 +++++++ .../ipgeolocation/tests/test_api_client.py | 142 ++++ .../tests/test_markdown_generator.py | 71 ++ .../ipgeolocation/tests/test_risk_scorer.py | 79 +++ .../ipgeolocation/tests/test_stix_mapper.py | 179 +++++ 26 files changed, 4102 insertions(+) create mode 100644 internal-enrichment/ipgeolocation/Dockerfile create mode 100644 internal-enrichment/ipgeolocation/README.md create mode 100644 internal-enrichment/ipgeolocation/config.yml.sample create mode 100644 internal-enrichment/ipgeolocation/docker-compose.yml create mode 100644 internal-enrichment/ipgeolocation/docs/ARCHITECTURE.md create mode 100644 internal-enrichment/ipgeolocation/docs/RISK_SCORING.md create mode 100644 internal-enrichment/ipgeolocation/docs/STIX_MAPPING.md create mode 100644 internal-enrichment/ipgeolocation/docs/TESTING_GUIDE.md create mode 100644 internal-enrichment/ipgeolocation/entrypoint.sh create mode 100644 internal-enrichment/ipgeolocation/requirements.txt create mode 100644 internal-enrichment/ipgeolocation/src/__init__.py create mode 100644 internal-enrichment/ipgeolocation/src/api_client.py create mode 100644 internal-enrichment/ipgeolocation/src/config.py create mode 100644 internal-enrichment/ipgeolocation/src/connector.py create mode 100644 internal-enrichment/ipgeolocation/src/main.py create mode 100644 internal-enrichment/ipgeolocation/src/markdown_generator.py create mode 100644 internal-enrichment/ipgeolocation/src/models.py create mode 100644 internal-enrichment/ipgeolocation/src/risk_scorer.py create mode 100644 internal-enrichment/ipgeolocation/src/stix_mapper.py create mode 100644 internal-enrichment/ipgeolocation/tests/__init__.py create mode 100644 internal-enrichment/ipgeolocation/tests/conftest.py create mode 100644 internal-enrichment/ipgeolocation/tests/mock_responses.py create mode 100644 internal-enrichment/ipgeolocation/tests/test_api_client.py create mode 100644 internal-enrichment/ipgeolocation/tests/test_markdown_generator.py create mode 100644 internal-enrichment/ipgeolocation/tests/test_risk_scorer.py create mode 100644 internal-enrichment/ipgeolocation/tests/test_stix_mapper.py diff --git a/internal-enrichment/ipgeolocation/Dockerfile b/internal-enrichment/ipgeolocation/Dockerfile new file mode 100644 index 00000000000..1daa8417487 --- /dev/null +++ b/internal-enrichment/ipgeolocation/Dockerfile @@ -0,0 +1,29 @@ +FROM python:3.12-slim + +# Metadata +LABEL maintainer="IPGeolocation.io OpenCTI Connector" +LABEL org.opencontainers.image.title="OpenCTI IPGeolocation.io Enrichment Connector" +LABEL org.opencontainers.image.description="Internal enrichment connector for IPGeolocation.io v3 APIs" + +# Security: run as non-root +RUN useradd -ms /bin/bash connector + +# Install system dependency required by pycti (python-magic) +RUN apt-get update && \ + apt-get install -y --no-install-recommends libmagic1 && \ + rm -rf /var/lib/apt/lists/* + +# Install dependencies +WORKDIR /opt/opencti-connector +COPY requirements.txt . +RUN pip install --no-cache-dir --break-system-packages -r requirements.txt + +# Copy source +COPY src/ ./src/ +COPY entrypoint.sh . +RUN chmod +x entrypoint.sh + +# Switch to non-root +USER connector + +ENTRYPOINT ["/opt/opencti-connector/entrypoint.sh"] diff --git a/internal-enrichment/ipgeolocation/README.md b/internal-enrichment/ipgeolocation/README.md new file mode 100644 index 00000000000..392875ef201 --- /dev/null +++ b/internal-enrichment/ipgeolocation/README.md @@ -0,0 +1,194 @@ +# OpenCTI IPGeolocation.io Enrichment Connector + +Production-quality **internal enrichment** connector that transforms +IPGeolocation.io v3 threat intelligence into semantically rich STIX 2.1 +knowledge within OpenCTI. + +## Features + +### Data Sources + +Consumes four IPGeolocation.io v3 APIs in a single enrichment: + +| API | Data | Credits | +|--------------------|----------------------------------------------------------|---------| +| IP Location API | Geo, ASN, company, timezone, network, currency, hostname | 1 | +| IP Security API | Threat score, VPN, proxy, TOR, bot, spam, attacker flags | 2 | +| ASN API | Routes, peers, upstreams, downstreams, WHOIS, registry | 1 | +| Abuse Contact API | Abuse email, phone, organization, address, route | 1 | + +### Enrichment Scope + +| Observable Type | Supported | +|-----------------|-----------| +| IPv4-Addr | ✅ | +| IPv6-Addr | ✅ | + +### STIX Objects Created + +| Object Type | Description | +|--------------------|------------------------------------------------| +| Location (Country) | Country with ISO code, continent, EU membership| +| Location (City) | City with lat/lon coordinates, postal code | +| Autonomous System | ASN number, name, registry, allocation date | +| Identity (Org) | ISP/Company/Hosting provider | +| Identity (Cloud) | Cloud infrastructure provider | +| Identity (Abuse) | Abuse contact with email, phone, address | +| Domain-Name | Hostname resolved from the IP | +| Indicator | STIX pattern with threat score (configurable) | +| Note | Rich markdown analyst summary | +| Opinion | Risk opinion (low→strongly-disagree, etc.) | +| Relationship | located-at, belongs-to, resolves-to, etc. | +| Labels | vpn, tor, proxy, known-attacker, risk:high... | + +### Differentiators + +Unlike basic IP enrichment connectors, this connector provides: + +1. **Security Narrative** — instead of raw booleans, generates analyst text + like *"This address originates from a commercial VPN (Nord VPN) hosted on + cloud infrastructure and has been flagged as known attacker."* +2. **Risk Explanation** — every score comes with a human-readable explanation + listing contributing factors and their weights. +3. **Infrastructure Profile** — categorizes the IP as Hosting, ISP, Business, + Education, Mobile, etc. +4. **Network Context** — ASN peers, upstreams, downstreams, routes, WHOIS. +5. **Abuse Workflow** — clickable abuse contact note with email, phone, and + recommended reporting target. +6. **Geo Intelligence** — coordinates, accuracy radius, confidence, timezone. +7. **Timeline** — when the IP was last seen as VPN, proxy, when ASN allocated. +8. **Confidence Explanation** — why the enrichment is reliable. +9. **Credit Optimization** — single-call vs dedicated-endpoint modes. + +--- + +## Quick Start + +### Docker Compose (recommended) + +1. Copy `docker-compose.yml` into your OpenCTI deployment. +2. Set environment variables (or use `.env`): + +```env +OPENCTI_ADMIN_TOKEN=your-opencti-token +IPGEOLOCATION_CONNECTOR_ID=a-uuid-v4 +IPGEOLOCATION_API_KEY=your-ipgeolocation-key +``` + +3. Start: + +```bash +docker compose up -d connector-ipgeolocation +``` + +### Manual (Python) + +```bash +pip install -r requirements.txt +cp config.yml.sample config.yml +# Edit config.yml with your credentials +python -m src.main +``` + +--- + +## Configuration Reference + +### Core Settings + +| Env Variable | YAML Path | Default | Description | +|----------------------------------|--------------------------------|-----------------|------------------------------------| +| `OPENCTI_URL` | `opencti.url` | — | OpenCTI platform URL | +| `OPENCTI_TOKEN` | `opencti.token` | — | API token | +| `CONNECTOR_ID` | `connector.id` | — | Unique UUIDv4 | +| `CONNECTOR_NAME` | `connector.name` | IPGeolocation.io| Display name | +| `CONNECTOR_SCOPE` | `connector.scope` | IPv4-Addr,IPv6-Addr | Observable types | +| `CONNECTOR_AUTO` | `connector.auto` | false | Auto-enrich on ingest | +| `CONNECTOR_CONFIDENCE_LEVEL` | `connector.confidence_level` | 80 | Connector confidence (0-100) | + +### API Settings + +| Env Variable | YAML Path | Default | Description | +|--------------------------------------|------------------------------------|---------|--------------------------------| +| `IPGEOLOCATION_API_KEY` | `ipgeolocation.api_key` | — | API key | +| `IPGEOLOCATION_BASE_URL` | `ipgeolocation.base_url` | https://api.ipgeolocation.io | API base | +| `IPGEOLOCATION_TIMEOUT` | `ipgeolocation.timeout` | 30 | HTTP timeout (seconds) | +| `IPGEOLOCATION_MAX_RETRIES` | `ipgeolocation.max_retries` | 3 | Max retry attempts | + +### API Modules + +| Env Variable | Default | Description | +|--------------------------------------|---------|-------------------------------------| +| `IPGEOLOCATION_USE_GEO_API` | true | Enable geolocation enrichment | +| `IPGEOLOCATION_USE_SECURITY_API` | true | Enable threat detection | +| `IPGEOLOCATION_USE_ASN_API` | true | Enable detailed ASN data | +| `IPGEOLOCATION_USE_ABUSE_API` | true | Enable abuse contact lookup | + +### Credit Optimization + +| Env Variable | Default | Description | +|--------------------------------------|---------|-------------------------------------| +| `IPGEOLOCATION_SINGLE_CALL_MODE` | true | Use unified endpoint (fewer calls) | + +**Single-call mode** (`true`): 2 HTTP calls (ipgeo+asn), ~4 credits +**Dedicated mode** (`false`): 4 HTTP calls, ~5 credits + +### Feature Toggles + +| Env Variable | Default | Description | +|-------------------------------------------|---------|----------------------------------| +| `IPGEOLOCATION_CREATE_LABELS` | true | Attach labels (vpn, tor, etc.) | +| `IPGEOLOCATION_CREATE_INDICATORS` | true | Create STIX indicators | +| `IPGEOLOCATION_CREATE_RELATIONSHIPS` | true | Create STIX relationships | +| `IPGEOLOCATION_CREATE_NOTES` | true | Create markdown analyst notes | +| `IPGEOLOCATION_CREATE_OPINIONS` | false | Create risk opinions | +| `IPGEOLOCATION_CREATE_SUMMARY` | true | Include summary in notes | + +### Thresholds + +| Env Variable | Default | Description | +|-------------------------------------------|---------|----------------------------------| +| `IPGEOLOCATION_MIN_THREAT_SCORE` | 0 | Skip enrichment below this score | +| `IPGEOLOCATION_INDICATOR_THREAT_THRESHOLD`| 50 | Create indicator only above this | + +### TLP + +| Env Variable | Default | Description | +|--------------------------------------|-----------|-----------------------------------| +| `IPGEOLOCATION_MAX_TLP` | TLP:AMBER | Max TLP to process | +| `IPGEOLOCATION_DEFAULT_MARKING` | TLP:WHITE | Default marking for created objects| + +--- + +## Risk Scoring + +The connector uses a weighted additive model. See +[docs/RISK_SCORING.md](docs/RISK_SCORING.md) for the full algorithm. + +| Score | Level | Meaning | +|-------|----------|----------------------------------------------| +| 0-20 | Low | No significant threat signals | +| 21-50 | Medium | Some anonymization or minor flags | +| 51-80 | High | Multiple threat signals present | +| 81+ | Critical | Strong indicators of malicious infrastructure| + +--- + +## Testing + +```bash +pip install pytest +pytest tests/ -v +``` + +--- + +## Architecture + +See [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) for the full component diagram. + +--- + +## License + +Apache 2.0 diff --git a/internal-enrichment/ipgeolocation/config.yml.sample b/internal-enrichment/ipgeolocation/config.yml.sample new file mode 100644 index 00000000000..f7853e6aba7 --- /dev/null +++ b/internal-enrichment/ipgeolocation/config.yml.sample @@ -0,0 +1,47 @@ +opencti: + url: "http://localhost:8080" + token: "ChangeMe" + +connector: + id: "ChangeMe" # UUIDv4 + type: "INTERNAL_ENRICHMENT" + name: "IPGeolocation.io" + scope: "IPv4-Addr,IPv6-Addr" + auto: false + confidence_level: 80 + log_level: "info" + update_existing_data: true + +ipgeolocation: + api_key: "ChangeMe" + base_url: "https://api.ipgeolocation.io" + timeout: 30 + max_retries: 3 + retry_delay: 2 + + # API modules + use_geo_api: true + use_security_api: true + use_asn_api: true + use_abuse_api: true + + # Credit optimization + # true = single /v3/ipgeo?include=security,abuse call (fewer HTTP requests) + # false = call /v3/security, /v3/asn, /v3/abuse separately + single_call_mode: true + + # Enrichment features + create_labels: true + create_indicators: true + create_relationships: true + create_notes: true + create_opinions: false + create_summary: true + + # Thresholds + min_threat_score: 0 # Skip IPs below this + indicator_threat_threshold: 50 # Create indicator only above this + + # TLP + max_tlp: "TLP:AMBER" + default_marking: "TLP:WHITE" diff --git a/internal-enrichment/ipgeolocation/docker-compose.yml b/internal-enrichment/ipgeolocation/docker-compose.yml new file mode 100644 index 00000000000..4ec04946a0b --- /dev/null +++ b/internal-enrichment/ipgeolocation/docker-compose.yml @@ -0,0 +1,51 @@ +services: + connector-ipgeolocation: + build: . + image: opencti/connector-ipgeolocation:1.0.0 + environment: + # --- OpenCTI Platform --- + - OPENCTI_URL=http://opencti:8080 + - OPENCTI_TOKEN=${OPENCTI_ADMIN_TOKEN} + # --- Connector Core --- + - CONNECTOR_ID=${IPGEOLOCATION_CONNECTOR_ID} # Generate a UUIDv4 + - CONNECTOR_NAME=IPGeolocation.io + - CONNECTOR_SCOPE=IPv4-Addr,IPv6-Addr + - CONNECTOR_AUTO=false # Set true for auto-enrichment + - CONNECTOR_CONFIDENCE_LEVEL=80 + - CONNECTOR_LOG_LEVEL=info + - CONNECTOR_UPDATE_EXISTING_DATA=true + # --- IPGeolocation.io API --- + - IPGEOLOCATION_API_KEY=${IPGEOLOCATION_API_KEY} + - IPGEOLOCATION_BASE_URL=https://api.ipgeolocation.io + - IPGEOLOCATION_TIMEOUT=30 + - IPGEOLOCATION_MAX_RETRIES=3 + # --- API Modules --- + - IPGEOLOCATION_USE_GEO_API=true + - IPGEOLOCATION_USE_SECURITY_API=true + - IPGEOLOCATION_USE_ASN_API=true + - IPGEOLOCATION_USE_ABUSE_API=true + # --- Credit Optimization --- + # true = single /v3/ipgeo?include=security,abuse call (fewer HTTP requests) + # false = call /v3/security, /v3/asn, /v3/abuse separately (per-endpoint control) + - IPGEOLOCATION_SINGLE_CALL_MODE=true + # --- Enrichment Features --- + - IPGEOLOCATION_CREATE_LABELS=true + - IPGEOLOCATION_CREATE_INDICATORS=true + - IPGEOLOCATION_CREATE_RELATIONSHIPS=true + - IPGEOLOCATION_CREATE_NOTES=true + - IPGEOLOCATION_CREATE_OPINIONS=false + - IPGEOLOCATION_CREATE_SUMMARY=true + # --- Thresholds --- + - IPGEOLOCATION_MIN_THREAT_SCORE=0 # Skip IPs below this score + - IPGEOLOCATION_INDICATOR_THREAT_THRESHOLD=50 # Create indicator only above this + # --- TLP --- + - IPGEOLOCATION_MAX_TLP=TLP:AMBER + - IPGEOLOCATION_DEFAULT_MARKING=TLP:WHITE + restart: unless-stopped + networks: + - default + +networks: + default: + external: true + name: docker_default diff --git a/internal-enrichment/ipgeolocation/docs/ARCHITECTURE.md b/internal-enrichment/ipgeolocation/docs/ARCHITECTURE.md new file mode 100644 index 00000000000..b768b155abf --- /dev/null +++ b/internal-enrichment/ipgeolocation/docs/ARCHITECTURE.md @@ -0,0 +1,91 @@ +# Architecture Document + +## Overview + +The IPGeolocation.io OpenCTI Connector is an **internal enrichment** connector +that transforms IP intelligence from IPGeolocation.io v3 APIs into semantically +rich STIX 2.1 knowledge within the OpenCTI platform. + +## Component Architecture + +``` +┌──────────────────────────────────────────────────────────────────┐ +│ OpenCTI Platform │ +│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ +│ │Observable │→ │ RabbitMQ │→ │Connector │→ │ Worker │ │ +│ │ (IPv4/6) │ │ Queue │ │ Process │ │ Ingests │ │ +│ └──────────┘ └──────────┘ └────┬─────┘ └──────────┘ │ +│ │ │ +└────────────────────────────────────┼─────────────────────────────┘ + │ + ┌────────────────────────┼────────────────────────┐ + │ Connector Process (Python) │ + │ │ + │ ┌─────────┐ │ + │ │ Config │ Typed configuration from │ + │ │ │ env vars / YAML │ + │ └────┬─────┘ │ + │ │ │ + │ ┌────▼─────┐ ┌──────────────┐ │ + │ │ Connector │──►│ API Client │──► IPGeolocation.io + │ │ (orchestr)│ │ (retry/cache)│ /v3/ipgeo │ + │ └────┬──────┘ └──────────────┘ /v3/security │ + │ │ /v3/asn │ + │ │ /v3/abuse │ + │ ┌────▼──────┐ │ + │ │Risk Scorer │ Unified score 0-100 │ + │ └────┬──────┘ │ + │ │ │ + │ ┌────▼──────┐ │ + │ │STIX Mapper │ Location, ASN, Identity, │ + │ │ │ Indicator, Note, Opinion, │ + │ │ │ Relationship objects │ + │ └────┬──────┘ │ + │ │ │ + │ ┌────▼──────┐ │ + │ │ Markdown │ Analyst-readable notes │ + │ │ Generator │ │ + │ └────┬──────┘ │ + │ │ │ + │ ▼ │ + │ STIX 2.1 Bundle → OpenCTI │ + └──────────────────────────────────────────────────┘ +``` + +## Module Responsibilities + +| Module | Responsibility | +|---------------------|------------------------------------------------------| +| `config.py` | Typed configuration from env/YAML | +| `models.py` | Typed dataclasses for API responses | +| `api_client.py` | HTTP client with retry, rate-limit, credit logic | +| `connector.py` | Orchestrator: read observable → enrich → send bundle | +| `risk_scorer.py` | Normalize threat signals into unified 0-100 score | +| `stix_mapper.py` | Transform intelligence into STIX 2.1 objects | +| `markdown_generator.py` | Produce analyst-readable enrichment notes | + +## Credit Optimization + +The connector supports two API consumption modes: + +**Single-call mode** (`IPGEOLOCATION_SINGLE_CALL_MODE=true`): +- Uses `/v3/ipgeo?include=security,abuse` for one HTTP call +- Adds dedicated `/v3/asn` for richer network data +- Total: 2 HTTP calls per enrichment (4 credits typical) + +**Dedicated mode** (`IPGEOLOCATION_SINGLE_CALL_MODE=false`): +- Calls `/v3/ipgeo`, `/v3/security`, `/v3/asn`, `/v3/abuse` separately +- Total: 4 HTTP calls per enrichment (5 credits typical) +- Useful when only some APIs are enabled + +## Data Flow + +1. OpenCTI triggers enrichment (auto or manual) +2. Connector reads observable via GraphQL +3. TLP check (respects `max_tlp`) +4. API client calls IPGeolocation.io +5. Response parsed into typed `IPIntelligence` +6. Risk scorer produces `RiskAssessment` +7. STIX mapper creates objects (Location, ASN, Identity, etc.) +8. Markdown generator creates analyst note +9. Bundle assembled and sent to OpenCTI via `send_stix2_bundle` diff --git a/internal-enrichment/ipgeolocation/docs/RISK_SCORING.md b/internal-enrichment/ipgeolocation/docs/RISK_SCORING.md new file mode 100644 index 00000000000..e5dd4301d90 --- /dev/null +++ b/internal-enrichment/ipgeolocation/docs/RISK_SCORING.md @@ -0,0 +1,82 @@ +# Risk Scoring Algorithm + +## Overview + +The connector normalizes heterogeneous risk signals from IPGeolocation.io into +a single unified score (0–100) and human-readable risk level. + +## Algorithm + +### Step 1: Base Score + +Start with the `threat_score` from the IPGeolocation.io Security API (0–100). + +### Step 2: Additive Modifiers + +Apply bonuses for each active boolean threat flag: + +| Flag | Weight | Rationale | +|------------------------|--------|----------------------------------------------| +| `is_tor` | +15 | Strong anonymization, common in attacks | +| `is_known_attacker` | +15 | Direct threat intelligence signal | +| `is_spam` | +10 | Active abuse source | +| `is_bot` | +10 | Automated malicious activity | +| `is_residential_proxy` | +8 | Hard-to-detect anonymization | +| `is_vpn` | +5 | Anonymization (commercial VPNs common) | +| `is_proxy` | +5 | Anonymization (datacenter proxies) | +| `is_relay` | +3 | Mild anonymization (Apple Relay, etc.) | +| `is_anonymous` | +3 | Only if not already captured by above flags | +| `is_cloud_provider` | +2 | Infrastructure signal, not inherently bad | + +### Step 3: Cap at 100 + +`unified_score = min(base + sum(bonuses), 100)` + +### Step 4: Risk Level + +| Score Range | Risk Level | +|-------------|------------| +| 0–20 | Low | +| 21–50 | Medium | +| 51–80 | High | +| 81–100 | Critical | + +### Step 5: Confidence + +Derived from API-provided confidence scores: + +- Collect `vpn_confidence_score`, `proxy_confidence_score`, and + `min(threat_score + 20, 100)` +- Average available scores +- Default to 70 if no confidence data is available + +## Examples + +### Clean DNS Resolver (8.8.8.8) +- Base threat_score: 0 +- Flags: `is_cloud_provider=true` (+2) +- Unified: 2 → **Low Risk** + +### VPN + Known Attacker (2.56.188.34) +- Base threat_score: 80 +- Flags: `is_vpn` (+5), `is_proxy` (+5), `is_residential_proxy` (+8), + `is_known_attacker` (+15), `is_cloud_provider` (+2) +- Raw: 80 + 35 = 115, capped to 100 → **Critical Risk** + +### Simple VPN User +- Base threat_score: 20 +- Flags: `is_vpn` (+5) +- Unified: 25 → **Medium Risk** + +## Design Decisions + +1. **Additive model**: Bonuses compound because multiple threat signals + simultaneously present a higher risk than any single signal. +2. **TOR/attacker weighted highest**: These are the strongest indicators of + malicious intent. +3. **Cloud provider weighted low**: Many legitimate services run on cloud + infrastructure. +4. **Deduplication of anonymous**: `is_anonymous` is only counted when it is + not already implied by VPN, proxy, TOR, or relay flags. +5. **Explanation always generated**: Analysts see *why* the score is what it + is, not just a number. diff --git a/internal-enrichment/ipgeolocation/docs/STIX_MAPPING.md b/internal-enrichment/ipgeolocation/docs/STIX_MAPPING.md new file mode 100644 index 00000000000..5eda1cbe7f0 --- /dev/null +++ b/internal-enrichment/ipgeolocation/docs/STIX_MAPPING.md @@ -0,0 +1,72 @@ +# STIX 2.1 Object Mapping + +## Overview + +This document describes how IPGeolocation.io API response fields are mapped to +STIX 2.1 objects within the OpenCTI data model. + +## Entity Mapping + +| IPGeolocation.io Field | STIX Type | OpenCTI Type | Notes | +|---------------------------|--------------------------|----------------------|--------------------------------| +| `location.country_name` | `location` | Country | `x_opencti_location_type` | +| `location.city` | `location` | City | Includes lat/lon | +| `asn.as_number` | `autonomous-system` | Autonomous System | Custom SCO in OpenCTI | +| `asn.organization` | `identity` | Organization | `identity_class=organization` | +| `company.name` | `identity` | Organization | Merged with ASN org | +| `security.cloud_provider` | `identity` | Organization | Hosting/cloud provider | +| `abuse.organization` | `identity` | Organization | Abuse contact identity | +| `hostname` | `domain-name` | Domain-Name | SCO observable | +| Threat flags | `indicator` | Indicator | STIX pattern, scored | +| Enrichment summary | `note` | Note | Markdown content | +| Risk opinion | `opinion` | Opinion | Mapped from risk level | + +## Relationship Mapping + +| Source | Relationship | Target | Condition | +|-----------------|----------------|-----------------|--------------------------------| +| IP Observable | `located-at` | Country | Always (if country exists) | +| City | `located-at` | Country | Always (if city exists) | +| IP Observable | `belongs-to` | ASN | Always (if ASN exists) | +| ASN | `belongs-to` | Organization | Always (if org exists) | +| IP Observable | `related-to` | Cloud Provider | If `is_cloud_provider=true` | +| IP Observable | `related-to` | Abuse Contact | If abuse data exists | +| IP Observable | `resolves-to` | Hostname | If hostname exists | +| Indicator | `based-on` | IP Observable | If indicator created | + +## Label Mapping + +| API Signal | Label | Condition | +|---------------------------|----------------------|-------------------------| +| `is_vpn=true` | `vpn` | Security flag | +| `is_proxy=true` | `proxy` | Security flag | +| `is_residential_proxy` | `residential-proxy` | Security flag | +| `is_tor=true` | `tor` | Security flag | +| `is_relay=true` | `relay` | Security flag | +| `is_bot=true` | `bot` | Security flag | +| `is_spam=true` | `spam` | Security flag | +| `is_known_attacker=true` | `known-attacker` | Security flag | +| `is_anonymous=true` | `anonymous` | Security flag | +| `is_cloud_provider=true` | `cloud-provider` | Security flag | +| `company.type` | e.g. `hosting`, `isp`| Infrastructure type | +| `network.is_anycast` | `anycast` | Network flag | +| Risk assessment | `risk:low/med/high/critical` | Computed | + +## Score Mapping + +| Source | OpenCTI Field | Range | +|----------------------------|------------------------|--------| +| Unified risk score | `x_opencti_score` | 0-100 | +| Risk confidence | `confidence` | 0-100 | + +## Opinion Mapping + +| Risk Level | STIX Opinion | +|------------|---------------------| +| Low | `strongly-disagree` | +| Medium | `neutral` | +| High | `agree` | +| Critical | `strongly-agree` | + +Context: The opinion represents "This IP is malicious." Low risk = strongly +disagree with that statement; critical = strongly agree. diff --git a/internal-enrichment/ipgeolocation/docs/TESTING_GUIDE.md b/internal-enrichment/ipgeolocation/docs/TESTING_GUIDE.md new file mode 100644 index 00000000000..d93d5b81464 --- /dev/null +++ b/internal-enrichment/ipgeolocation/docs/TESTING_GUIDE.md @@ -0,0 +1,314 @@ +# Testing the IPGeolocation.io Connector on a Fresh OpenCTI Instance + +Complete guide: zero → enriched observable in the UI. + +--- + +## Prerequisites + +- **Machine**: 8 GB RAM minimum (16 GB recommended). Linux, macOS, or Windows with WSL2. +- **Docker**: Docker Engine + Docker Compose v2 installed. +- **IPGeolocation.io account**: Free tier gives 1,000 credits/day (enough for testing). Sign up at https://ipgeolocation.io — copy your API key from the dashboard. + +--- + +## Step 1 — Set the Elasticsearch kernel parameter (Linux only) + +```bash +# Required by Elasticsearch — without this, the container crashes +sudo sysctl -w vm.max_map_count=262144 + +# Make it persistent across reboots +echo "vm.max_map_count=262144" | sudo tee -a /etc/sysctl.conf +``` + +On macOS/Windows (Docker Desktop), this is handled automatically. + +--- + +## Step 2 — Clone the official OpenCTI Docker repo + +```bash +git clone https://github.com/OpenCTI-Platform/docker.git opencti +cd opencti +``` + +--- + +## Step 3 — Create your .env file + +```bash +cp .env.sample .env +``` + +Now edit `.env` and fill in the required values. At minimum you need: + +```bash +# Generate these with: uuidgen (run it once per field) +OPENCTI_ADMIN_EMAIL=admin@opencti.io +OPENCTI_ADMIN_PASSWORD=ChangeMeNow123! +OPENCTI_ADMIN_TOKEN= + +# For MinIO (S3 storage) +MINIO_ROOT_USER= +MINIO_ROOT_PASSWORD= + +# For RabbitMQ +RABBITMQ_DEFAULT_USER=guest +RABBITMQ_DEFAULT_PASS=guest + +# Elasticsearch memory (use 4G if you have 16GB RAM, 2G if 8GB) +ELASTIC_MEMORY_SIZE=4G + +# Health check key +OPENCTI_HEALTHCHECK_ACCESS_KEY= + +# Connector IDs (one uuidgen per connector) +CONNECTOR_HISTORY_ID= +CONNECTOR_EXPORT_FILE_STIX_ID= +CONNECTOR_EXPORT_FILE_CSV_ID= +CONNECTOR_EXPORT_FILE_TXT_ID= +CONNECTOR_IMPORT_FILE_STIX_ID= +CONNECTOR_IMPORT_DOCUMENT_ID= +``` + +Quick way to generate all the UUIDs at once: + +```bash +for i in $(seq 1 10); do uuidgen; done +``` + +--- + +## Step 4 — Start the OpenCTI core stack (without XTM One) + +The official repo now includes XTM One services you don't need. Start only what matters: + +```bash +SERVICES="opencti worker \ + connector-export-file-stix connector-export-file-csv connector-export-file-txt \ + connector-import-file-stix connector-import-document \ + connector-analysis connector-opencti connector-mitre" + +docker compose up -d $SERVICES +``` + +First run pulls ~5 GB of images. Wait 3–5 minutes for Elasticsearch to become healthy. + +Check status: + +```bash +docker compose ps # All services should show "healthy" or "running" +docker compose logs opencti --tail 50 # Look for "Platform started on port 8080" +``` + +--- + +## Step 5 — Log in and verify + +Open **http://localhost:8080** in your browser. + +- Email: whatever you set in `OPENCTI_ADMIN_EMAIL` +- Password: whatever you set in `OPENCTI_ADMIN_PASSWORD` + +You should see the empty OpenCTI dashboard. Give MITRE a few minutes to populate (the mitre connector auto-imports ATT&CK data). + +--- + +## Step 6 — Add the IPGeolocation.io connector + +**Option A: Add to docker-compose (recommended)** + +Create a file called `docker-compose.override.yml` in the same `opencti/` directory: + +```yaml +version: "3" +services: + connector-ipgeolocation: + build: /path/to/your/opencti-ipgeolocation + # Or if you don't want to build, mount the source: + # image: python:3.12-slim + # command: bash -c "pip install -r /connector/requirements.txt && python -m src.main" + # volumes: + # - /path/to/opencti-ipgeolocation:/connector + # working_dir: /connector + environment: + - OPENCTI_URL=http://opencti:8080 + - OPENCTI_TOKEN=${OPENCTI_ADMIN_TOKEN} + - CONNECTOR_ID= + - CONNECTOR_NAME=IPGeolocation.io + - CONNECTOR_SCOPE=IPv4-Addr,IPv6-Addr + - CONNECTOR_AUTO=false + - CONNECTOR_CONFIDENCE_LEVEL=80 + - CONNECTOR_LOG_LEVEL=info + - CONNECTOR_UPDATE_EXISTING_DATA=true + - IPGEOLOCATION_API_KEY= + - IPGEOLOCATION_USE_GEO_API=true + - IPGEOLOCATION_USE_SECURITY_API=true + - IPGEOLOCATION_USE_ASN_API=true + - IPGEOLOCATION_USE_ABUSE_API=true + - IPGEOLOCATION_SINGLE_CALL_MODE=true + - IPGEOLOCATION_CREATE_LABELS=true + - IPGEOLOCATION_CREATE_INDICATORS=true + - IPGEOLOCATION_CREATE_RELATIONSHIPS=true + - IPGEOLOCATION_CREATE_NOTES=true + - IPGEOLOCATION_CREATE_OPINIONS=false + - IPGEOLOCATION_CREATE_SUMMARY=true + - IPGEOLOCATION_MAX_TLP=TLP:AMBER + - IPGEOLOCATION_DEFAULT_MARKING=TLP:WHITE + restart: unless-stopped + depends_on: + - opencti +``` + +Then: + +```bash +docker compose up -d connector-ipgeolocation +``` + +**Option B: Run locally with Python (easier for development)** + +```bash +cd /path/to/opencti-ipgeolocation +pip install -r requirements.txt +cp config.yml.sample config.yml +``` + +Edit `config.yml`: + +```yaml +opencti: + url: "http://localhost:8080" + token: "" + +connector: + id: "" + type: "INTERNAL_ENRICHMENT" + name: "IPGeolocation.io" + scope: "IPv4-Addr,IPv6-Addr" + auto: false + confidence_level: 80 + log_level: "info" + +ipgeolocation: + api_key: "" + use_geo_api: true + use_security_api: true + use_asn_api: true + use_abuse_api: true + single_call_mode: true + create_labels: true + create_indicators: true + create_relationships: true + create_notes: true + create_summary: true +``` + +Run: + +```bash +python -m src.main +``` + +You should see: +``` +IPGeolocation.io connector starting (scope=IPv4-Addr,IPv6-Addr, single_call=True) +``` + +--- + +## Step 7 — Verify the connector is registered + +In OpenCTI UI: + +1. Go to **Data** → **Ingestion** → **Connectors** +2. You should see "IPGeolocation.io" listed as an INTERNAL_ENRICHMENT connector +3. Status should show a green indicator + +--- + +## Step 8 — Create a test observable and trigger enrichment + +**Method A: Via the UI** + +1. Go to **Observations** → **Observables** +2. Click the **+** button (bottom right) +3. Choose **IPv4 Address** +4. Enter: `2.56.188.34` (a known VPN/proxy IP with high threat score) +5. Click **Create** +6. Open the observable you just created +7. Click the **cloud icon** (☁️) at the top right → "Enrichment" +8. You should see "IPGeolocation.io" in the list → click it +9. Wait a few seconds for the enrichment to complete + +**Method B: Via the API (curl)** + +```bash +# Replace with your actual values +OPENCTI_URL="http://localhost:8080" +OPENCTI_TOKEN="your-admin-token" + +# Create an IPv4 observable +curl -X POST "$OPENCTI_URL/graphql" \ + -H "Authorization: Bearer $OPENCTI_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{ + "query": "mutation { stixCyberObservableAdd(type: \"IPv4-Addr\", IPv4Addr: { value: \"2.56.188.34\" }) { id standard_id entity_type observable_value } }" + }' +``` + +Then trigger enrichment from the UI as described above. + +--- + +## Step 9 — See the results + +After enrichment completes, on the observable page you should see: + +### Knowledge tab +- **Relationships**: located-at (United States), belongs-to (AS15169), related-to (Packethub S.A.) +- **Indicators**: A malicious IP indicator (if threat score ≥ 50) +- **Locations**: Country (United States) and City (Dallas, Texas) +- **Organizations**: Google LLC, Packethub S.A., Abuse Contact + +### Overview tab +- **Labels**: `vpn`, `proxy`, `residential-proxy`, `known-attacker`, `cloud-provider`, `hosting`, `risk:critical` +- **Score**: Updated to the unified risk score (e.g., 100/100 for this IP) +- **External References**: Link to ipgeolocation.io + +### Notes tab +- A rich markdown note with: Executive Summary, IP Summary table, Security Assessment narrative, Infrastructure Profile, Network Context, Abuse Contact info, Geo Intelligence, Timeline, Confidence explanation + +--- + +## Good test IPs to try + +| IP | Expected result | +|---|---| +| `2.56.188.34` | High risk: VPN + proxy + known attacker | +| `8.8.8.8` | Low risk: Google DNS, cloud provider | +| `1.1.1.1` | Low risk: Cloudflare DNS, anycast | +| `185.220.101.1` | TOR exit node (usually) | +| `103.224.182.250` | Various threat signals | + +--- + +## Troubleshooting + +**Connector not showing up in OpenCTI:** +- Check logs: `docker compose logs connector-ipgeolocation` or the terminal where you ran `python -m src.main` +- Verify the `OPENCTI_URL` is reachable from the connector (use `http://opencti:8080` in Docker, `http://localhost:8080` locally) +- Verify the `OPENCTI_TOKEN` matches your admin token + +**Enrichment fails silently:** +- Check connector logs for API errors +- Verify your IPGeolocation.io API key is valid (test with: `curl "https://api.ipgeolocation.io/v3/ipgeo?apiKey=YOUR_KEY&ip=8.8.8.8"`) +- Free plan does NOT include security/abuse modules — you need a paid plan for full enrichment + +**"Observable not found" error:** +- The observable needs to exist in OpenCTI before enrichment is triggered +- Make sure the connector scope matches: `IPv4-Addr,IPv6-Addr` + +**Rate limiting:** +- IPGeolocation.io free plan: 1,000 credits/day \ No newline at end of file diff --git a/internal-enrichment/ipgeolocation/entrypoint.sh b/internal-enrichment/ipgeolocation/entrypoint.sh new file mode 100644 index 00000000000..1de972ff038 --- /dev/null +++ b/internal-enrichment/ipgeolocation/entrypoint.sh @@ -0,0 +1,5 @@ +#!/bin/bash +set -e + +cd /opt/opencti-connector +python -m src.main diff --git a/internal-enrichment/ipgeolocation/requirements.txt b/internal-enrichment/ipgeolocation/requirements.txt new file mode 100644 index 00000000000..a235adbc952 --- /dev/null +++ b/internal-enrichment/ipgeolocation/requirements.txt @@ -0,0 +1,5 @@ +pycti>=6.4.0,<7.0.0 +stix2>=3.0.0,<4.0.0 +requests>=2.31.0,<3.0.0 +pyyaml>=6.0 +urllib3>=2.0.0,<3.0.0 diff --git a/internal-enrichment/ipgeolocation/src/__init__.py b/internal-enrichment/ipgeolocation/src/__init__.py new file mode 100644 index 00000000000..190b1e8eaa4 --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/__init__.py @@ -0,0 +1,3 @@ +"""IPGeolocation.io OpenCTI Internal Enrichment Connector.""" + +__version__ = "1.0.0" diff --git a/internal-enrichment/ipgeolocation/src/api_client.py b/internal-enrichment/ipgeolocation/src/api_client.py new file mode 100644 index 00000000000..f59c1d29716 --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/api_client.py @@ -0,0 +1,229 @@ +""" +IPGeolocation.io OpenCTI Connector — API Client +================================================= + +HTTP client for the IPGeolocation.io v3 REST API with: +* Automatic retry + exponential back-off +* Credit-aware single-call vs dedicated-endpoint modes +* Structured logging via the connector helper +""" + +from __future__ import annotations + +from typing import Any, Optional + +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry + +from .models import IPIntelligence + + +class IPGeolocationAPIError(Exception): + """Raised on non-recoverable API errors.""" + + def __init__(self, status_code: int, message: str): + self.status_code = status_code + self.message = message + super().__init__(f"HTTP {status_code}: {message}") + + +class IPGeolocationClient: + """Thread-safe client for IPGeolocation.io v3 endpoints.""" + + # Endpoint paths + _IPGEO = "/v3/ipgeo" + _SECURITY = "/v3/security" + _ASN = "/v3/asn" + _ABUSE = "/v3/abuse" + + def __init__( + self, + api_key: str, + base_url: str = "https://api.ipgeolocation.io", + timeout: int = 30, + max_retries: int = 3, + retry_delay: int = 2, + logger=None, + ): + self._api_key = api_key + self._base_url = base_url.rstrip("/") + self._timeout = timeout + self._max_retries = max_retries + self._retry_delay = retry_delay + self._log = logger + + # Build a resilient session + self._session = requests.Session() + retry_strategy = Retry( + total=max_retries, + backoff_factor=retry_delay, + status_forcelist=[429, 500, 502, 503, 504], + allowed_methods=["GET", "POST"], + ) + adapter = HTTPAdapter(max_retries=retry_strategy) + self._session.mount("https://", adapter) + self._session.mount("http://", adapter) + + # --------------------------------------------------------------------- # + # Low-level + # --------------------------------------------------------------------- # + + def _get(self, path: str, params: Optional[dict] = None) -> dict: + """Execute an authenticated GET and return parsed JSON.""" + url = f"{self._base_url}{path}" + params = params or {} + params["apiKey"] = self._api_key + + if self._log: + self._log(f"GET {path} params={_redact(params)}") + + resp = self._session.get(url, params=params, timeout=self._timeout) + + # Log credits charged (header from IPGeolocation) + credits = resp.headers.get("X-Credits-Charged", "?") + if self._log: + self._log(f" → {resp.status_code} | credits={credits}") + + if resp.status_code == 200: + return resp.json() + + # Handle known error codes + body = "" + try: + body = resp.json().get("message", resp.text[:300]) + except Exception: + body = resp.text[:300] + + raise IPGeolocationAPIError(resp.status_code, body) + + # --------------------------------------------------------------------- # + # Public: single-call mode + # --------------------------------------------------------------------- # + + def lookup_unified( + self, + ip: str, + include_security: bool = True, + include_abuse: bool = True, + ) -> IPIntelligence: + """Single /v3/ipgeo call with include= for credit efficiency. + + Credits: + base (geo+asn+company+tz+network) = 1 + +security = +2 (total 3) + +abuse = +1 (total 4 with security) + """ + includes: list[str] = [] + if include_security: + includes.append("security") + if include_abuse: + includes.append("abuse") + + params: dict[str, Any] = {"ip": ip} + if includes: + params["include"] = ",".join(includes) + + data = self._get(self._IPGEO, params) + return IPIntelligence.from_ipgeo_response(data) + + # --------------------------------------------------------------------- # + # Public: dedicated-endpoint mode + # --------------------------------------------------------------------- # + + def lookup_geo(self, ip: str) -> dict: + """Geolocation only — 1 credit.""" + return self._get(self._IPGEO, {"ip": ip}) + + def lookup_security(self, ip: str) -> dict: + """Dedicated security endpoint — 2 credits.""" + return self._get(self._SECURITY, {"ip": ip}) + + def lookup_asn(self, ip: str) -> dict: + """ASN details (routes, peers, upstreams) — 1 credit.""" + return self._get(self._ASN, {"ip": ip}) + + def lookup_abuse(self, ip: str) -> dict: + """Abuse contact — 1 credit.""" + return self._get(self._ABUSE, {"ip": ip}) + + # --------------------------------------------------------------------- # + # Public: composite lookup respecting config + # --------------------------------------------------------------------- # + + def enrich( + self, + ip: str, + *, + single_call: bool = True, + use_geo: bool = True, + use_security: bool = True, + use_asn: bool = True, + use_abuse: bool = True, + ) -> IPIntelligence: + """High-level entry-point: returns full IPIntelligence. + + When *single_call* is True the connector uses ``/v3/ipgeo`` with + ``include=security,abuse`` to consolidate HTTP requests. When + False it calls each dedicated endpoint separately, which can be + useful when only some modules are enabled or when the admin + prefers per-endpoint freshness. + """ + if single_call and use_geo: + intel = self.lookup_unified( + ip, + include_security=use_security, + include_abuse=use_abuse, + ) + # Dedicated ASN is richer (peers, upstreams, routes) than ipgeo. + if use_asn: + try: + intel.merge_asn(self.lookup_asn(ip)) + except IPGeolocationAPIError: + pass # graceful: keep basic ASN from ipgeo + return intel + + # Dedicated mode + intel = IPIntelligence(ip=ip) + if use_geo: + try: + geo = self.lookup_geo(ip) + intel = IPIntelligence.from_ipgeo_response(geo) + except IPGeolocationAPIError as exc: + if self._log: + self._log(f"Geo lookup failed: {exc}") + if use_security: + try: + intel.merge_security(self.lookup_security(ip)) + except IPGeolocationAPIError as exc: + if self._log: + self._log(f"Security lookup failed: {exc}") + if use_asn: + try: + intel.merge_asn(self.lookup_asn(ip)) + except IPGeolocationAPIError as exc: + if self._log: + self._log(f"ASN lookup failed: {exc}") + if use_abuse: + try: + intel.merge_abuse(self.lookup_abuse(ip)) + except IPGeolocationAPIError as exc: + if self._log: + self._log(f"Abuse lookup failed: {exc}") + return intel + + def close(self) -> None: + self._session.close() + + +# -------------------------------------------------------------------------- +# Helpers +# -------------------------------------------------------------------------- + + +def _redact(params: dict) -> dict: + """Hide the API key in log output.""" + out = dict(params) + if "apiKey" in out: + out["apiKey"] = out["apiKey"][:4] + "****" + return out diff --git a/internal-enrichment/ipgeolocation/src/config.py b/internal-enrichment/ipgeolocation/src/config.py new file mode 100644 index 00000000000..52226fe1a6e --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/config.py @@ -0,0 +1,293 @@ +""" +IPGeolocation.io OpenCTI Connector — Configuration +==================================================== + +Typed configuration with validation. All parameters are sourced from +environment variables (Docker/K8s style) with YAML fallback via pycti's +``get_config_variable``. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + +import yaml +from pycti import get_config_variable + + +def _load_yaml_config() -> dict: + """Try to load config.yml next to the entrypoint.""" + for path in ("config.yml", "/opt/opencti-connector/config.yml"): + if os.path.isfile(path): + with open(path, "r") as fh: + return yaml.safe_load(fh) or {} + return {} + + +_CFG = _load_yaml_config() + + +def _var( + env_key: str, + yaml_path: list[str], + *, + required: bool = False, + default=None, + is_number: bool = False, +): + """Thin wrapper around pycti get_config_variable.""" + val = get_config_variable( + env_key, yaml_path, _CFG, required=required, default=default, isNumber=is_number + ) + return val + + +# --------------------------------------------------------------------------- +# Data-classes +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class OpenCTIConfig: + url: str = field( + default_factory=lambda: _var("OPENCTI_URL", ["opencti", "url"], required=True) + ) + token: str = field( + default_factory=lambda: _var( + "OPENCTI_TOKEN", ["opencti", "token"], required=True + ) + ) + + +@dataclass(frozen=True) +class ConnectorConfig: + id: str = field( + default_factory=lambda: _var("CONNECTOR_ID", ["connector", "id"], required=True) + ) + name: str = field( + default_factory=lambda: _var( + "CONNECTOR_NAME", ["connector", "name"], default="IPGeolocation.io" + ) + ) + scope: str = field( + default_factory=lambda: _var( + "CONNECTOR_SCOPE", ["connector", "scope"], default="IPv4-Addr,IPv6-Addr" + ) + ) + type: str = "INTERNAL_ENRICHMENT" + auto: bool = field( + default_factory=lambda: _var( + "CONNECTOR_AUTO", ["connector", "auto"], default=True + ) + ) + confidence_level: int = field( + default_factory=lambda: _var( + "CONNECTOR_CONFIDENCE_LEVEL", + ["connector", "confidence_level"], + default=80, + is_number=True, + ) + ) + log_level: str = field( + default_factory=lambda: _var( + "CONNECTOR_LOG_LEVEL", ["connector", "log_level"], default="info" + ) + ) + update_existing_data: bool = field( + default_factory=lambda: _var( + "CONNECTOR_UPDATE_EXISTING_DATA", + ["connector", "update_existing_data"], + default=True, + ) + ) + + +@dataclass(frozen=True) +class APIConfig: + """IPGeolocation.io API settings.""" + + api_key: str = field( + default_factory=lambda: _var( + "IPGEOLOCATION_API_KEY", ["ipgeolocation", "api_key"], required=True + ) + ) + base_url: str = field( + default_factory=lambda: _var( + "IPGEOLOCATION_BASE_URL", + ["ipgeolocation", "base_url"], + default="https://api.ipgeolocation.io", + ) + ) + timeout: int = field( + default_factory=lambda: _var( + "IPGEOLOCATION_TIMEOUT", + ["ipgeolocation", "timeout"], + default=30, + is_number=True, + ) + ) + max_retries: int = field( + default_factory=lambda: _var( + "IPGEOLOCATION_MAX_RETRIES", + ["ipgeolocation", "max_retries"], + default=3, + is_number=True, + ) + ) + retry_delay: int = field( + default_factory=lambda: _var( + "IPGEOLOCATION_RETRY_DELAY", + ["ipgeolocation", "retry_delay"], + default=2, + is_number=True, + ) + ) + + +@dataclass(frozen=True) +class EnrichmentConfig: + """Which APIs and features to enable.""" + + # API toggles + use_geo_api: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_USE_GEO_API", ["ipgeolocation", "use_geo_api"], default=True + ) + ) + use_security_api: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_USE_SECURITY_API", + ["ipgeolocation", "use_security_api"], + default=True, + ) + ) + use_asn_api: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_USE_ASN_API", ["ipgeolocation", "use_asn_api"], default=True + ) + ) + use_abuse_api: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_USE_ABUSE_API", + ["ipgeolocation", "use_abuse_api"], + default=True, + ) + ) + + # Credit optimization: single-call vs dedicated + use_single_call_mode: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_SINGLE_CALL_MODE", + ["ipgeolocation", "single_call_mode"], + default=True, + ) + ) + + # Feature toggles + create_labels: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_CREATE_LABELS", + ["ipgeolocation", "create_labels"], + default=True, + ) + ) + create_indicators: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_CREATE_INDICATORS", + ["ipgeolocation", "create_indicators"], + default=True, + ) + ) + create_relationships: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_CREATE_RELATIONSHIPS", + ["ipgeolocation", "create_relationships"], + default=True, + ) + ) + create_notes: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_CREATE_NOTES", + ["ipgeolocation", "create_notes"], + default=True, + ) + ) + create_opinions: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_CREATE_OPINIONS", + ["ipgeolocation", "create_opinions"], + default=False, + ) + ) + create_summary: bool = field( + default_factory=lambda: _var( + "IPGEOLOCATION_CREATE_SUMMARY", + ["ipgeolocation", "create_summary"], + default=True, + ) + ) + + # Thresholds + min_threat_score: int = field( + default_factory=lambda: _var( + "IPGEOLOCATION_MIN_THREAT_SCORE", + ["ipgeolocation", "min_threat_score"], + default=0, + is_number=True, + ) + ) + indicator_threat_threshold: int = field( + default_factory=lambda: _var( + "IPGEOLOCATION_INDICATOR_THREAT_THRESHOLD", + ["ipgeolocation", "indicator_threat_threshold"], + default=50, + is_number=True, + ) + ) + + # TLP + max_tlp: str = field( + default_factory=lambda: _var( + "IPGEOLOCATION_MAX_TLP", ["ipgeolocation", "max_tlp"], default="TLP:AMBER" + ) + ) + default_marking: str = field( + default_factory=lambda: _var( + "IPGEOLOCATION_DEFAULT_MARKING", + ["ipgeolocation", "default_marking"], + default="TLP:WHITE", + ) + ) + + +# --------------------------------------------------------------------------- +# Aggregate +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class IPGeolocationConnectorConfig: + opencti: OpenCTIConfig = field(default_factory=OpenCTIConfig) + connector: ConnectorConfig = field(default_factory=ConnectorConfig) + api: APIConfig = field(default_factory=APIConfig) + enrichment: EnrichmentConfig = field(default_factory=EnrichmentConfig) + + def to_helper_config(self) -> dict: + """Return dict consumable by ``OpenCTIConnectorHelper``.""" + return { + "opencti": { + "url": self.opencti.url, + "token": self.opencti.token, + }, + "connector": { + "id": self.connector.id, + "type": self.connector.type, + "name": self.connector.name, + "scope": self.connector.scope, + "auto": self.connector.auto, + "confidence_level": self.connector.confidence_level, + "log_level": self.connector.log_level, + "update_existing_data": self.connector.update_existing_data, + }, + } diff --git a/internal-enrichment/ipgeolocation/src/connector.py b/internal-enrichment/ipgeolocation/src/connector.py new file mode 100644 index 00000000000..aa211acb2d9 --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/connector.py @@ -0,0 +1,201 @@ +""" +IPGeolocation.io OpenCTI Connector — Enrichment Connector +============================================================ + +This is the main connector class that: +1. Listens for enrichment requests from OpenCTI +2. Reads the observable (IPv4/IPv6) +3. Calls the IPGeolocation.io API +4. Runs risk scoring +5. Produces STIX 2.1 objects +6. Sends the enrichment bundle back to OpenCTI +""" + +from __future__ import annotations + +import traceback + +from pycti import OpenCTIConnectorHelper + +from .api_client import IPGeolocationAPIError, IPGeolocationClient +from .config import IPGeolocationConnectorConfig +from .risk_scorer import RiskScorer +from .stix_mapper import STIXMapper + + +class IPGeolocationConnector: + """OpenCTI internal enrichment connector for IPGeolocation.io.""" + + # Observable types we can enrich (scope) + SUPPORTED_TYPES = {"IPv4-Addr", "IPv6-Addr"} + # Future: "Domain-Name", "Hostname" + + def __init__(self): + self._config = IPGeolocationConnectorConfig() + self._helper = OpenCTIConnectorHelper(self._config.to_helper_config()) + + # Sub-components + self._client = IPGeolocationClient( + api_key=self._config.api.api_key, + base_url=self._config.api.base_url, + timeout=self._config.api.timeout, + max_retries=self._config.api.max_retries, + retry_delay=self._config.api.retry_delay, + logger=self._helper.log_debug, + ) + self._scorer = RiskScorer() + self._mapper = STIXMapper( + author_name="IPGeolocation.io", + default_marking=self._config.enrichment.default_marking, + confidence=self._config.connector.confidence_level, + ) + + # ------------------------------------------------------------------ # + # Entry point + # ------------------------------------------------------------------ # + + def start(self) -> None: + """Start listening for enrichment jobs.""" + self._helper.log_info( + "IPGeolocation.io connector starting " + f"(scope={self._config.connector.scope}, " + f"single_call={self._config.enrichment.use_single_call_mode})" + ) + self._helper.listen(self._process_message) + + # ------------------------------------------------------------------ # + # Message handler + # ------------------------------------------------------------------ # + + def _process_message(self, data: dict) -> str: + """Called by the helper when an enrichment job arrives.""" + entity_id = data.get("entity_id") + if not entity_id: + return "No entity_id in message" + + self._helper.log_info(f"Enrichment request for entity {entity_id}") + + try: + return self._enrich(entity_id) + except IPGeolocationAPIError as exc: + msg = f"IPGeolocation API error: {exc}" + self._helper.log_error(msg) + return msg + except Exception as exc: + msg = f"Unexpected error: {exc}\n{traceback.format_exc()}" + self._helper.log_error(msg) + return msg + + # ------------------------------------------------------------------ # + # Core enrichment logic + # ------------------------------------------------------------------ # + + def _enrich(self, entity_id: str) -> str: + """Perform end-to-end enrichment of a single observable.""" + + # 1. Read the observable from OpenCTI + observable = self._helper.api.stix_cyber_observable.read(id=entity_id) + if observable is None: + return f"Observable {entity_id} not found" + + entity_type = observable.get("entity_type", "") + if entity_type not in self.SUPPORTED_TYPES: + return ( + f"Unsupported entity type: {entity_type}. " + f"Supported: {self.SUPPORTED_TYPES}" + ) + + ip_value = observable.get("value", "") + if not ip_value: + return "Observable has no value" + + stix_id = observable.get("standard_id", entity_id) + + self._helper.log_info(f"Enriching {entity_type} {ip_value} (stix_id={stix_id})") + + # 2. Check TLP + if not self._check_tlp(observable): + return "Observable TLP exceeds max_tlp — skipping" + + # 3. Call IPGeolocation.io API + enrich_cfg = self._config.enrichment + intel = self._client.enrich( + ip=ip_value, + single_call=enrich_cfg.use_single_call_mode, + use_geo=enrich_cfg.use_geo_api, + use_security=enrich_cfg.use_security_api, + use_asn=enrich_cfg.use_asn_api, + use_abuse=enrich_cfg.use_abuse_api, + ) + + # 4. Skip if below minimum threat score + if ( + enrich_cfg.min_threat_score > 0 + and intel.security.threat_score < enrich_cfg.min_threat_score + ): + self._helper.log_info( + f"Threat score {intel.security.threat_score} below minimum " + f"{enrich_cfg.min_threat_score} — skipping enrichment" + ) + return "Below minimum threat score" + + # 5. Score risk + risk = self._scorer.assess(intel) + self._helper.log_info( + f"Risk assessment: {risk.risk_level} ({risk.unified_score}/100)" + ) + + # 6. Build STIX objects + stix_objects = self._mapper.build_bundle_objects( + intel=intel, + risk=risk, + observable_id=stix_id, + observable_type=entity_type, + create_labels=enrich_cfg.create_labels, + create_indicators=enrich_cfg.create_indicators, + create_relationships=enrich_cfg.create_relationships, + create_notes=enrich_cfg.create_notes, + create_opinions=enrich_cfg.create_opinions, + create_summary=enrich_cfg.create_summary, + indicator_threshold=enrich_cfg.indicator_threat_threshold, + ) + + self._helper.log_info( + f"Produced {len(stix_objects)} STIX objects for {ip_value}" + ) + + # 7. Create + send bundle + bundle = self._helper.stix2_create_bundle(stix_objects) + bundles_sent = self._helper.send_stix2_bundle( + bundle, + update=self._config.connector.update_existing_data, + cleanup_inconsistent_bundle=True, + ) + + return ( + f"Enrichment complete for {ip_value}: " + f"{len(stix_objects)} objects, " + f"{len(bundles_sent)} bundle(s) sent. " + f"Risk: {risk.risk_level} ({risk.unified_score}/100)" + ) + + # ------------------------------------------------------------------ # + # TLP check + # ------------------------------------------------------------------ # + + def _check_tlp(self, observable: dict) -> bool: + """Respect the max TLP setting.""" + tlp_order = ["TLP:WHITE", "TLP:CLEAR", "TLP:GREEN", "TLP:AMBER", "TLP:RED"] + max_tlp = self._config.enrichment.max_tlp + max_idx = tlp_order.index(max_tlp) if max_tlp in tlp_order else 3 + + markings = observable.get("objectMarking", []) or [] + for m in markings: + definition = m.get("definition", "") + if definition in tlp_order: + if tlp_order.index(definition) > max_idx: + self._helper.log_info( + f"Observable TLP {definition} exceeds max {max_tlp}" + ) + return False + return True diff --git a/internal-enrichment/ipgeolocation/src/main.py b/internal-enrichment/ipgeolocation/src/main.py new file mode 100644 index 00000000000..56f47ba1e7a --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/main.py @@ -0,0 +1,34 @@ +""" +IPGeolocation.io OpenCTI Connector — Entrypoint +================================================= + +Usage:: + + python -m src.main + +Or as a Docker container with the appropriate environment variables. +""" + +import sys +import time +import traceback + +from .connector import IPGeolocationConnector + + +def main() -> None: + try: + connector = IPGeolocationConnector() + connector.start() + except Exception as exc: + print( + f"[FATAL] IPGeolocation.io connector failed to start: {exc}", + file=sys.stderr, + ) + traceback.print_exc() + time.sleep(10) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/internal-enrichment/ipgeolocation/src/markdown_generator.py b/internal-enrichment/ipgeolocation/src/markdown_generator.py new file mode 100644 index 00000000000..e5447909f23 --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/markdown_generator.py @@ -0,0 +1,427 @@ +""" +IPGeolocation.io OpenCTI Connector — Markdown Generator +========================================================= + +Produces rich, analyst-readable markdown notes that appear in the +OpenCTI UI when an observable is enriched. Includes: + +1. Executive Summary (one-paragraph narrative) +2. IP Summary table +3. Security Narrative (not just booleans) +4. Infrastructure Profile +5. Network Context +6. Abuse Workflow +7. Geo Intelligence +8. Timeline (if timestamps exist) +""" + +from __future__ import annotations + +from .models import IPIntelligence +from .risk_scorer import RiskAssessment + + +class MarkdownGenerator: + """Stateless generator: call ``generate`` with intel + risk.""" + + def generate(self, intel: IPIntelligence, risk: RiskAssessment) -> str: + sections: list[str] = [] + sections.append(self._executive_summary(intel, risk)) + sections.append(self._summary_table(intel, risk)) + sections.append(self._security_narrative(intel, risk)) + sections.append(self._infrastructure_profile(intel)) + sections.append(self._network_context(intel)) + sections.append(self._geo_intelligence(intel)) + sections.append(self._abuse_workflow(intel)) + sections.append(self._timeline(intel)) + sections.append(self._confidence_note(risk)) + sections.append(self._footer()) + return "\n\n".join(s for s in sections if s) + + # ------------------------------------------------------------------ # + # 1. Executive Summary + # ------------------------------------------------------------------ # + + def _executive_summary(self, intel: IPIntelligence, risk: RiskAssessment) -> str: + loc = intel.location + sec = intel.security + parts: list[str] = ["## Executive Summary", ""] + + geo = ( + f"{loc.city}, {loc.state_prov}, {loc.country_name}" + if loc.city + else (loc.country_name or "Unknown location") + ) + + org = intel.asn.organization or intel.company.name or "an unknown organization" + asn = intel.asn.as_number or "N/A" + + narrative = ( + f"`{intel.ip}` is geolocated in **{geo}** and belongs to " + f"**{org}** ({asn})." + ) + + # Anonymization details + anon: list[str] = [] + if sec.is_vpn: + providers = ( + ", ".join(sec.vpn_provider_names) + if sec.vpn_provider_names + else "unknown provider" + ) + anon.append(f"a VPN ({providers})") + if sec.is_proxy: + anon.append("a proxy server") + if sec.is_residential_proxy: + anon.append("a residential proxy") + if sec.is_tor: + anon.append("a TOR exit node") + if sec.is_relay: + anon.append("a relay node") + + if anon: + narrative += f" The address is identified as {', '.join(anon)}." + + threat: list[str] = [] + if sec.is_known_attacker: + threat.append("known attacker infrastructure") + if sec.is_spam: + threat.append("a spam source") + if sec.is_bot: + threat.append("bot activity") + if threat: + narrative += f" It has been flagged for {', '.join(threat)}." + + if sec.is_cloud_provider: + narrative += ( + f" The IP is hosted on cloud infrastructure " + f"({sec.cloud_provider_name or 'unknown'})." + ) + + narrative += ( + f" Overall risk assessment: **{risk.risk_level}** " + f"({risk.unified_score}/100)." + ) + + parts.append(narrative) + return "\n".join(parts) + + # ------------------------------------------------------------------ # + # 2. Summary Table + # ------------------------------------------------------------------ # + + def _summary_table(self, intel: IPIntelligence, risk: RiskAssessment) -> str: + loc = intel.location + sec = intel.security + rows = [ + ("IP Address", f"`{intel.ip}`"), + ( + "Country", + ( + f"{loc.country_emoji} {loc.country_name}" + if loc.country_name + else "N/A" + ), + ), + ("City", loc.city or "N/A"), + ("State/Province", loc.state_prov or "N/A"), + ("Continent", loc.continent_name or "N/A"), + ("ASN", intel.asn.as_number or "N/A"), + ("Organization", intel.asn.organization or intel.company.name or "N/A"), + ("Network Type", intel.asn.type or intel.company.type or "N/A"), + ("ISP", intel.company.name or "N/A"), + ("Hostname", f"`{intel.hostname}`" if intel.hostname else "N/A"), + ("Threat Score", f"{sec.threat_score}/100"), + ("Risk Level", f"**{risk.risk_level}** ({risk.unified_score}/100)"), + ("VPN", _yn(sec.is_vpn)), + ("Proxy", _yn(sec.is_proxy)), + ("Residential Proxy", _yn(sec.is_residential_proxy)), + ("TOR", _yn(sec.is_tor)), + ("Known Attacker", _yn(sec.is_known_attacker)), + ("Bot", _yn(sec.is_bot)), + ("Spam", _yn(sec.is_spam)), + ( + "Cloud Provider", + sec.cloud_provider_name if sec.is_cloud_provider else "No", + ), + ("Timezone", intel.timezone.name or "N/A"), + ] + + lines = ["## IP Summary", "", "| Field | Value |", "| :--- | :--- |"] + for label, value in rows: + lines.append(f"| {label} | {value} |") + return "\n".join(lines) + + # ------------------------------------------------------------------ # + # 3. Security Narrative + # ------------------------------------------------------------------ # + + def _security_narrative(self, intel: IPIntelligence, risk: RiskAssessment) -> str: + sec = intel.security + if not sec.threat_score and not any( + [ + sec.is_vpn, + sec.is_proxy, + sec.is_tor, + sec.is_known_attacker, + sec.is_spam, + sec.is_bot, + ] + ): + return "" + + lines = ["## Security Assessment", ""] + + # Narrative instead of booleans + if sec.is_anonymous: + parts = [] + if sec.is_vpn: + p = "a commercial VPN" + if sec.vpn_provider_names: + p += f" ({', '.join(sec.vpn_provider_names)})" + if sec.vpn_confidence_score: + p += f" [confidence: {sec.vpn_confidence_score}%]" + parts.append(p) + if sec.is_proxy: + p = "a proxy" + if sec.proxy_provider_names: + p += f" ({', '.join(sec.proxy_provider_names)})" + parts.append(p) + if sec.is_residential_proxy: + parts.append("residential proxy network") + if sec.is_tor: + parts.append("a known TOR exit node") + if sec.is_relay: + p = "a relay service" + if sec.relay_provider_name: + p += f" ({sec.relay_provider_name})" + parts.append(p) + + lines.append( + f"This address appears to originate from {', '.join(parts)}. " + f"Traffic from this IP may be masking its true origin." + ) + else: + lines.append("No anonymization signals detected for this IP.") + + # Threat signals + threats: list[str] = [] + if sec.is_known_attacker: + threats.append("known attacker infrastructure") + if sec.is_spam: + threats.append("a spam origination point") + if sec.is_bot: + threats.append("automated bot traffic") + if threats: + lines.append("") + lines.append( + f"**Threat flags:** This IP has been identified as " + f"{', '.join(threats)}." + ) + + lines.append("") + lines.append(risk.explanation) + return "\n".join(lines) + + # ------------------------------------------------------------------ # + # 4. Infrastructure Profile + # ------------------------------------------------------------------ # + + def _infrastructure_profile(self, intel: IPIntelligence) -> str: + comp = intel.company + net = intel.network + if not comp.name and not net.connection_type: + return "" + + lines = ["## Infrastructure Profile", ""] + + profile_type = comp.type or intel.asn.type or "Unknown" + profile_map = { + "hosting": "Hosting / Data Center", + "isp": "Internet Service Provider", + "business": "Business / Enterprise", + "education": "Education / Research", + } + human = profile_map.get(profile_type.lower(), profile_type.title()) + lines.append(f"**Category:** {human}") + + if comp.name: + lines.append(f"**Company:** {comp.name}") + if comp.domain: + lines.append(f"**Domain:** {comp.domain}") + if net.connection_type: + lines.append(f"**Connection Type:** {net.connection_type}") + if net.route: + lines.append(f"**Route:** `{net.route}`") + if net.is_anycast: + lines.append("**Anycast:** Yes — this IP is an anycast address") + + return "\n".join(lines) + + # ------------------------------------------------------------------ # + # 5. Network Context + # ------------------------------------------------------------------ # + + def _network_context(self, intel: IPIntelligence) -> str: + asn = intel.asn + if not asn.as_number: + return "" + + lines = ["## Network Context", ""] + lines.append(f"**ASN:** {asn.as_number}") + lines.append(f"**Organization:** {asn.organization or 'N/A'}") + + if asn.asn_name: + lines.append(f"**ASN Name:** {asn.asn_name}") + if asn.type: + lines.append(f"**Type:** {asn.type}") + if asn.rir: + lines.append(f"**RIR:** {asn.rir}") + if asn.date_allocated: + lines.append(f"**Allocated:** {asn.date_allocated}") + if asn.domain: + lines.append(f"**Domain:** {asn.domain}") + if asn.num_of_ipv4_routes: + lines.append(f"**IPv4 Routes:** {asn.num_of_ipv4_routes}") + if asn.num_of_ipv6_routes: + lines.append(f"**IPv6 Routes:** {asn.num_of_ipv6_routes}") + + if asn.routes: + routes_str = ", ".join(f"`{r}`" for r in asn.routes[:10]) + lines.append(f"**Routes:** {routes_str}") + + if asn.peers: + peer_strs = [f"{p.as_number} ({p.description})" for p in asn.peers[:10]] + lines.append(f"**Peers:** {', '.join(peer_strs)}") + if asn.upstreams: + up_strs = [f"{u.as_number} ({u.description})" for u in asn.upstreams[:10]] + lines.append(f"**Upstreams:** {', '.join(up_strs)}") + if asn.downstreams: + down_strs = [ + f"{d.as_number} ({d.description})" for d in asn.downstreams[:10] + ] + lines.append(f"**Downstreams:** {', '.join(down_strs)}") + + return "\n".join(lines) + + # ------------------------------------------------------------------ # + # 6. Geo Intelligence + # ------------------------------------------------------------------ # + + def _geo_intelligence(self, intel: IPIntelligence) -> str: + loc = intel.location + if not loc.country_name: + return "" + + lines = ["## Geo Intelligence", ""] + lines.append( + f"**Location:** {loc.city or 'N/A'}, " + f"{loc.state_prov or 'N/A'}, {loc.country_name}" + ) + if loc.latitude and loc.longitude: + lines.append(f"**Coordinates:** {loc.latitude}, {loc.longitude}") + if loc.accuracy_radius: + conf = f" ({loc.confidence} confidence)" if loc.confidence else "" + lines.append(f"**Accuracy Radius:** {loc.accuracy_radius} km{conf}") + if loc.district: + lines.append(f"**District:** {loc.district}") + if loc.zipcode: + lines.append(f"**Postal Code:** {loc.zipcode}") + if loc.is_eu: + lines.append("**EU Member:** Yes") + if loc.continent_name: + lines.append(f"**Continent:** {loc.continent_name}") + + return "\n".join(lines) + + # ------------------------------------------------------------------ # + # 7. Abuse Workflow + # ------------------------------------------------------------------ # + + def _abuse_workflow(self, intel: IPIntelligence) -> str: + ab = intel.abuse + if not ab.emails and not ab.name: + return "" + + lines = ["## Abuse Contact", ""] + lines.append( + "Use the following contact information to report abuse " + "originating from this IP range." + ) + lines.append("") + + if ab.organization: + lines.append(f"**Organization:** {ab.organization}") + if ab.name: + lines.append(f"**Contact Name:** {ab.name}") + if ab.kind: + lines.append(f"**Kind:** {ab.kind}") + if ab.emails: + for email in ab.emails: + lines.append(f"**Email:** `{email}`") + if ab.phone_numbers: + for phone in ab.phone_numbers: + lines.append(f"**Phone:** {phone}") + if ab.address: + lines.append(f"**Address:** {ab.address}") + if ab.route: + lines.append(f"**Network:** `{ab.route}`") + if ab.country: + lines.append(f"**Registered Country:** {ab.country}") + + return "\n".join(lines) + + # ------------------------------------------------------------------ # + # 8. Timeline + # ------------------------------------------------------------------ # + + def _timeline(self, intel: IPIntelligence) -> str: + sec = intel.security + events: list[tuple[str, str]] = [] + + if sec.vpn_last_seen: + events.append((sec.vpn_last_seen, "Last seen as VPN")) + if sec.proxy_last_seen: + events.append((sec.proxy_last_seen, "Last seen as Proxy")) + if intel.asn.date_allocated: + events.append((intel.asn.date_allocated, "ASN allocated")) + + if not events: + return "" + + events.sort() + lines = ["## Timeline", ""] + for date, label in events: + lines.append(f"- **{date}** — {label}") + return "\n".join(lines) + + # ------------------------------------------------------------------ # + # 9. Confidence Explanation + # ------------------------------------------------------------------ # + + def _confidence_note(self, risk: RiskAssessment) -> str: + return ( + f"## Enrichment Confidence\n\n" + f"Assessment confidence: **{risk.confidence}/100**. " + f"This value is derived from the API-provided confidence " + f"scores for VPN/proxy detection and the overall threat " + f"score. Higher values indicate stronger signal correlation " + f"across multiple detection methods." + ) + + # ------------------------------------------------------------------ # + # Footer + # ------------------------------------------------------------------ # + + @staticmethod + def _footer() -> str: + return "---\n*Enriched by IPGeolocation.io OpenCTI Connector*" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _yn(val: bool) -> str: + return "**Yes** ⚠️" if val else "No" diff --git a/internal-enrichment/ipgeolocation/src/models.py b/internal-enrichment/ipgeolocation/src/models.py new file mode 100644 index 00000000000..3974e3233ce --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/models.py @@ -0,0 +1,401 @@ +""" +IPGeolocation.io OpenCTI Connector — Typed Data Models +======================================================== + +Dataclasses that faithfully represent the IPGeolocation.io v3 API +response schemas. Using ``from_dict`` factory methods so missing / +null fields degrade gracefully. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Optional + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _g(d: dict, *keys, default=None): + """Safe nested dict get.""" + for k in keys: + if not isinstance(d, dict): + return default + d = d.get(k, default) + return d if d is not None else default + + +# --------------------------------------------------------------------------- +# Location +# --------------------------------------------------------------------------- + + +@dataclass +class LocationData: + continent_code: str = "" + continent_name: str = "" + country_code2: str = "" + country_code3: str = "" + country_name: str = "" + country_name_official: str = "" + country_capital: str = "" + state_prov: str = "" + state_code: str = "" + district: str = "" + city: str = "" + locality: str = "" + zipcode: str = "" + latitude: str = "" + longitude: str = "" + is_eu: bool = False + geoname_id: str = "" + accuracy_radius: str = "" + confidence: str = "" + country_flag: str = "" + country_emoji: str = "" + + @classmethod + def from_dict(cls, d: Optional[dict]) -> LocationData: + if not d: + return cls() + return cls( + **{k: d.get(k, getattr(cls, k, "")) for k in cls.__dataclass_fields__} + ) + + +# --------------------------------------------------------------------------- +# ASN Peer / Upstream / Downstream entry +# --------------------------------------------------------------------------- + + +@dataclass +class ASNRelEntry: + """A single peer, upstream, or downstream AS.""" + + as_number: str = "" + description: str = "" + country: str = "" + + @classmethod + def from_dict(cls, d: Optional[dict]) -> ASNRelEntry: + if not d or not isinstance(d, dict): + return cls() + return cls( + as_number=str(d.get("as_number", "")), + description=str(d.get("description", "")), + country=str(d.get("country", "")), + ) + + +# --------------------------------------------------------------------------- +# ASN (from /v3/ipgeo default or /v3/asn) +# +# Free /v3/ipgeo: as_number, organization, country +# Paid /v3/ipgeo: + type, domain, date_allocated, rir +# Paid /v3/asn: + asn_name, allocation_status, num_of_ipv4_routes, +# num_of_ipv6_routes, routes[], peers[], upstreams[], +# downstreams[], whois_response +# --------------------------------------------------------------------------- + + +@dataclass +class ASNData: + # Common (free + paid) + as_number: str = "" + organization: str = "" + country: str = "" + # Paid /v3/ipgeo extras + type: str = "" # ISP, hosting, business, education + domain: str = "" + date_allocated: str = "" + rir: str = "" # ARIN, RIPE, APNIC, etc. + # Dedicated /v3/asn extras + asn_name: str = "" + allocation_status: str = "" + num_of_ipv4_routes: str = "" + num_of_ipv6_routes: str = "" + # Routes: list of CIDR strings like ["1.0.0.0/24", "1.0.4.0/22"] + routes: list[str] = field(default_factory=list) + # Network relationships: arrays of {as_number, description, country} + peers: list[ASNRelEntry] = field(default_factory=list) + upstreams: list[ASNRelEntry] = field(default_factory=list) + downstreams: list[ASNRelEntry] = field(default_factory=list) + # WHOIS + whois_response: str = "" + + @classmethod + def from_dict(cls, d: Optional[dict]) -> ASNData: + if not d: + return cls() + obj = cls() + # Simple string fields + for k in ( + "as_number", + "organization", + "country", + "type", + "domain", + "date_allocated", + "rir", + "asn_name", + "allocation_status", + "num_of_ipv4_routes", + "num_of_ipv6_routes", + "whois_response", + ): + if k in d: + setattr(obj, k, str(d[k]) if d[k] is not None else "") + # Routes: list of CIDR strings + if "routes" in d and isinstance(d["routes"], list): + obj.routes = [str(r) for r in d["routes"] if r] + # Peers / upstreams / downstreams: list of objects + for rel_key in ("peers", "upstreams", "downstreams"): + if rel_key in d and isinstance(d[rel_key], list): + entries = [] + for item in d[rel_key]: + if isinstance(item, dict): + entries.append(ASNRelEntry.from_dict(item)) + elif isinstance(item, str): + # Fallback for unexpected plain strings + entries.append(ASNRelEntry(as_number=item)) + setattr(obj, rel_key, entries) + return obj + + +# --------------------------------------------------------------------------- +# Company +# --------------------------------------------------------------------------- + + +@dataclass +class CompanyData: + name: str = "" + type: str = "" # business, isp, hosting, education + domain: str = "" + + @classmethod + def from_dict(cls, d: Optional[dict]) -> CompanyData: + if not d: + return cls() + return cls( + name=d.get("name", ""), + type=d.get("type", ""), + domain=d.get("domain", ""), + ) + + +# --------------------------------------------------------------------------- +# Network +# --------------------------------------------------------------------------- + + +@dataclass +class NetworkData: + connection_type: str = "" + route: str = "" + is_anycast: bool = False + + @classmethod + def from_dict(cls, d: Optional[dict]) -> NetworkData: + if not d: + return cls() + return cls( + connection_type=d.get("connection_type", ""), + route=d.get("route", ""), + is_anycast=bool(d.get("is_anycast", False)), + ) + + +# --------------------------------------------------------------------------- +# Timezone +# --------------------------------------------------------------------------- + + +@dataclass +class TimezoneData: + name: str = "" + offset: float = 0.0 + offset_with_dst: float = 0.0 + current_time: str = "" + is_dst: bool = False + dst_savings: int = 0 + dst_exists: bool = False + dst_start: str = "" + dst_end: str = "" + + @classmethod + def from_dict(cls, d: Optional[dict]) -> TimezoneData: + if not d: + return cls() + return cls( + name=d.get("name", ""), + offset=float(d.get("offset", 0)), + offset_with_dst=float(d.get("offset_with_dst", 0)), + current_time=d.get("current_time", ""), + is_dst=bool(d.get("is_dst", False)), + dst_savings=int(d.get("dst_savings", 0)), + dst_exists=bool(d.get("dst_exists", False)), + dst_start=d.get("dst_start", ""), + dst_end=d.get("dst_end", ""), + ) + + +# --------------------------------------------------------------------------- +# Currency +# --------------------------------------------------------------------------- + + +@dataclass +class CurrencyData: + code: str = "" + name: str = "" + symbol: str = "" + + @classmethod + def from_dict(cls, d: Optional[dict]) -> CurrencyData: + if not d: + return cls() + return cls( + code=d.get("code", ""), + name=d.get("name", ""), + symbol=d.get("symbol", ""), + ) + + +# --------------------------------------------------------------------------- +# Security +# --------------------------------------------------------------------------- + + +@dataclass +class SecurityData: + threat_score: int = 0 + is_tor: bool = False + is_proxy: bool = False + proxy_provider_names: list[str] = field(default_factory=list) + proxy_confidence_score: int = 0 + proxy_last_seen: str = "" + is_residential_proxy: bool = False + is_vpn: bool = False + vpn_provider_names: list[str] = field(default_factory=list) + vpn_confidence_score: int = 0 + vpn_last_seen: str = "" + is_relay: bool = False + relay_provider_name: str = "" + is_anonymous: bool = False + is_known_attacker: bool = False + is_bot: bool = False + is_spam: bool = False + is_cloud_provider: bool = False + cloud_provider_name: str = "" + + @classmethod + def from_dict(cls, d: Optional[dict]) -> SecurityData: + if not d: + return cls() + obj = cls() + for k in cls.__dataclass_fields__: + if k in d: + val = d[k] + ann = cls.__dataclass_fields__[k].type + if "list" in str(ann): + setattr(obj, k, val if isinstance(val, list) else []) + elif "bool" in str(ann): + setattr(obj, k, bool(val)) + elif "int" in str(ann): + setattr(obj, k, int(val) if val else 0) + else: + setattr(obj, k, str(val) if val is not None else "") + return obj + + +# --------------------------------------------------------------------------- +# Abuse Contact +# --------------------------------------------------------------------------- + + +@dataclass +class AbuseData: + route: str = "" + country: str = "" + name: str = "" + organization: str = "" + kind: str = "" + address: str = "" + emails: list[str] = field(default_factory=list) + phone_numbers: list[str] = field(default_factory=list) + + @classmethod + def from_dict(cls, d: Optional[dict]) -> AbuseData: + if not d: + return cls() + return cls( + route=d.get("route", ""), + country=d.get("country", ""), + name=d.get("name", ""), + organization=d.get("organization", ""), + kind=d.get("kind", ""), + address=d.get("address", ""), + emails=d.get("emails", []) or [], + phone_numbers=d.get("phone_numbers", []) or [], + ) + + +# --------------------------------------------------------------------------- +# Aggregate IP Intelligence +# --------------------------------------------------------------------------- + + +@dataclass +class IPIntelligence: + """Full merged intelligence for a single IP address.""" + + ip: str = "" + hostname: str = "" + domain: str = "" + location: LocationData = field(default_factory=LocationData) + asn: ASNData = field(default_factory=ASNData) + company: CompanyData = field(default_factory=CompanyData) + network: NetworkData = field(default_factory=NetworkData) + timezone: TimezoneData = field(default_factory=TimezoneData) + currency: CurrencyData = field(default_factory=CurrencyData) + security: SecurityData = field(default_factory=SecurityData) + abuse: AbuseData = field(default_factory=AbuseData) + raw_responses: dict = field(default_factory=dict) + + @classmethod + def from_ipgeo_response(cls, data: dict) -> IPIntelligence: + """Build from /v3/ipgeo (possibly with include=security,abuse).""" + return cls( + ip=data.get("ip", ""), + hostname=data.get("hostname", ""), + domain=data.get("domain", ""), + location=LocationData.from_dict(data.get("location")), + asn=ASNData.from_dict(data.get("asn")), + company=CompanyData.from_dict(data.get("company")), + network=NetworkData.from_dict(data.get("network")), + timezone=TimezoneData.from_dict(data.get("time_zone")), + currency=CurrencyData.from_dict(data.get("currency")), + security=SecurityData.from_dict(data.get("security")), + abuse=AbuseData.from_dict(data.get("abuse")), + raw_responses={"ipgeo": data}, + ) + + def merge_security(self, data: dict) -> None: + """Overlay dedicated /v3/security response.""" + sec = data.get("security", data) + self.security = SecurityData.from_dict(sec) + self.raw_responses["security"] = data + + def merge_asn(self, data: dict) -> None: + """Overlay detailed /v3/asn response.""" + asn_block = data.get("asn", data) + self.asn = ASNData.from_dict(asn_block) + self.raw_responses["asn"] = data + + def merge_abuse(self, data: dict) -> None: + """Overlay /v3/abuse response.""" + abuse_block = data.get("abuse", data) + self.abuse = AbuseData.from_dict(abuse_block) + self.raw_responses["abuse"] = data diff --git a/internal-enrichment/ipgeolocation/src/risk_scorer.py b/internal-enrichment/ipgeolocation/src/risk_scorer.py new file mode 100644 index 00000000000..c5d7c63781c --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/risk_scorer.py @@ -0,0 +1,214 @@ +""" +IPGeolocation.io OpenCTI Connector — Risk Scorer +================================================== + +Normalises the heterogeneous risk signals from IPGeolocation.io into: + +* **unified_score** (0–100): single number for OpenCTI ``x_opencti_score`` +* **risk_level**: Low / Medium / High / Critical +* **explanation**: human-readable analyst paragraph + +Algorithm +--------- +1. Start with ``threat_score`` from the Security API (0-100). +2. Apply additive modifiers for binary risk flags: + - TOR exit: +15 + - Known attacker: +15 + - Spam source: +10 + - Bot: +10 + - VPN (non-residential): +5 + - Proxy (non-residential): +5 + - Residential proxy: +8 (harder to detect = higher risk) + - Relay: +3 + - Anonymous: +3 (only if not already captured above) + - Cloud provider: +2 (infrastructure, not inherently malicious) +3. Cap at 100. +4. Map to risk level: + - 0-20: Low + - 21-50: Medium + - 51-80: High + - 81+: Critical +5. Build an English explanation listing every contributing factor. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .models import IPIntelligence, SecurityData + +# --------------------------------------------------------------------------- +# Risk level enum (simple str) +# --------------------------------------------------------------------------- + +RISK_LOW = "Low" +RISK_MEDIUM = "Medium" +RISK_HIGH = "High" +RISK_CRITICAL = "Critical" + + +@dataclass +class RiskAssessment: + """Outcome of the risk scoring process.""" + + unified_score: int = 0 + risk_level: str = RISK_LOW + explanation: str = "" + contributing_factors: list[str] = field(default_factory=list) + opencti_score: int = 0 # identical to unified_score, for convenience + confidence: int = 0 # our confidence in the assessment (0-100) + + @property + def opinion_text(self) -> str: + mapping = { + RISK_LOW: "strongly-disagree", + RISK_MEDIUM: "neutral", + RISK_HIGH: "agree", + RISK_CRITICAL: "strongly-agree", + } + return mapping.get(self.risk_level, "neutral") + + +class RiskScorer: + """Stateless scorer: call ``assess`` with an ``IPIntelligence``.""" + + # Additive weights + _WEIGHTS = { + "is_tor": 15, + "is_known_attacker": 15, + "is_spam": 10, + "is_bot": 10, + "is_residential_proxy": 8, + "is_vpn": 5, + "is_proxy": 5, + "is_relay": 3, + "is_anonymous": 3, + "is_cloud_provider": 2, + } + + # Human labels for each flag + _LABELS = { + "is_tor": "TOR exit node", + "is_known_attacker": "known attacker infrastructure", + "is_spam": "spam source", + "is_bot": "bot activity detected", + "is_residential_proxy": "residential proxy", + "is_vpn": "VPN endpoint", + "is_proxy": "proxy server", + "is_relay": "relay node", + "is_anonymous": "anonymous traffic", + "is_cloud_provider": "cloud/hosting provider", + } + + def assess(self, intel: IPIntelligence) -> RiskAssessment: + sec = intel.security + base = sec.threat_score + factors: list[str] = [] + bonus = 0 + already_anon = False + + for flag, weight in self._WEIGHTS.items(): + val = getattr(sec, flag, False) + if not val: + continue + # Avoid double-counting anonymous if VPN/proxy/TOR already on + if flag == "is_anonymous": + if already_anon: + continue + if flag in ( + "is_vpn", + "is_proxy", + "is_tor", + "is_relay", + "is_residential_proxy", + ): + already_anon = True + bonus += weight + label = self._LABELS[flag] + # Add provider detail where available + if flag == "is_vpn" and sec.vpn_provider_names: + label += f" ({', '.join(sec.vpn_provider_names)})" + elif flag == "is_proxy" and sec.proxy_provider_names: + label += f" ({', '.join(sec.proxy_provider_names)})" + elif flag == "is_cloud_provider" and sec.cloud_provider_name: + label += f" ({sec.cloud_provider_name})" + factors.append(label) + + raw = base + bonus + score = min(raw, 100) + + level = self._level(score) + confidence = self._derive_confidence(sec) + explanation = self._build_explanation( + intel.ip, score, level, factors, sec, confidence + ) + + return RiskAssessment( + unified_score=score, + risk_level=level, + explanation=explanation, + contributing_factors=factors, + opencti_score=score, + confidence=confidence, + ) + + # ------------------------------------------------------------------ # + # Internal + # ------------------------------------------------------------------ # + + @staticmethod + def _level(score: int) -> str: + if score <= 20: + return RISK_LOW + if score <= 50: + return RISK_MEDIUM + if score <= 80: + return RISK_HIGH + return RISK_CRITICAL + + @staticmethod + def _derive_confidence(sec: SecurityData) -> int: + """Derive our confidence in the enrichment quality. + + Uses API-provided confidence scores where available, otherwise + defaults to 70 (reasonable for a commercial TI feed). + """ + scores: list[int] = [] + if sec.vpn_confidence_score: + scores.append(sec.vpn_confidence_score) + if sec.proxy_confidence_score: + scores.append(sec.proxy_confidence_score) + if sec.threat_score: + scores.append(min(sec.threat_score + 20, 100)) + if scores: + return min(int(sum(scores) / len(scores)), 100) + return 70 + + @staticmethod + def _build_explanation( + ip: str, + score: int, + level: str, + factors: list[str], + sec: SecurityData, + confidence: int, + ) -> str: + if not factors: + return ( + f"**{ip}** received a threat score of {sec.threat_score}/100 " + f"from IPGeolocation.io with no specific threat flags raised. " + f"Unified risk: **{level}** ({score}/100)." + ) + factor_str = ", ".join(factors) + parts = [ + f"**{ip}** is assessed as **{level} Risk** " + f"(unified score {score}/100) based on the following signals: " + f"{factor_str}.", + ] + if sec.threat_score: + parts.append( + f"The upstream threat score from IPGeolocation.io is " + f"{sec.threat_score}/100." + ) + parts.append(f"Assessment confidence: {confidence}/100.") + return " ".join(parts) diff --git a/internal-enrichment/ipgeolocation/src/stix_mapper.py b/internal-enrichment/ipgeolocation/src/stix_mapper.py new file mode 100644 index 00000000000..aafc0dfbfe0 --- /dev/null +++ b/internal-enrichment/ipgeolocation/src/stix_mapper.py @@ -0,0 +1,635 @@ +""" +IPGeolocation.io OpenCTI Connector — STIX 2.1 Mapper +====================================================== + +Transforms :class:`IPIntelligence` into a list of STIX 2.1 objects ready +for bundling and ingestion into OpenCTI. + +STIX mapping decisions: + Country/City → stix2.Location + ASN → stix2.AutonomousSystem (custom SCO) + Org/ISP → stix2.Identity (class=organization) + Hostname → stix2.DomainName (SCO) + Threat flags → stix2.Indicator + labels + Abuse contact → stix2.Identity (class=organization) + Notes → stix2.Note (markdown) + Opinions → stix2.Opinion +""" + +from __future__ import annotations + +import datetime as dt +import uuid +from typing import Any, Optional + +import stix2 + +from .models import IPIntelligence +from .risk_scorer import RISK_CRITICAL, RISK_HIGH, RISK_LOW, RISK_MEDIUM, RiskAssessment + +# --------------------------------------------------------------------------- +# Deterministic ID helpers (OpenCTI-style UUIDv5) +# --------------------------------------------------------------------------- +_NAMESPACE = uuid.UUID("d1f3a8c0-7e5b-4c3a-9f2d-6b8e1a0c5d4f") + + +def _det_id(prefix: str, value: str) -> str: + """Deterministic STIX id: ``prefix--``.""" + return f"{prefix}--{uuid.uuid5(_NAMESPACE, f'{prefix}:{value}')}" + + +def _now_iso() -> str: + return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --------------------------------------------------------------------------- +# TLP marking definitions (re-usable singletons) +# --------------------------------------------------------------------------- +TLP_WHITE = stix2.TLP_WHITE +TLP_GREEN = stix2.TLP_GREEN +TLP_AMBER = stix2.TLP_AMBER +TLP_RED = stix2.TLP_RED + +TLP_MAP = { + "TLP:WHITE": TLP_WHITE, + "TLP:CLEAR": TLP_WHITE, + "TLP:GREEN": TLP_GREEN, + "TLP:AMBER": TLP_AMBER, + "TLP:RED": TLP_RED, +} + + +class STIXMapper: + """Stateless mapper: call ``build_bundle_objects`` to get STIX objects.""" + + def __init__( + self, + author_name: str = "IPGeolocation.io", + default_marking: str = "TLP:WHITE", + confidence: int = 80, + ): + # Author identity (created once, referenced everywhere) + self._author = stix2.Identity( + id=_det_id("identity", author_name), + name=author_name, + identity_class="organization", + description=( + "IPGeolocation.io — Enterprise-grade IP geolocation " + "and threat intelligence provider." + ), + created_by_ref=None, + allow_custom=True, + ) + self._marking = TLP_MAP.get(default_marking, TLP_WHITE) + self._confidence = confidence + + # ------------------------------------------------------------------ # + # Public API + # ------------------------------------------------------------------ # + + def build_bundle_objects( + self, + intel: IPIntelligence, + risk: RiskAssessment, + observable_id: str, + observable_type: str, + *, + create_labels: bool = True, + create_indicators: bool = True, + create_relationships: bool = True, + create_notes: bool = True, + create_opinions: bool = False, + create_summary: bool = True, + indicator_threshold: int = 50, + ) -> list[Any]: + """Return all STIX objects for a single IP enrichment.""" + objects: list[Any] = [self._author, self._marking] + ext_refs = self._external_references(intel) + + # --- Locations ------------------------------------------------ + country_loc = self._make_country(intel) + if country_loc: + objects.append(country_loc) + if create_relationships: + objects.append( + self._relationship( + observable_id, + "located-at", + country_loc.id, + f"IP is geolocated in {intel.location.country_name}", + ) + ) + + city_loc = self._make_city(intel) + if city_loc: + objects.append(city_loc) + if create_relationships and country_loc: + objects.append( + self._relationship( + city_loc.id, + "located-at", + country_loc.id, + f"{intel.location.city} is in {intel.location.country_name}", + ) + ) + + # --- Autonomous System ---------------------------------------- + asn_obj = self._make_asn(intel) + if asn_obj: + objects.append(asn_obj) + if create_relationships: + objects.append( + self._relationship( + observable_id, + "belongs-to", + asn_obj.id, + f"IP belongs to {intel.asn.as_number}", + ) + ) + + # --- Organization / ISP / Company ----------------------------- + org = self._make_organization(intel) + if org: + objects.append(org) + if create_relationships and asn_obj: + objects.append( + self._relationship( + asn_obj.id, + "related-to", + org.id, + f"{intel.asn.as_number} operated by " + f"{intel.asn.organization or intel.company.name}", + ) + ) + + # --- Cloud / Hosting Provider --------------------------------- + cloud_org = self._make_cloud_provider(intel) + if cloud_org: + objects.append(cloud_org) + if create_relationships: + objects.append( + self._relationship( + observable_id, + "related-to", + cloud_org.id, + f"IP hosted on {intel.security.cloud_provider_name}", + ) + ) + + # --- Abuse Contact -------------------------------------------- + abuse_org = self._make_abuse_contact(intel) + if abuse_org: + objects.append(abuse_org) + if create_relationships: + objects.append( + self._relationship( + observable_id, + "related-to", + abuse_org.id, + "Abuse contact for IP network", + ) + ) + + # --- Hostname ------------------------------------------------- + hostname_obj = self._make_hostname(intel) + if hostname_obj: + objects.append(hostname_obj) + if create_relationships: + objects.append( + self._relationship( + observable_id, + "resolves-to", + hostname_obj.id, + f"IP resolves to {intel.hostname}", + ) + ) + + # --- Labels --------------------------------------------------- + labels: list[str] = [] + if create_labels: + labels = self._derive_labels(intel, risk) + + # --- Indicators ----------------------------------------------- + if create_indicators and risk.unified_score >= indicator_threshold: + indicator = self._make_indicator(intel, risk, labels, ext_refs) + if indicator: + objects.append(indicator) + if create_relationships: + objects.append( + self._relationship( + indicator.id, + "based-on", + observable_id, + "Indicator derived from enrichment", + ) + ) + + # --- Notes ---------------------------------------------------- + if create_notes and create_summary: + note = self._make_note(intel, risk, observable_id, labels) + if note: + objects.append(note) + + # --- Opinions ------------------------------------------------- + if create_opinions: + opinion = self._make_opinion(risk, observable_id) + if opinion: + objects.append(opinion) + + # --- Custom properties on observable -------------------------- + # OpenCTI allows updating observable custom props via STIX + obs_update = self._observable_update( + intel, + risk, + observable_id, + observable_type, + labels, + ext_refs, + ) + if obs_update: + objects.append(obs_update) + + return objects + + # ------------------------------------------------------------------ # + # Object Factories + # ------------------------------------------------------------------ # + + def _make_country(self, intel: IPIntelligence) -> Optional[stix2.Location]: + loc = intel.location + if not loc.country_name: + return None + return stix2.Location( + id=_det_id("location", f"country:{loc.country_code2}"), + name=loc.country_name, + country=loc.country_code2, + region=loc.continent_name, + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + allow_custom=True, + custom_properties={ + "x_opencti_location_type": "Country", + "x_opencti_aliases": [loc.country_code3] if loc.country_code3 else [], + }, + ) + + def _make_city(self, intel: IPIntelligence) -> Optional[stix2.Location]: + loc = intel.location + if not loc.city: + return None + lat = _safe_float(loc.latitude) + lon = _safe_float(loc.longitude) + props: dict[str, Any] = { + "x_opencti_location_type": "City", + } + kwargs: dict[str, Any] = dict( + id=_det_id("location", f"city:{loc.city}:{loc.country_code2}"), + name=f"{loc.city}, {loc.state_prov}" if loc.state_prov else loc.city, + country=loc.country_code2, + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + allow_custom=True, + custom_properties=props, + ) + if lat is not None: + kwargs["latitude"] = lat + if lon is not None: + kwargs["longitude"] = lon + return stix2.Location(**kwargs) + + def _make_asn(self, intel: IPIntelligence) -> Optional[Any]: + asn = intel.asn + if not asn.as_number: + return None + # AutonomousSystem is an OpenCTI custom SCO + asn_num = asn.as_number.replace("AS", "").strip() + try: + asn_int = int(asn_num) + except ValueError: + return None + return stix2.AutonomousSystem( + id=_det_id("autonomous-system", f"AS{asn_int}"), + number=asn_int, + name=asn.organization or asn.asn_name or f"AS{asn_int}", + allow_custom=True, + custom_properties={ + "x_opencti_description": ( + f"RIR: {asn.rir or 'N/A'} | " + f"Allocated: {asn.date_allocated or 'N/A'} | " + f"Type: {asn.type or 'N/A'}" + ), + }, + ) + + def _make_organization(self, intel: IPIntelligence) -> Optional[stix2.Identity]: + name = (intel.asn.organization or intel.company.name or "").strip() + if not name: + return None + desc_parts = [] + if intel.company.type: + desc_parts.append(f"Type: {intel.company.type}") + if intel.company.domain: + desc_parts.append(f"Domain: {intel.company.domain}") + if intel.asn.type: + desc_parts.append(f"Network type: {intel.asn.type}") + return stix2.Identity( + id=_det_id("identity", f"org:{name.lower()}"), + name=name, + identity_class="organization", + description=" | ".join(desc_parts) if desc_parts else None, + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + allow_custom=True, + ) + + def _make_cloud_provider(self, intel: IPIntelligence) -> Optional[stix2.Identity]: + if not intel.security.is_cloud_provider: + return None + name = intel.security.cloud_provider_name or "Unknown Cloud Provider" + return stix2.Identity( + id=_det_id("identity", f"cloud:{name.lower()}"), + name=name, + identity_class="organization", + description="Cloud / hosting infrastructure provider", + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + allow_custom=True, + ) + + def _make_abuse_contact(self, intel: IPIntelligence) -> Optional[stix2.Identity]: + ab = intel.abuse + if not ab.emails and not ab.name: + return None + name = ab.organization or ab.name or "Abuse Contact" + desc_parts = [] + if ab.emails: + desc_parts.append(f"Email: {', '.join(ab.emails)}") + if ab.phone_numbers: + desc_parts.append(f"Phone: {', '.join(ab.phone_numbers)}") + if ab.address: + desc_parts.append(f"Address: {ab.address}") + if ab.route: + desc_parts.append(f"Network: {ab.route}") + return stix2.Identity( + id=_det_id("identity", f"abuse:{name.lower()}:{ab.route}"), + name=f"Abuse Contact — {name}", + identity_class="organization", + description=" | ".join(desc_parts), + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + allow_custom=True, + ) + + def _make_hostname(self, intel: IPIntelligence) -> Optional[Any]: + if not intel.hostname: + return None + return stix2.DomainName( + value=intel.hostname, + allow_custom=True, + custom_properties={ + "created_by_ref": self._author.id, + }, + ) + + # ------------------------------------------------------------------ # + # Indicators + # ------------------------------------------------------------------ # + + def _make_indicator( + self, + intel: IPIntelligence, + risk: RiskAssessment, + labels: list[str], + ext_refs: list[dict], + ) -> Optional[stix2.Indicator]: + sec = intel.security + # Build description based on actual flags + reasons: list[str] = [] + if sec.is_tor: + reasons.append("TOR exit node") + if sec.is_known_attacker: + reasons.append("known attacker") + if sec.is_spam: + reasons.append("spam source") + if sec.is_bot: + reasons.append("bot activity") + if sec.is_vpn: + reasons.append("VPN") + if sec.is_proxy: + reasons.append("proxy") + if not reasons: + reasons.append(f"threat score {sec.threat_score}") + + ip_type = "ipv6" if ":" in intel.ip else "ipv4" + pattern = f"[{ip_type}-addr:value = '{intel.ip}']" + name = f"Malicious IP: {intel.ip}" + desc = ( + f"IPGeolocation.io enrichment flagged {intel.ip} as " + f"{risk.risk_level} risk ({risk.unified_score}/100). " + f"Signals: {', '.join(reasons)}." + ) + valid_from = _now_iso() + + return stix2.Indicator( + id=_det_id("indicator", f"ipgeo:{intel.ip}"), + name=name, + description=desc, + pattern=pattern, + pattern_type="stix", + valid_from=valid_from, + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + labels=labels[:10] if labels else [], + confidence=risk.confidence, + external_references=ext_refs, + allow_custom=True, + custom_properties={ + "x_opencti_score": risk.opencti_score, + "x_opencti_main_observable_type": ( + "IPv6-Addr" if ":" in intel.ip else "IPv4-Addr" + ), + }, + ) + + # ------------------------------------------------------------------ # + # Notes + # ------------------------------------------------------------------ # + + def _make_note( + self, + intel: IPIntelligence, + risk: RiskAssessment, + observable_id: str, + labels: list[str], + ) -> stix2.Note: + from .markdown_generator import MarkdownGenerator + + md = MarkdownGenerator().generate(intel, risk) + return stix2.Note( + id=_det_id("note", f"ipgeo:enrichment:{intel.ip}"), + abstract=f"IPGeolocation.io Enrichment — {intel.ip}", + content=md, + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + object_refs=[observable_id], + labels=labels[:5] if labels else [], + allow_custom=True, + ) + + # ------------------------------------------------------------------ # + # Opinions + # ------------------------------------------------------------------ # + + def _make_opinion( + self, + risk: RiskAssessment, + observable_id: str, + ) -> Optional[stix2.Opinion]: + opinion_map = { + RISK_LOW: "strongly-disagree", + RISK_MEDIUM: "neutral", + RISK_HIGH: "agree", + RISK_CRITICAL: "strongly-agree", + } + return stix2.Opinion( + id=_det_id("opinion", f"ipgeo:risk:{observable_id}"), + opinion=opinion_map.get(risk.risk_level, "neutral"), + explanation=risk.explanation, + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + object_refs=[observable_id], + allow_custom=True, + ) + + # ------------------------------------------------------------------ # + # Observable update (score + external refs) + # ------------------------------------------------------------------ # + + def _observable_update( + self, + intel: IPIntelligence, + risk: RiskAssessment, + observable_id: str, + observable_type: str, + labels: list[str], + ext_refs: list[dict], + ) -> Optional[Any]: + """Build a STIX SCO that OpenCTI will merge with the existing one.""" + ip_val = intel.ip + if not ip_val: + return None + + custom = { + "x_opencti_score": risk.opencti_score, + "x_opencti_description": risk.explanation, + } + kwargs: dict[str, Any] = { + "id": observable_id, + "value": ip_val, + "allow_custom": True, + "custom_properties": custom, + } + if ext_refs: + kwargs["external_references"] = ext_refs + + if "IPv6" in observable_type: + return stix2.IPv6Address(**kwargs) + return stix2.IPv4Address(**kwargs) + + # ------------------------------------------------------------------ # + # Labels + # ------------------------------------------------------------------ # + + @staticmethod + def _derive_labels(intel: IPIntelligence, risk: RiskAssessment) -> list[str]: + """Derive semantically meaningful labels — no duplicates.""" + labels: set[str] = set() + sec = intel.security + + if sec.is_vpn: + labels.add("vpn") + if sec.is_proxy: + labels.add("proxy") + if sec.is_residential_proxy: + labels.add("residential-proxy") + if sec.is_tor: + labels.add("tor") + if sec.is_relay: + labels.add("relay") + if sec.is_bot: + labels.add("bot") + if sec.is_spam: + labels.add("spam") + if sec.is_known_attacker: + labels.add("known-attacker") + if sec.is_anonymous: + labels.add("anonymous") + if sec.is_cloud_provider: + labels.add("cloud-provider") + + # Company / network type labels + comp_type = (intel.company.type or "").lower() + if comp_type: + labels.add(comp_type) # e.g. hosting, isp, business, education + net_type = (intel.asn.type or "").lower() + if net_type and net_type != comp_type: + labels.add(net_type) + + if intel.network.is_anycast: + labels.add("anycast") + + # Risk tier label + risk_label = f"risk:{risk.risk_level.lower()}" + labels.add(risk_label) + + return sorted(labels) + + # ------------------------------------------------------------------ # + # Relationships + # ------------------------------------------------------------------ # + + def _relationship( + self, + source: str, + rel_type: str, + target: str, + desc: str = "", + ) -> stix2.Relationship: + return stix2.Relationship( + id=_det_id("relationship", f"{source}:{rel_type}:{target}"), + relationship_type=rel_type, + source_ref=source, + target_ref=target, + description=desc, + created_by_ref=self._author.id, + object_marking_refs=[self._marking.id], + confidence=self._confidence, + allow_custom=True, + ) + + # ------------------------------------------------------------------ # + # External references + # ------------------------------------------------------------------ # + + @staticmethod + def _external_references(intel: IPIntelligence) -> list[dict]: + refs = [ + { + "source_name": "IPGeolocation.io", + "url": f"https://ipgeolocation.io/what-is-my-ip/{intel.ip}", + "description": f"IPGeolocation.io lookup for {intel.ip}", + } + ] + return refs + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _safe_float(val: str) -> Optional[float]: + try: + return float(val) if val else None + except (ValueError, TypeError): + return None diff --git a/internal-enrichment/ipgeolocation/tests/__init__.py b/internal-enrichment/ipgeolocation/tests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/internal-enrichment/ipgeolocation/tests/conftest.py b/internal-enrichment/ipgeolocation/tests/conftest.py new file mode 100644 index 00000000000..b20479d3975 --- /dev/null +++ b/internal-enrichment/ipgeolocation/tests/conftest.py @@ -0,0 +1,46 @@ +"""Shared test fixtures.""" + +import os +import sys + +import pytest + +# Ensure src is importable +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.markdown_generator import MarkdownGenerator +from src.models import IPIntelligence +from src.risk_scorer import RiskScorer +from src.stix_mapper import STIXMapper +from tests.mock_responses import MOCK_IPGEO_CLEAN, MOCK_IPGEO_FULL + + +@pytest.fixture +def high_risk_intel() -> IPIntelligence: + """An IP with multiple threat flags (VPN + proxy + known attacker).""" + return IPIntelligence.from_ipgeo_response(MOCK_IPGEO_FULL) + + +@pytest.fixture +def clean_intel() -> IPIntelligence: + """A clean IP (Google DNS) with no threat flags.""" + return IPIntelligence.from_ipgeo_response(MOCK_IPGEO_CLEAN) + + +@pytest.fixture +def scorer() -> RiskScorer: + return RiskScorer() + + +@pytest.fixture +def mapper() -> STIXMapper: + return STIXMapper( + author_name="IPGeolocation.io", + default_marking="TLP:WHITE", + confidence=80, + ) + + +@pytest.fixture +def md_gen() -> MarkdownGenerator: + return MarkdownGenerator() diff --git a/internal-enrichment/ipgeolocation/tests/mock_responses.py b/internal-enrichment/ipgeolocation/tests/mock_responses.py new file mode 100644 index 00000000000..15a9055074a --- /dev/null +++ b/internal-enrichment/ipgeolocation/tests/mock_responses.py @@ -0,0 +1,258 @@ +""" +Mock IPGeolocation.io API responses for unit tests. +Matches the actual v3 API response schemas. +""" + +MOCK_IPGEO_FULL = { + "ip": "2.56.188.34", + "hostname": "malicious-host.example.com", + "domain": "example.com", + "location": { + "continent_code": "NA", + "continent_name": "North America", + "country_code2": "US", + "country_code3": "USA", + "country_name": "United States", + "country_name_official": "United States of America", + "country_capital": "Washington, D.C.", + "state_prov": "Texas", + "state_code": "US-TX", + "district": "Dallas", + "city": "Dallas", + "locality": "Dallas", + "accuracy_radius": "20", + "confidence": "High", + "zipcode": "75201", + "latitude": "32.7767", + "longitude": "-96.7970", + "is_eu": False, + "country_flag": "https://ipgeolocation.io/static/flags/us_64.png", + "geoname_id": "4684888", + "country_emoji": "\U0001f1fa\U0001f1f8", + }, + "asn": { + "as_number": "AS15169", + "organization": "Google LLC", + "country": "US", + "type": "hosting", + "domain": "google.com", + "date_allocated": "2000-03-30", + "rir": "ARIN", + }, + "company": { + "name": "Google LLC", + "type": "hosting", + "domain": "google.com", + }, + "network": { + "connection_type": "hosting", + "route": "2.56.188.0/24", + "is_anycast": False, + }, + "time_zone": { + "name": "America/Chicago", + "offset": -6, + "offset_with_dst": -5, + "current_time": "2026-01-15 10:30:00.000-0600", + "is_dst": False, + "dst_savings": 1, + "dst_exists": True, + "dst_start": "2026-03-08T02:00:00", + "dst_end": "2026-11-01T02:00:00", + }, + "currency": { + "code": "USD", + "name": "US Dollar", + "symbol": "$", + }, + "security": { + "threat_score": 80, + "is_tor": False, + "is_proxy": True, + "proxy_provider_names": ["Zyte Proxy"], + "proxy_confidence_score": 80, + "proxy_last_seen": "2025-12-12", + "is_residential_proxy": True, + "is_vpn": True, + "vpn_provider_names": ["Nord VPN"], + "vpn_confidence_score": 80, + "vpn_last_seen": "2026-01-19", + "is_relay": False, + "relay_provider_name": "", + "is_anonymous": True, + "is_known_attacker": True, + "is_bot": False, + "is_spam": False, + "is_cloud_provider": True, + "cloud_provider_name": "Packethub S.A.", + }, + "abuse": { + "route": "2.56.188.0/24", + "country": "PA", + "name": "Packethub Abuse", + "organization": "Packethub S.A.", + "kind": "group", + "address": "Office 76, Plaza 2000, Panama City, Panama", + "emails": ["abuse@packethub.net"], + "phone_numbers": ["+507-123-4567"], + }, +} + + +MOCK_IPGEO_CLEAN = { + "ip": "8.8.8.8", + "hostname": "", + "domain": "", + "location": { + "continent_code": "NA", + "continent_name": "North America", + "country_code2": "US", + "country_code3": "USA", + "country_name": "United States", + "country_name_official": "United States of America", + "country_capital": "Washington, D.C.", + "state_prov": "California", + "state_code": "US-CA", + "district": "Santa Clara", + "city": "Mountain View", + "locality": "Mountain View", + "zipcode": "94043", + "latitude": "37.42240", + "longitude": "-122.08421", + "is_eu": False, + "geoname_id": "6301403", + "country_emoji": "\U0001f1fa\U0001f1f8", + }, + "asn": { + "as_number": "AS15169", + "organization": "Google LLC", + "country": "US", + "type": "hosting", + "domain": "google.com", + "date_allocated": "", + "rir": "ARIN", + }, + "company": { + "name": "Google LLC", + "type": "hosting", + "domain": "google.com", + }, + "network": { + "connection_type": "hosting", + "route": "8.8.8.0/24", + "is_anycast": True, + }, + "time_zone": { + "name": "America/Los_Angeles", + "offset": -8, + "current_time": "2026-01-15 08:30:00.000-0800", + }, + "currency": {"code": "USD", "name": "US Dollar", "symbol": "$"}, + "security": { + "threat_score": 0, + "is_tor": False, + "is_proxy": False, + "is_residential_proxy": False, + "is_vpn": False, + "is_relay": False, + "is_anonymous": False, + "is_known_attacker": False, + "is_bot": False, + "is_spam": False, + "is_cloud_provider": True, + "cloud_provider_name": "Google LLC", + }, + "abuse": { + "route": "8.8.8.0/24", + "country": "", + "name": "Abuse", + "organization": "Abuse", + "kind": "group", + "address": "1600 Amphitheatre Parkway, Mountain View, CA 94043", + "emails": ["network-abuse@google.com"], + "phone_numbers": ["+1-650-253-0000"], + }, +} + + +MOCK_ASN_DETAILED = { + "asn": { + "as_number": "AS17557", + "organization": "Pakistan Telecommunication Company Limited", + "country": "PK", + "type": "ISP", + "domain": "ptcl.net", + "date_allocated": "", + "asn_name": "PKTELECOM-AS-PK", + "allocation_status": "", + "num_of_ipv4_routes": "1305", + "num_of_ipv6_routes": "7", + "rir": "APNIC", + "routes": ["39.32.0.0/14", "39.36.0.0/16", "58.27.128.0/17"], + "downstreams": [ + {"as_number": "AS9387", "description": "AUGERE-PK", "country": "PK"}, + { + "as_number": "AS9541", + "description": "Cyber Internet Services", + "country": "PK", + }, + ], + "upstreams": [ + { + "as_number": "AS6453", + "description": "TATA COMMUNICATIONS", + "country": "US", + }, + ], + "peers": [ + { + "as_number": "AS6412", + "description": "Gulfnet International KEMS LLC", + "country": "KW", + }, + {"as_number": "AS13335", "description": "Cloudflare Inc.", "country": "US"}, + ], + "whois_response": "% [whois.apnic.net]\nas-block: AS17408 - AS17651\n", + } +} + + +MOCK_SECURITY_DEDICATED = { + "ip": "2.56.188.34", + "security": { + "threat_score": 80, + "is_tor": False, + "is_proxy": True, + "proxy_provider_names": ["Zyte Proxy"], + "proxy_confidence_score": 80, + "proxy_last_seen": "2025-12-12", + "is_residential_proxy": True, + "is_vpn": True, + "vpn_provider_names": ["Nord VPN"], + "vpn_confidence_score": 80, + "vpn_last_seen": "2026-01-19", + "is_relay": False, + "relay_provider_name": "", + "is_anonymous": True, + "is_known_attacker": True, + "is_bot": False, + "is_spam": False, + "is_cloud_provider": True, + "cloud_provider_name": "Packethub S.A.", + }, +} + + +MOCK_ABUSE_RESPONSE = { + "ip": "8.8.8.8", + "abuse": { + "route": "8.8.8.0/24", + "country": "", + "name": "Abuse", + "organization": "Abuse", + "kind": "group", + "address": "1600 Amphitheatre Parkway, Mountain View, CA 94043", + "emails": ["network-abuse@google.com"], + "phone_numbers": ["+1-650-253-0000"], + }, +} diff --git a/internal-enrichment/ipgeolocation/tests/test_api_client.py b/internal-enrichment/ipgeolocation/tests/test_api_client.py new file mode 100644 index 00000000000..9d18fc62269 --- /dev/null +++ b/internal-enrichment/ipgeolocation/tests/test_api_client.py @@ -0,0 +1,142 @@ +"""Tests for the IPGeolocation API client (mocked HTTP layer).""" + +from unittest.mock import MagicMock, patch + +import pytest + +from src.api_client import IPGeolocationAPIError, IPGeolocationClient +from tests.mock_responses import ( + MOCK_ABUSE_RESPONSE, + MOCK_ASN_DETAILED, + MOCK_IPGEO_FULL, + MOCK_SECURITY_DEDICATED, +) + + +def _mock_response(json_data, status_code=200, headers=None): + resp = MagicMock() + resp.status_code = status_code + resp.json.return_value = json_data + resp.text = str(json_data) + resp.headers = headers or {"X-Credits-Charged": "1"} + return resp + + +class TestIPGeolocationClient: + + def setup_method(self): + self.client = IPGeolocationClient( + api_key="test_key_1234", + base_url="https://api.ipgeolocation.io", + timeout=10, + max_retries=1, + ) + + @patch("src.api_client.requests.Session.get") + def test_lookup_unified(self, mock_get): + mock_get.return_value = _mock_response(MOCK_IPGEO_FULL) + intel = self.client.lookup_unified("2.56.188.34") + assert intel.ip == "2.56.188.34" + assert intel.location.country_name == "United States" + assert intel.security.threat_score == 80 + assert intel.security.is_vpn is True + assert intel.abuse.emails == ["abuse@packethub.net"] + + @patch("src.api_client.requests.Session.get") + def test_lookup_geo(self, mock_get): + mock_get.return_value = _mock_response(MOCK_IPGEO_FULL) + data = self.client.lookup_geo("2.56.188.34") + assert data["ip"] == "2.56.188.34" + + @patch("src.api_client.requests.Session.get") + def test_lookup_security(self, mock_get): + mock_get.return_value = _mock_response(MOCK_SECURITY_DEDICATED) + data = self.client.lookup_security("2.56.188.34") + assert data["security"]["threat_score"] == 80 + + @patch("src.api_client.requests.Session.get") + def test_lookup_asn(self, mock_get): + mock_get.return_value = _mock_response(MOCK_ASN_DETAILED) + data = self.client.lookup_asn("8.8.8.8") + assert "asn" in data + + @patch("src.api_client.requests.Session.get") + def test_lookup_abuse(self, mock_get): + mock_get.return_value = _mock_response(MOCK_ABUSE_RESPONSE) + data = self.client.lookup_abuse("8.8.8.8") + assert data["abuse"]["emails"] == ["network-abuse@google.com"] + + @patch("src.api_client.requests.Session.get") + def test_api_error_raises_exception(self, mock_get): + error_resp = _mock_response( + {"message": "Invalid API key"}, + status_code=401, + ) + mock_get.return_value = error_resp + with pytest.raises(IPGeolocationAPIError) as exc_info: + self.client.lookup_geo("1.1.1.1") + assert exc_info.value.status_code == 401 + + @patch("src.api_client.requests.Session.get") + def test_enrich_single_call_mode(self, mock_get): + """Single-call mode should make 2 calls: ipgeo + asn.""" + mock_get.return_value = _mock_response(MOCK_IPGEO_FULL) + intel = self.client.enrich( + "2.56.188.34", + single_call=True, + use_geo=True, + use_security=True, + use_asn=True, + use_abuse=True, + ) + assert intel.ip == "2.56.188.34" + # 2 calls: unified + ASN dedicated + assert mock_get.call_count == 2 + + @patch("src.api_client.requests.Session.get") + def test_enrich_dedicated_mode(self, mock_get): + """Dedicated mode calls each endpoint separately.""" + mock_get.side_effect = [ + _mock_response(MOCK_IPGEO_FULL), # geo + _mock_response(MOCK_SECURITY_DEDICATED), # security + _mock_response(MOCK_ASN_DETAILED), # asn + _mock_response(MOCK_ABUSE_RESPONSE), # abuse + ] + intel = self.client.enrich( + "2.56.188.34", + single_call=False, + use_geo=True, + use_security=True, + use_asn=True, + use_abuse=True, + ) + assert intel.ip == "2.56.188.34" + assert mock_get.call_count == 4 + + @patch("src.api_client.requests.Session.get") + def test_graceful_failure_on_asn(self, mock_get): + """ASN failure in single-call mode should not break enrichment.""" + # First call succeeds, second (ASN) fails + mock_get.side_effect = [ + _mock_response(MOCK_IPGEO_FULL), + _mock_response({"message": "Not found"}, status_code=404), + ] + intel = self.client.enrich( + "2.56.188.34", + single_call=True, + use_geo=True, + use_security=True, + use_asn=True, + use_abuse=True, + ) + # Should still have basic ASN from ipgeo response + assert intel.asn.as_number == "AS15169" + + def test_api_key_redacted_in_logs(self): + from src.api_client import _redact + + params = {"apiKey": "secret_key_12345", "ip": "1.1.1.1"} + redacted = _redact(params) + assert "secret_key_12345" not in str(redacted) + assert redacted["apiKey"].startswith("secr") + assert redacted["ip"] == "1.1.1.1" diff --git a/internal-enrichment/ipgeolocation/tests/test_markdown_generator.py b/internal-enrichment/ipgeolocation/tests/test_markdown_generator.py new file mode 100644 index 00000000000..d6f29ff20d7 --- /dev/null +++ b/internal-enrichment/ipgeolocation/tests/test_markdown_generator.py @@ -0,0 +1,71 @@ +"""Tests for the markdown note generator.""" + + +class TestMarkdownGenerator: + + def test_executive_summary_present(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Executive Summary" in md + assert "2.56.188.34" in md + + def test_summary_table_present(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## IP Summary" in md + assert "| Field | Value |" in md + assert "United States" in md + + def test_security_section_for_threats(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Security Assessment" in md + assert "VPN" in md or "vpn" in md.lower() + + def test_no_security_section_for_clean(self, clean_intel, md_gen, scorer): + risk = scorer.assess(clean_intel) + md = md_gen.generate(clean_intel, risk) + # Clean IP should not have threat narrative about being an attacker + assert "flagged for known attacker" not in md.lower() + + def test_infrastructure_profile(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Infrastructure Profile" in md + assert "Google LLC" in md + + def test_network_context(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Network Context" in md + assert "AS15169" in md + + def test_abuse_workflow(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Abuse Contact" in md + assert "abuse@packethub.net" in md + + def test_geo_intelligence(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Geo Intelligence" in md + assert "Dallas" in md + assert "32.7767" in md # latitude + + def test_timeline_with_dates(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Timeline" in md + assert "2025-12-12" in md # proxy_last_seen + assert "2026-01-19" in md # vpn_last_seen + + def test_confidence_explanation(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "## Enrichment Confidence" in md + + def test_footer_present(self, high_risk_intel, md_gen, scorer): + risk = scorer.assess(high_risk_intel) + md = md_gen.generate(high_risk_intel, risk) + assert "IPGeolocation.io OpenCTI Connector" in md diff --git a/internal-enrichment/ipgeolocation/tests/test_risk_scorer.py b/internal-enrichment/ipgeolocation/tests/test_risk_scorer.py new file mode 100644 index 00000000000..e7f4f398925 --- /dev/null +++ b/internal-enrichment/ipgeolocation/tests/test_risk_scorer.py @@ -0,0 +1,79 @@ +"""Tests for the risk scoring algorithm.""" + +from src.models import IPIntelligence, SecurityData +from src.risk_scorer import ( + RISK_CRITICAL, + RISK_LOW, + RISK_MEDIUM, +) + + +class TestRiskScorer: + + def test_clean_ip_is_low_risk(self, clean_intel, scorer): + result = scorer.assess(clean_intel) + assert result.risk_level == RISK_LOW + assert result.unified_score <= 20 + + def test_high_threat_ip_is_critical(self, high_risk_intel, scorer): + """IP with threat_score=80 + VPN + proxy + known attacker.""" + result = scorer.assess(high_risk_intel) + assert result.risk_level == RISK_CRITICAL + assert result.unified_score >= 81 + + def test_score_capped_at_100(self, scorer): + """Even with all flags, score should not exceed 100.""" + intel = IPIntelligence(ip="10.0.0.1") + intel.security = SecurityData( + threat_score=90, + is_tor=True, + is_vpn=True, + is_proxy=True, + is_known_attacker=True, + is_spam=True, + is_bot=True, + is_cloud_provider=True, + is_anonymous=True, + is_residential_proxy=True, + is_relay=True, + ) + result = scorer.assess(intel) + assert result.unified_score == 100 + + def test_explanation_contains_factors(self, high_risk_intel, scorer): + result = scorer.assess(high_risk_intel) + assert "known attacker" in result.explanation + assert ( + "VPN" in result.explanation.lower() or "vpn" in result.explanation.lower() + ) + + def test_contributing_factors_list(self, high_risk_intel, scorer): + result = scorer.assess(high_risk_intel) + assert len(result.contributing_factors) > 0 + factor_text = " ".join(result.contributing_factors).lower() + assert "vpn" in factor_text or "known attacker" in factor_text + + def test_confidence_derived_from_api(self, high_risk_intel, scorer): + result = scorer.assess(high_risk_intel) + # VPN confidence=80, proxy confidence=80, threat+20=100 + # average ~ 86 + assert 70 <= result.confidence <= 100 + + def test_no_threat_flags_defaults(self, scorer): + intel = IPIntelligence(ip="192.168.1.1") + intel.security = SecurityData(threat_score=10) + result = scorer.assess(intel) + assert result.risk_level == RISK_LOW + assert result.confidence == 30 # min(10+20, 100) = 30 + + def test_medium_range(self, scorer): + intel = IPIntelligence(ip="10.0.0.2") + intel.security = SecurityData(threat_score=30, is_vpn=True) + result = scorer.assess(intel) + assert result.risk_level == RISK_MEDIUM + + def test_opinion_text_mapping(self, scorer, high_risk_intel, clean_intel): + critical = scorer.assess(high_risk_intel) + low = scorer.assess(clean_intel) + assert critical.opinion_text == "strongly-agree" + assert low.opinion_text == "strongly-disagree" diff --git a/internal-enrichment/ipgeolocation/tests/test_stix_mapper.py b/internal-enrichment/ipgeolocation/tests/test_stix_mapper.py new file mode 100644 index 00000000000..c001fb8b675 --- /dev/null +++ b/internal-enrichment/ipgeolocation/tests/test_stix_mapper.py @@ -0,0 +1,179 @@ +"""Tests for STIX 2.1 object creation.""" + + +class TestSTIXMapper: + + def test_bundle_contains_author_identity(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + ) + identities = [o for o in objects if getattr(o, "type", "") == "identity"] + author = [i for i in identities if i.name == "IPGeolocation.io"] + assert len(author) >= 1 + + def test_country_location_created(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + ) + locations = [o for o in objects if getattr(o, "type", "") == "location"] + country_locs = [l for l in locations if l.country == "US"] + assert len(country_locs) >= 1 + + def test_city_location_has_coordinates(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + ) + locations = [o for o in objects if getattr(o, "type", "") == "location"] + cities = [l for l in locations if hasattr(l, "latitude")] + assert any(l.latitude is not None for l in cities) + + def test_asn_object_created(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + ) + asns = [o for o in objects if getattr(o, "type", "") == "autonomous-system"] + assert len(asns) >= 1 + assert asns[0].number == 15169 + + def test_relationships_created(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + create_relationships=True, + ) + rels = [o for o in objects if getattr(o, "type", "") == "relationship"] + assert len(rels) >= 2 # at least located-at + belongs-to + rel_types = {r.relationship_type for r in rels} + assert "located-at" in rel_types + assert "belongs-to" in rel_types + + def test_no_relationships_when_disabled(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + create_relationships=False, + ) + rels = [o for o in objects if getattr(o, "type", "") == "relationship"] + assert len(rels) == 0 + + def test_indicator_created_above_threshold(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + create_indicators=True, + indicator_threshold=50, + ) + indicators = [o for o in objects if getattr(o, "type", "") == "indicator"] + assert len(indicators) >= 1 + assert "ipv4-addr" in indicators[0].pattern + + def test_no_indicator_below_threshold(self, clean_intel, mapper, scorer): + risk = scorer.assess(clean_intel) + objects = mapper.build_bundle_objects( + intel=clean_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + create_indicators=True, + indicator_threshold=50, + ) + indicators = [o for o in objects if getattr(o, "type", "") == "indicator"] + assert len(indicators) == 0 + + def test_note_created(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + create_notes=True, + create_summary=True, + ) + notes = [o for o in objects if getattr(o, "type", "") == "note"] + assert len(notes) >= 1 + assert "Executive Summary" in notes[0].content + + def test_labels_derived_correctly(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + labels = mapper._derive_labels(high_risk_intel, risk) + assert "vpn" in labels + assert "proxy" in labels + assert "known-attacker" in labels + assert "cloud-provider" in labels + + def test_clean_ip_labels(self, clean_intel, mapper, scorer): + risk = scorer.assess(clean_intel) + labels = mapper._derive_labels(clean_intel, risk) + assert "risk:low" in labels + assert "cloud-provider" in labels # Google is a cloud provider + assert "vpn" not in labels + assert "known-attacker" not in labels + + def test_abuse_contact_identity(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + ) + identities = [o for o in objects if getattr(o, "type", "") == "identity"] + abuse_ids = [i for i in identities if "Abuse" in i.name] + assert len(abuse_ids) >= 1 + + def test_hostname_domain_created(self, high_risk_intel, mapper, scorer): + risk = scorer.assess(high_risk_intel) + objects = mapper.build_bundle_objects( + intel=high_risk_intel, + risk=risk, + observable_id="ipv4-addr--212aed99-b284-414d-960e-1c77921cb867", + observable_type="IPv4-Addr", + ) + domains = [o for o in objects if getattr(o, "type", "") == "domain-name"] + assert len(domains) >= 1 + assert domains[0].value == "malicious-host.example.com" + + def test_ipv6_pattern(self, mapper, scorer): + """IPv6 addresses should produce ipv6-addr patterns.""" + from src.models import IPIntelligence, SecurityData + + intel = IPIntelligence(ip="2001:db8::1") + intel.security = SecurityData(threat_score=60, is_vpn=True) + risk = scorer.assess(intel) + objects = mapper.build_bundle_objects( + intel=intel, + risk=risk, + observable_id="ipv6-addr--dedf9ff5-eed4-4d9c-bfe3-29e6e0f19b35", + observable_type="IPv6-Addr", + create_indicators=True, + indicator_threshold=50, + ) + indicators = [o for o in objects if getattr(o, "type", "") == "indicator"] + assert len(indicators) >= 1 + assert "ipv6-addr" in indicators[0].pattern From 512b1b0d05505725558ea949eaefb9976892d13e Mon Sep 17 00:00:00 2001 From: Abdul Mateen <105978906+mateen993@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:55:13 +0500 Subject: [PATCH 2/2] fix: connector auto default value Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- internal-enrichment/ipgeolocation/src/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal-enrichment/ipgeolocation/src/config.py b/internal-enrichment/ipgeolocation/src/config.py index 52226fe1a6e..971fb413eb0 100644 --- a/internal-enrichment/ipgeolocation/src/config.py +++ b/internal-enrichment/ipgeolocation/src/config.py @@ -78,7 +78,7 @@ class ConnectorConfig: type: str = "INTERNAL_ENRICHMENT" auto: bool = field( default_factory=lambda: _var( - "CONNECTOR_AUTO", ["connector", "auto"], default=True + "CONNECTOR_AUTO", ["connector", "auto"], default=False ) ) confidence_level: int = field(