import os
import requests
import json
import time
import schedule
import uuid
from datetime import datetime
# === CẤU HÌNH ===
PLATFORM_URL = os.getenv('PLATFORM_URL', 'http://localhost:8080')
PLATFORM_TOKEN = os.getenv('PLATFORM_TOKEN')
CONNECTOR_ID = os.getenv('CONNECTOR_ID', str(uuid.uuid4()))
def create_stix_bundle(observables, indicators, relationships):
"""
Tạo STIX 2.1 Bundle hoàn chỉnh
"""
bundle_id = f"bundle--{uuid.uuid4()}"
now = datetime.utcnow().isoformat() + 'Z'
bundle = {
"type": "bundle",
"id": bundle_id,
"spec_version": "2.1",
"objects": []
}
# Thêm observables
for obs in observables:
stix_type = map_to_stix_type(obs.get("type"))
bundle["objects"].append({
"type": stix_type,
"id": f"{stix_type}--{uuid.uuid4()}",
"spec_version": "2.1",
"value": obs.get("value"),
"created": now,
"modified": now,
"labels": obs.get("labels", [])
})
# Thêm indicators
for ind in indicators:
indicator_id = f"indicator--{uuid.uuid4()}"
bundle["objects"].append({
"type": "indicator",
"id": indicator_id,
"spec_version": "2.1",
"name": ind.get("name"),
"description": ind.get("description", ""),
"pattern": ind.get("pattern"),
"pattern_type": ind.get("pattern_type", "stix"),
"valid_from": now,
"confidence": ind.get("confidence", 80),
"labels": ind.get("labels", []),
"created": now,
"modified": now
})
# Nếu có relationship, thêm luôn vào bundle
if ind.get("observable_id"):
bundle["objects"].append({
"type": "relationship",
"id": f"relationship--{uuid.uuid4()}",
"spec_version": "2.1",
"source_ref": indicator_id,
"target_ref": ind.get("observable_id"),
"relationship_type": "indicates",
"confidence": ind.get("confidence", 80),
"created": now,
"modified": now
})
return bundle
def map_to_stix_type(platform_type):
"""
Map từ type của platform bạn sang STIX type
"""
mapping = {
"Domain": "domain-name",
"IPv4-Addr": "ipv4-addr",
"IPv6-Addr": "ipv6-addr",
"URL": "url",
"File": "file",
"Email-Addr": "email-addr"
}
return mapping.get(platform_type, platform_type.lower())
def send_stix_bundle(bundle):
"""
Gửi STIX Bundle lên platform
"""
url = f"{PLATFORM_URL}/api/import/stix"
headers = {
"Authorization": f"Bearer {PLATFORM_TOKEN}",
"Content-Type": "application/json"
}
try:
response = requests.post(url, json=bundle, headers=headers)
print(f"[SEND] STIX Bundle Status: {response.status_code}")
return response.json()
except Exception as e:
print(f"[SEND] Error: {e}")
return None
def process_and_send_data():
"""
Lấy dữ liệu từ bên thứ 3 và gửi STIX Bundle lên platform
"""
# 1. Lấy dữ liệu từ nguồn bên thứ 3
third_party_data = fetch_from_third_party()
# 2. Chuẩn bị dữ liệu cho STIX Bundle
observables = []
indicators = []
for item in third_party_data:
# Tạo observable
observable_type = item.get("type")
observable_value = item.get("value")
# Lưu observable_id để tạo relationship sau
observable_id = f"{map_to_stix_type(observable_type)}--{uuid.uuid4()}"
observables.append({
"type": observable_type,
"value": observable_value,
"labels": ["malicious", item.get("source", "third-party")]
})
# Tạo indicator tương ứng
indicators.append({
"name": f"Malicious {observable_type} - {observable_value}",
"description": f"Detected from {item.get('source', 'third-party')} feed",
"pattern": f"[{map_to_stix_type(observable_type)}:value = '{observable_value}']",
"pattern_type": "stix",
"confidence": item.get("confidence", 80),
"labels": ["malicious"],
"observable_id": observable_id # Để tạo relationship
})
# 3. Tạo STIX Bundle
bundle = create_stix_bundle(observables, indicators, [])
# 4. Gửi lên platform
result = send_stix_bundle(bundle)
if result:
print(f"[SEND] Successfully imported {len(observables)} observables")
else:
print("[SEND] Failed to import data")
def fetch_from_third_party():
"""
Hàm giả định lấy dữ liệu từ bên thứ 3
"""
return [
{"type": "Domain", "value": "example.com", "confidence": 95, "source": "AlienVault"},
{"type": "IPv4-Addr", "value": "8.8.8.8", "confidence": 80, "source": "VirusTotal"},
{"type": "URL", "value": "https://malicious.com", "confidence": 90, "source": "PhishingFeed"}
]
# === CÁC HÀM REGISTER, HEARTBEAT ===
def register_connector():
url = f"{PLATFORM_URL}/api/connectors/register"
payload = {
"connector_id": CONNECTOR_ID,
"name": "STIX Demo Connector",
"type": "IMPORT",
"configuration_schema": {
"interval": {"type": "integer", "label": "Sync Interval (s)", "default": 60}
}
}
headers = {"Authorization": f"Bearer {PLATFORM_TOKEN}"}
try:
response = requests.post(url, json=payload, headers=headers)
print(f"[REGISTER] Status: {response.status_code}")
except Exception as e:
print(f"[REGISTER] Error: {e}")
def send_heartbeat():
url = f"{PLATFORM_URL}/api/connectors/heartbeat"
payload = {"connector_id": CONNECTOR_ID, "status": "RUNNING"}
headers = {"Authorization": f"Bearer {PLATFORM_TOKEN}"}
try:
requests.post(url, json=payload, headers=headers)
except:
pass
if __name__ == "__main__":
register_connector()
schedule.every(30).seconds.do(send_heartbeat)
schedule.every(10).minutes.do(process_and_send_data)
while True:
schedule.run_pending()
time.sleep(1)