OpenCTI로 IOC 보강 파이프라인 구축은(는) 무엇을 하나요?
개요
OpenCTI는 STIX 2.1을 네이티브 데이터 모델로 사용하는 사이버 위협 인텔리전스 지식 관리를 위한 오픈소스 플랫폼입니다. 이 스킬은 OpenCTI의 커넥터 생태계를 활용하여 VirusTotal, Shodan, AbuseIPDB, GreyNoise 등 다양한 소스의 컨텍스트로 지표를 보강하는 자동화된 IOC 보강 파이프라인을 구축하는 방법을 다룹니다. 파이프라인은 새로 수집된 지표를 자동으로 보강하고, 알려진 위협 행위자 및 캠페인과 연관시키며, 분석가의 우선순위 지정을 위해 점수를 매깁니다.
사용 시기
- 환경에서 IOC 보강 파이프라인을 구축하거나 구성할 때
- 규정 준수 요구 사항에 맞춰 보안 통제를 수립할 때
- 이 도메인의 보안 아키텍처를 구축하거나 개선할 때
- 이 구현이 필요한 보안 평가를 수행할 때
사전 요구 사항
- OpenCTI 배포를 위한 Docker 및 Docker Compose
- Python 3.9+ 및
pycti라이브러리 - VirusTotal, Shodan, AbuseIPDB, GreyNoise용 API 키
- STIX 2.1 데이터 모델 및 관계 이해
- OpenCTI 백엔드용 ElasticSearch 또는 OpenSearch
- 커넥터 메시징용 RabbitMQ 또는 Redis
주요 개념
OpenCTI 아키텍처
OpenCTI는 ElasticSearch를 스토리지로, Redis/RabbitMQ를 커넥터 통신으로 사용하는 GraphQL API 프론트엔드를 사용합니다. 데이터는 관계와 함께 STIX 2.1 객체로 네이티브 저장됩니다. 커넥터는 외부 가져오기(피드 수집), 내부 가져오기(파일 구문 분석), 내부 보강(컨텍스트 추가), 스트림(실시간 내보내기)으로 분류됩니다.
보강 커넥터 모델
내부 보강 커넥터는 새로운 관찰 가능 항목이 생성될 때 자동으로 트리거되거나 분석가가 수동으로 트리거합니다. 각 커넥터는 STIX 객체를 수신하고, 외부 서비스를 쿼리한 후, 추가 컨텍스트, 레이블 및 관계로 원래 관찰 가능 항목을 보강하는 STIX 2.1 번들을 반환합니다.
신뢰도 점수
OpenCTI는 지표에 대해 0-100 신뢰도 척도를 사용합니다. 보강 커넥터는 외부 검증(VirusTotal 탐지 비율, Shodan 노출 데이터, AbuseIPDB 보고서 수, GreyNoise 분류 결과)을 기반으로 신뢰도 점수를 업데이트할 수 있습니다.
워크플로우
1단계: Docker Compose로 OpenCTI 배포
# docker-compose.yml (핵심 서비스)
version: '3'
services:
opencti:
image: opencti/platform:6.4.4
environment:
- APP__PORT=8080
- [email protected]
- APP__ADMIN__PASSWORD=ChangeMeNow
- APP__ADMIN__TOKEN=your-admin-token-uuid
- ELASTICSEARCH__URL=http://elasticsearch:9200
- MINIO__ENDPOINT=minio
- RABBITMQ__HOSTNAME=rabbitmq
ports:
- "8080:8080"
depends_on:
- elasticsearch
- minio
- rabbitmq
- redis
connector-virustotal:
image: opencti/connector-virustotal:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-virustotal-id
- CONNECTOR_NAME=VirusTotal
- CONNECTOR_SCOPE=StixFile,Artifact,IPv4-Addr,Domain-Name,Url
- CONNECTOR_AUTO=true
- VIRUSTOTAL_TOKEN=your-vt-api-key
- VIRUSTOTAL_MAX_TLP=TLP:AMBER
connector-shodan:
image: opencti/connector-shodan:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-shodan-id
- CONNECTOR_NAME=Shodan
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- SHODAN_TOKEN=your-shodan-api-key
- SHODAN_MAX_TLP=TLP:AMBER
connector-abuseipdb:
image: opencti/connector-abuseipdb:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-abuseipdb-id
- CONNECTOR_NAME=AbuseIPDB
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- ABUSEIPDB_API_KEY=your-abuseipdb-key
2단계: 사용자 정의 보강 커넥터 구축
import os
from pycti import OpenCTIConnectorHelper, get_config_variable
from stix2 import (
Bundle, Indicator, Note, Relationship,
IPv4Address, DomainName
)
import requests
class CustomEnrichmentConnector:
def __init__(self):
config = {
"opencti": {
"url": os.environ.get("OPENCTI_URL"),
"token": os.environ.get("OPENCTI_TOKEN"),
},
"connector": {
"id": os.environ.get("CONNECTOR_ID"),
"name": "CustomEnrichment",
"scope": "IPv4-Addr,Domain-Name,Url",
"auto": True,
"type": "INTERNAL_ENRICHMENT",
},
}
self.helper = OpenCTIConnectorHelper(config)
self.helper.listen(self._process_message)
def _process_message(self, data):
entity_id = data["entity_id"]
stix_object = self.helper.api.stix_cyber_observable.read(id=entity_id)
if not stix_object:
return "관찰 가능 항목을 찾을 수 없음"
observable_type = stix_object["entity_type"]
observable_value = stix_object.get("value", "")
enrichment_results = []
if observable_type == "IPv4-Addr":
enrichment_results = self._enrich_ip(observable_value, entity_id)
elif observable_type == "Domain-Name":
enrichment_results = self._enrich_domain(observable_value, entity_id)
if enrichment_results:
bundle = Bundle(objects=enrichment_results, allow_custom=True)
self.helper.send_stix2_bundle(bundle.serialize())
return "보강 완료"
def _enrich_ip(self, ip_address, entity_id):
"""IP 주소를 GreyNoise, AbuseIPDB 컨텍스트로 보강."""
objects = []
# GreyNoise Community API
try:
gn_response = requests.get(
f"https://api.greynoise.io/v3/community/{ip_address}",
headers={"key": os.environ.get("GREYNOISE_API_KEY")},
timeout=30,
)
if gn_response.status_code == 200:
gn_data = gn_response.json()
classification = gn_data.get("classification", "unknown")
noise = gn_data.get("noise", False)
riot = gn_data.get("riot", False)
note_content = (
f"## GreyNoise 보강\n"
f"- 분류: {classification}\n"
f"- 인터넷 노이즈: {noise}\n"
f"- RIOT(정상 서비스): {riot}\n"
f"- 이름: {gn_data.get('name', 'N/A')}\n"
f"- 마지막 발견: {gn_data.get('last_seen', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"GreyNoise: {classification}",
allow_custom=True,
)
objects.append(note)
# 분류에 따라 레이블 추가
if classification == "malicious":
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:malicious"
)
elif riot:
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:benign-service"
)
except Exception as e:
self.helper.log_error(f"GreyNoise 보강 실패: {e}")
return objects
def _enrich_domain(self, domain, entity_id):
"""도메인을 WHOIS 및 DNS 컨텍스트로 보강."""
objects = []
try:
# 도메인 보강에 SecurityTrails API 사용
st_response = requests.get(
f"https://api.securitytrails.com/v1/domain/{domain}",
headers={"APIKEY": os.environ.get("SECURITYTRAILS_API_KEY")},
timeout=30,
)
if st_response.status_code == 200:
st_data = st_response.json()
current_dns = st_data.get("current_dns", {})
a_records = [
r.get("ip") for r in current_dns.get("a", {}).get("values", [])
]
note_content = (
f"## SecurityTrails 보강\n"
f"- A 레코드: {', '.join(a_records)}\n"
f"- Alexa 순위: {st_data.get('alexa_rank', 'N/A')}\n"
f"- 호스트명: {st_data.get('hostname', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"SecurityTrails: {domain}",
allow_custom=True,
)
objects.append(note)
except Exception as e:
self.helper.log_error(f"SecurityTrails 보강 실패: {e}")
return objects
if __name__ == "__main__":
connector = CustomEnrichmentConnector()