tessl/pypi-taxii2-client

Python TAXII 2.X client library for sharing cyber threat intelligence via STIX protocol

Overview

Eval results

Files

Collection Operations

Name: tessl/pypi-taxii2-client
Author: tessl

Collection-level operations for managing STIX objects including retrieval, addition, deletion, and manifest operations. Collections are repositories within TAXII API roots that store and serve cyber threat intelligence data in STIX format.

Capabilities

Collection Connection

Connect to a specific TAXII collection endpoint with authentication and configuration options.

class Collection:
    def __init__(self, url, conn=None, user=None, password=None, verify=True,
                 proxies=None, collection_info=None, auth=None, cert=None):
        """
        Create a TAXII collection endpoint connection.
        
        Parameters:
        - url (str): URL of TAXII collection endpoint
        - conn (_HTTPConnection, optional): Reuse existing connection
        - user (str, optional): Username for HTTP basic authentication
        - password (str, optional): Password for HTTP basic authentication
        - verify (bool): Validate SSL certificates (default: True)
        - proxies (dict, optional): HTTP/HTTPS proxy settings
        - collection_info (dict, optional): Pre-loaded collection metadata
        - auth (requests.auth.AuthBase, optional): Custom authentication object
        - cert (str or tuple, optional): SSL client certificate path or (cert, key) tuple
        """

Collection Information

Access collection metadata including identity, permissions, and supported media types.

@property
def id(self) -> str:
    """Collection identifier (required)."""

@property
def title(self) -> str:
    """Collection title (required)."""

@property
def description(self) -> str:
    """Collection description (optional)."""

@property
def alias(self) -> str:
    """Collection alias (optional, TAXII 2.1 only)."""

@property
def can_read(self) -> bool:
    """Whether collection allows read operations (required)."""

@property
def can_write(self) -> bool:
    """Whether collection allows write operations (required)."""

@property
def media_types(self) -> list[str]:
    """List of supported media types for objects (optional)."""

@property
def custom_properties(self) -> dict:
    """Custom collection properties not defined in TAXII spec."""

@property
def objects_url(self) -> str:
    """URL for objects endpoint."""

@property
def _raw(self) -> dict:
    """Raw collection information response (parsed JSON)."""

Object Retrieval

Retrieve STIX objects from the collection with filtering and pagination support.

def get_objects(self, accept=None, **filter_kwargs) -> dict:
    """
    Retrieve objects from the collection.
    
    Parameters:
    - accept (str, optional): Media type for Accept header
    - **filter_kwargs: Filter parameters (added_after, match[type], match[id], 
                       match[version], limit, next for TAXII 2.1; 
                       start, per_request for TAXII 2.0)
    
    Returns:
    dict: Response envelope/bundle containing objects
    
    Raises:
    AccessError: If collection doesn't allow reading
    """

def get_object(self, obj_id, accept=None, **filter_kwargs) -> dict:
    """
    Retrieve a specific object by ID.
    
    Parameters:
    - obj_id (str): STIX object identifier
    - accept (str, optional): Media type for Accept header
    - **filter_kwargs: Filter parameters (match[version] for version filtering)
    
    Returns:
    dict: Response envelope/bundle containing the object
    
    Raises:
    AccessError: If collection doesn't allow reading
    """

def get_manifest(self, accept=None, **filter_kwargs) -> dict:
    """
    Retrieve object manifests (metadata without full objects).
    
    Parameters:
    - accept (str, optional): Media type for Accept header
    - **filter_kwargs: Filter parameters (same as get_objects)
    
    Returns:
    dict: Response containing object manifests
    
    Raises:
    AccessError: If collection doesn't allow reading
    """

Object Management

Add, modify, and delete STIX objects in the collection.

def add_objects(self, envelope, wait_for_completion=True, poll_interval=1,
               timeout=60, accept=None, content_type=None) -> Status:
    """
    Add objects to the collection.
    
    Parameters:
    - envelope (dict|str|bytes): STIX envelope/bundle containing objects to add
    - wait_for_completion (bool): Whether to wait for async completion
    - poll_interval (int): Polling interval in seconds for completion
    - timeout (int): Maximum wait time in seconds (0 or negative for no limit)
    - accept (str, optional): Media type for Accept header
    - content_type (str, optional): Media type for Content-Type header
    
    Returns:
    Status: Status object tracking the operation
    
    Raises:
    AccessError: If collection doesn't allow writing
    TypeError: If envelope type is not supported
    """

def delete_object(self, obj_id, accept=None, **filter_kwargs) -> dict:
    """
    Delete a specific object by ID (TAXII 2.1 only).
    
    Parameters:
    - obj_id (str): STIX object identifier to delete
    - accept (str, optional): Media type for Accept header
    - **filter_kwargs: Filter parameters (match[version] for version filtering)
    
    Returns:
    dict: Response confirming deletion
    
    Raises:
    AccessError: If collection doesn't allow writing
    """

def object_versions(self, obj_id, accept=None, **filter_kwargs) -> dict:
    """
    Get all versions of a specific object (TAXII 2.1 only).
    
    Parameters:
    - obj_id (str): STIX object identifier
    - accept (str, optional): Media type for Accept header
    - **filter_kwargs: Filter parameters for version filtering
    
    Returns:
    dict: Response containing object versions
    
    Raises:
    AccessError: If collection doesn't allow reading
    """

Collection Operations

Refresh collection information and manage the collection connection.

def refresh(self, accept=None) -> None:
    """
    Update collection information.
    
    Parameters:
    - accept (str, optional): Media type for Accept header
    """

def close(self) -> None:
    """Close the collection connection."""

def __enter__(self):
    """Context manager entry."""

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit."""

Usage Examples

Basic Object Retrieval

from taxii2client import Server

# Get collection from server
server = Server("https://taxii-server.example.com/taxii2/")
api_root = server.default
collection = api_root.collections[0]

# Check permissions
if not collection.can_read:
    print("Collection doesn't allow reading")
    exit(1)

# Get all objects
response = collection.get_objects()
objects = response.get('objects', [])
print(f"Retrieved {len(objects)} objects")

# Print first object
if objects:
    obj = objects[0]
    print(f"First object: {obj.get('type')} - {obj.get('id')}")

Filtered Object Retrieval

from datetime import datetime, timezone

# Filter by object type
indicators = collection.get_objects(type="indicator")
print(f"Found {len(indicators.get('objects', []))} indicators")

# Filter by multiple types
malware_and_tools = collection.get_objects(type=["malware", "tool"])

# Filter by date (objects added after specific time)
recent_date = datetime(2023, 1, 1, tzinfo=timezone.utc)
recent_objects = collection.get_objects(added_after=recent_date)

# Filter by specific object IDs
specific_ids = [
    "indicator--12345678-1234-5678-9012-123456789012",
    "malware--87654321-4321-8765-2109-876543210987"
]
specific_objects = collection.get_objects(id=specific_ids)

Paginated Retrieval

from taxii2client import as_pages

# TAXII 2.1 pagination
for page in as_pages(collection.get_objects, per_request=100):
    objects = page.get('objects', [])
    print(f"Processing page with {len(objects)} objects")
    
    for obj in objects:
        print(f"  {obj.get('type')}: {obj.get('id')}")

# TAXII 2.0 pagination (if using v20 client)
# for page in as_pages(collection.get_objects, start=0, per_request=100):
#     objects = page.get('objects', [])
#     print(f"Processing page with {len(objects)} objects")

Object Manifest Retrieval

# Get manifests instead of full objects (lighter weight)
manifest_response = collection.get_manifest()
manifests = manifest_response.get('objects', [])

print(f"Collection contains {len(manifests)} objects:")
for manifest in manifests:
    print(f"  {manifest.get('id')}")
    print(f"    Versions: {manifest.get('versions', [])}")
    print(f"    Media Types: {manifest.get('media_types', [])}")

Adding Objects

# Create STIX envelope/bundle
stix_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"],
            "spec_version": "2.1"
        },
        {
            "type": "malware",
            "id": "malware--87654321-4321-8765-2109-876543210987",
            "created": "2023-01-01T00:00:00.000Z",
            "modified": "2023-01-01T00:00:00.000Z",
            "name": "BadStuff",
            "labels": ["trojan"],
            "spec_version": "2.1"
        }
    ]
}

# Check write permissions
if not collection.can_write:
    print("Collection doesn't allow writing")
    exit(1)

# Add objects synchronously (wait for completion)
status = collection.add_objects(stix_envelope, wait_for_completion=True)
print(f"Add operation status: {status.status}")
print(f"Total objects: {status.total_count}")
print(f"Successful: {status.success_count}")
print(f"Failed: {status.failure_count}")

# Add objects asynchronously
status = collection.add_objects(stix_envelope, wait_for_completion=False)
print(f"Operation started with status ID: {status.id}")
print(f"Initial status: {status.status}")

# Poll manually for completion
import time
while status.status != "complete":
    time.sleep(1)
    status.refresh()
    print(f"Status: {status.status} ({status.success_count}/{status.total_count})")

TAXII 2.1 Specific Operations

# These operations only work with TAXII 2.1 collections
from taxii2client.v21 import Collection

# Get specific object
obj_id = "indicator--12345678-1234-5678-9012-123456789012"
response = collection.get_object(obj_id)
if response.get('objects'):
    obj = response['objects'][0]
    print(f"Retrieved: {obj.get('type')} - {obj.get('name', 'N/A')}")

# Get all versions of an object
versions_response = collection.object_versions(obj_id)
versions = versions_response.get('versions', [])
print(f"Object {obj_id} has {len(versions)} versions")

# Delete an object
delete_response = collection.delete_object(obj_id)
print(f"Delete response: {delete_response}")

Error Handling

from taxii2client.exceptions import AccessError, ValidationError, TAXIIServiceException

try:
    # Attempt to read from collection
    objects = collection.get_objects()
except AccessError as e:
    print(f"Access denied: {e}")
except ValidationError as e:
    print(f"Validation error: {e}")
except TAXIIServiceException as e:
    print(f"TAXII service error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

try:
    # Attempt to add objects
    status = collection.add_objects(malformed_envelope)
except TypeError as e:
    print(f"Invalid envelope format: {e}")
except AccessError as e:
    print(f"Write access denied: {e}")

Context Manager Usage

# Direct collection connection with automatic cleanup
collection_url = "https://taxii-server.example.com/taxii2/api1/collections/indicators/"
with Collection(collection_url, user="user", password="pass") as collection:
    if collection.can_read:
        objects = collection.get_objects()
        print(f"Retrieved {len(objects.get('objects', []))} objects")
    # Connection automatically closed when exiting context

Install with Tessl CLI