메뉴
STIX (Structured Threat Information eXpression) and TAXII (Trusted Automated eXchange of Intelligence Information) are OASIS open standards for representing and transporting cyber threat intelligence.
기본적으로 소스를 먼저 확인하는 Prompt가 선택됩니다. 직접 명령으로 전환하거나 로컬 사본을 다운로드할 수도 있습니다.
설치 여부를 결정하기 전에 SKILL.md와 SkillsMP에 표시된 보조 파일을 읽어 보세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/AgentFlocks/flocks --skill implementing-stix-taxii-feed-integration명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
深信服 EDR 登录态管理、首页仪表盘、威胁资产分析和资产清点分类统计 API 采集。用户提到深信服 EDR、EDR、资产清点或 sangfor EDR 时必须先加载本 skill。
统一处理浏览器使用任务，支持可见浏览器 CDP 直连、专用 headless CDP、agent-browser。Use when the user asks to browse websites, interact with pages, fill forms, capture screenshots, reuse an existing Chrome/Chromium/Edge login session, work with an already-open browser/sidebar browser, access login-only/internal/dynamic pages, or automate browser actions.
用于处理 OneSEC/OneDNS 终端安全平台相关任务，支持通过API或者结合浏览器进行操作。只要用户提到 OneSEC、微步 EDR等相关操纵需求时，必须先加载本 skill。本 skill 是 OneSEC 平台操作的唯一决策入口：在未阅读本 skill 并完成模式判断前，不要直接调用任何 `onesec_*` tool。
SOC 직업 분류 기준
SKILL.md 표시 중
| name | implementing-stix-taxii-feed-integration |
| description | STIX (Structured Threat Information eXpression) and TAXII (Trusted Automated eXchange of Intelligence Information) are OASIS open standards for representing and transporting cyber threat intelligence. |
| domain | cybersecurity |
| subdomain | threat-intelligence |
| tags | ["threat-intelligence","cti","ioc","mitre-attack","stix","taxii","feed-integration","oasis"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
| nist_csf | ["ID.RA-01","ID.RA-05","DE.CM-01","DE.AE-02"] |
STIX (Structured Threat Information eXpression) and TAXII (Trusted Automated eXchange of Intelligence Information) are OASIS open standards for representing and transporting cyber threat intelligence. This skill covers implementing a STIX/TAXII 2.1 feed consumer and producer using Python, configuring TAXII server discovery, collection management, polling for new intelligence, parsing STIX 2.1 objects, and integrating feeds into SIEM and TIP platforms.
taxii2-client, stix2, cti-taxii-client librariesTAXII defines a RESTful API with three service types:
STIX objects are categorized as:
A Bundle is a collection of STIX objects transmitted together. Bundles have a unique ID and contain an array of objects. TAXII collections serve bundles in response to GET requests.
taxii2client.v21 Server, Collection, as_pages
server = Server(, user=, password=)
()
()
api_root server.api_roots:
()
()
collection api_root.collections:
()
()
()
from taxii2client.v21 import Collection, as_pages
import json
# Connect to Enterprise ATT&CK collection
ENTERPRISE_ATTACK_ID = "95ecc380-afe9-11e4-9b6c-751b66dd541e"
collection = Collection(
f"https://cti-taxii.mitre.org/stix/collections/{ENTERPRISE_ATTACK_ID}/",
user="",
password="",
)
print(f"Collection: {collection.title}")
# Fetch all objects (paginated)
all_objects = []
for envelope in as_pages(collection.get_objects, per_request=50):
objects = envelope.get("objects", [])
all_objects.extend(objects)
print(f" Fetched {len(objects)} objects (total: {len(all_objects)})")
print(f"\nTotal objects retrieved: {len(all_objects)}")
# Categorize by type
type_counts = {}
for obj in all_objects:
obj_type = obj.get("type", "unknown")
type_counts[obj_type] = type_counts.get(obj_type, 0) + 1
for obj_type, count in sorted(type_counts.items()):
print(f" {obj_type}: {count}")
from stix2 import parse, Filter, MemoryStore
# Load objects into a MemoryStore for querying
store = MemoryStore(stix_data=all_objects)
# Query for all indicators
indicators = store.query([Filter("type", "=", "indicator")])
print(f"Indicators: {len(indicators)}")
for ind in indicators[:5]:
print(f" {ind.name}: {ind.pattern}")
# Query for malware
malware_list = store.query([Filter("type", "=", "malware")])
print(f"\nMalware families: {len(malware_list)}")
# Query for threat actors
actors = store.query([Filter("type", "=", "intrusion-set")])
print(f"Threat actors: {len(actors)}")
# Find relationships for a specific object
def get_related(store, source_id):
relationships = store.query([
Filter("type", "=", "relationship"),
Filter("source_ref", "=", source_id),
])
return relationships
# Example: Get all techniques used by APT28
apt28 = store.query([
Filter("type", "=", "intrusion-set"),
Filter("name", "=", "APT28"),
])
if apt28:
rels = get_related(store, apt28[0].id)
for rel in rels:
target = store.get(rel.target_ref)
if target:
print(f" {rel.relationship_type} -> {target.name} ({target.type})")
from taxii2client.v21 import Collection, as_pages
from stix2 import parse, Bundle
from datetime import datetime, timedelta
import json
class TAXIIConsumer:
"""Consume STIX/TAXII 2.1 feeds and extract IOCs."""
def __init__(self, collection_url, user="", password=""):
self.collection = Collection(collection_url, user=user, password=password)
self.last_poll = None
def poll_new_objects(self, added_after=None):
"""Poll for objects added after a specific timestamp."""
if added_after is None:
added_after = (
self.last_poll or
(datetime.utcnow() - timedelta(days=1)).strftime(
"%Y-%m-%dT%H:%M:%S.000Z"
)
)
all_objects = []
kwargs = {"added_after": added_after}
for envelope in as_pages(
self.collection.get_objects, per_request=100, **kwargs
):
objects = envelope.get("objects", [])
all_objects.extend(objects)
self.last_poll = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z")
return all_objects
def extract_indicators(self, objects):
"""Extract actionable indicators from STIX objects."""
indicators = []
for obj in objects:
if obj.get("type") == "indicator":
indicators.append({
"id": obj.get("id"),
"name": obj.get("name", ""),
"pattern": obj.get("pattern", ""),
"pattern_type": obj.get("pattern_type", ""),
"valid_from": obj.get("valid_from", ""),
"valid_until": obj.get("valid_until", ""),
"indicator_types": obj.get("indicator_types", []),
"confidence": obj.get("confidence", 0),
"labels": obj.get("labels", []),
})
return indicators
def extract_observables(self, objects):
"""Extract STIX Cyber Observables."""
observables = []
observable_types = {
"ipv4-addr", "ipv6-addr", "domain-name", "url",
"file", "email-addr", "network-traffic",
}
for obj in objects:
if obj.get("type") in observable_types:
observables.append({
"type": obj["type"],
"value": obj.get("value", ""),
"id": obj.get("id"),
})
return observables
# Usage
consumer = TAXIIConsumer(
f"https://cti-taxii.mitre.org/stix/collections/{ENTERPRISE_ATTACK_ID}/"
)
new_objects = consumer.poll_new_objects()
indicators = consumer.extract_indicators(new_objects)
print(f"New indicators: {len(indicators)}")
# medallion configuration (medallion.conf)
TAXII_CONFIG = {
"backend": {
"module_class": "MemoryBackend",
},
"users": {
"admin": "admin_password",
"readonly": "readonly_password",
},
"taxii": {
"max_content_length": 10485760,
},
}
# Run medallion server:
# pip install medallion
# python -m medallion --config medallion.conf --port 5000
# Add objects to local TAXII server
import requests
def push_to_taxii(server_url, collection_id, stix_bundle, user, password):
"""Push STIX bundle to a TAXII 2.1 collection."""
url = f"{server_url}/collections/{collection_id}/objects/"
headers = {
"Content-Type": "application/stix+json;version=2.1",
"Accept": "application/taxii+json;version=2.1",
}
response = requests.post(
url,
json=stix_bundle,
headers=headers,
auth=(user, password),
timeout=30,
)
return response.json()