Python TAXII 2.X client library for sharing cyber threat intelligence via STIX protocol
npx @tessl/cli install tessl/pypi-taxii2-client@2.3.0A Python client library for interacting with TAXII 2.X (Trusted Automated eXchange of Indicator Information) servers. This library enables cyber threat intelligence sharing through the standardized TAXII protocol, supporting both TAXII 2.0 and 2.1 specifications for consuming and sharing structured threat data in STIX format.
pip install taxii2-clientfrom taxii2client import Server, ApiRoot, Collection, Status, as_pagesFor TAXII 2.1 (default/latest):
from taxii2client.v21 import Server, ApiRoot, Collection, Status, as_pagesFor TAXII 2.0 (legacy):
from taxii2client.v20 import Server, ApiRoot, Collection, Status, as_pagesAuthentication and utilities:
from taxii2client.common import TokenAuth
from taxii2client.exceptions import TAXIIServiceException, AccessError, ValidationErrorfrom taxii2client import Server, Collection
import json
# Connect to a TAXII server
server = Server(
url="https://example.com/taxii2/",
user="username",
password="password"
)
# Get server information
print(f"Server: {server.title}")
print(f"Available API Roots: {len(server.api_roots)}")
# Access the default API root
api_root = server.default or server.api_roots[0]
print(f"API Root: {api_root.title}")
# List available collections
for collection in api_root.collections:
print(f"Collection: {collection.title} (ID: {collection.id})")
print(f" Can Read: {collection.can_read}, Can Write: {collection.can_write}")
# Work with a specific collection
collection = api_root.collections[0] # Get first collection
# Get objects from the collection
if collection.can_read:
objects = collection.get_objects()
print(f"Retrieved {len(objects.get('objects', []))} objects")
# Get paginated results
for page in as_pages(collection.get_objects, per_request=100):
objects = page.get('objects', [])
print(f"Page contains {len(objects)} objects")
# Add objects to collection (if allowed)
if collection.can_write:
# STIX bundle/envelope to add
envelope = {
"objects": [
{
"type": "indicator",
"id": "indicator--12345678-1234-5678-9012-123456789012",
"created": "2023-01-01T00:00:00.000Z",
"modified": "2023-01-01T00:00:00.000Z",
"pattern": "[file:hashes.MD5 = 'd41d8cd98f00b204e9800998ecf8427e']",
"labels": ["malicious-activity"]
}
]
}
# Add objects and wait for completion
status = collection.add_objects(envelope, wait_for_completion=True)
print(f"Add status: {status.status}")
print(f"Success count: {status.success_count}")TAXII2-Client follows the TAXII protocol hierarchy:
The library supports both TAXII 2.0 and 2.1 protocols with nearly identical APIs, automatically handling protocol differences including pagination mechanisms, media types, and endpoint variations.
Server-level operations for discovering TAXII services, API roots, and server capabilities. Provides the entry point for all TAXII interactions.
class Server:
def __init__(self, url, conn=None, user=None, password=None, verify=True,
proxies=None, auth=None, cert=None): ...
@property
def title(self) -> str: ...
@property
def description(self) -> str: ...
@property
def contact(self) -> str: ...
@property
def default(self) -> ApiRoot: ...
@property
def api_roots(self) -> list[ApiRoot]: ...
def refresh(self) -> None: ...API root operations for managing collections, checking capabilities, and retrieving status information within a specific TAXII API root.
class ApiRoot:
def __init__(self, url, conn=None, user=None, password=None, verify=True,
proxies=None, auth=None, cert=None): ...
@property
def title(self) -> str: ...
@property
def description(self) -> str: ...
@property
def versions(self) -> list[str]: ...
@property
def max_content_length(self) -> int: ...
@property
def collections(self) -> list[Collection]: ...
def refresh(self, accept=None) -> None: ...
def refresh_information(self, accept=None) -> None: ...
def refresh_collections(self, accept=None) -> None: ...
def get_status(self, status_id, accept=None) -> Status: ...Collection-level operations for managing STIX objects including retrieval, addition, deletion, and manifest operations.
class Collection:
def __init__(self, url, conn=None, user=None, password=None, verify=True,
proxies=None, collection_info=None, auth=None, cert=None): ...
@property
def id(self) -> str: ...
@property
def title(self) -> str: ...
@property
def can_read(self) -> bool: ...
@property
def can_write(self) -> bool: ...
def get_objects(self, accept=None, **filter_kwargs) -> dict: ...
def get_object(self, obj_id, accept=None, **filter_kwargs) -> dict: ...
def add_objects(self, envelope, wait_for_completion=True, poll_interval=1,
timeout=60, accept=None, content_type=None) -> Status: ...
def get_manifest(self, accept=None, **filter_kwargs) -> dict: ...
# TAXII 2.1 only:
def delete_object(self, obj_id, accept=None, **filter_kwargs) -> dict: ...
def object_versions(self, obj_id, accept=None, **filter_kwargs) -> dict: ...Status tracking for asynchronous operations including polling, completion checking, and result analysis.
class Status:
def __init__(self, url, conn=None, user=None, password=None, verify=True,
proxies=None, status_info=None, auth=None, cert=None): ...
@property
def id(self) -> str: ...
@property
def status(self) -> str: ...
@property
def total_count(self) -> int: ...
@property
def success_count(self) -> int: ...
@property
def failure_count(self) -> int: ...
@property
def pending_count(self) -> int: ...
@property
def successes(self) -> list: ...
@property
def failures(self) -> list: ...
@property
def pendings(self) -> list: ...
def refresh(self, accept=None) -> None: ...
def wait_until_final(self, poll_interval=1, timeout=60) -> None: ...
def __bool__(self) -> bool: ...Pagination utilities for handling large result sets across different TAXII versions with automatic page traversal.
def as_pages(func, per_request=0, *args, **kwargs):
"""
Generator for TAXII endpoints supporting pagination.
Parameters:
- func: Collection method supporting pagination (get_objects, get_manifest)
- per_request: Items per request (0 for server default)
Yields:
dict: Response envelope/bundle for each page
"""Authentication mechanisms and connection management including basic auth, token auth, and SSL client certificates.
class TokenAuth:
def __init__(self, key: str): ...
def __call__(self, r): ...DEFAULT_USER_AGENT: str = "taxii2-client/2.3.0"
MEDIA_TYPE_STIX_V20: str = "application/vnd.oasis.stix+json; version=2.0"
MEDIA_TYPE_TAXII_V20: str = "application/vnd.oasis.taxii+json; version=2.0"
MEDIA_TYPE_TAXII_V21: str = "application/taxii+json;version=2.1"class TAXIIServiceException(Exception):
"""Base exception for all TAXII client errors."""
class InvalidArgumentsError(TAXIIServiceException):
"""Invalid arguments passed to method."""
class AccessError(TAXIIServiceException):
"""Read/write access denied to collection."""
class ValidationError(TAXIIServiceException):
"""Data validation failed."""
class InvalidJSONError(TAXIIServiceException):
"""Invalid JSON received from server."""