tessl/pypi-taxii2-client

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

Overview

Eval results

Files

Authentication & Connection

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

Authentication mechanisms and connection management including basic auth, token auth, SSL client certificates, and connection pooling. TAXII2-Client supports multiple authentication methods for secure connections to TAXII servers.

Capabilities

Basic Authentication

HTTP Basic Authentication using username and password credentials.

# Basic auth via constructor parameters
Server(url="https://taxii-server.example.com/taxii2/", 
       user="username", password="password")

ApiRoot(url="https://taxii-server.example.com/taxii2/api1/",
        user="username", password="password")

Collection(url="https://taxii-server.example.com/taxii2/api1/collections/indicators/",
           user="username", password="password")

Token Authentication

Token-based authentication using Authorization headers with Bearer or custom token schemes.

class TokenAuth:
    def __init__(self, key: str):
        """
        Create token-based authentication.
        
        Parameters:
        - key (str): Authentication token/API key
        
        Note:
        Adds 'Authorization: Token <key>' header to requests
        """
    
    def __call__(self, r):
        """Apply token authentication to request."""

Custom Authentication

Support for custom authentication mechanisms using requests authentication objects.

# Any requests.auth.AuthBase subclass can be used
Server(url="https://taxii-server.example.com/taxii2/", 
       auth=custom_auth_object)

SSL/TLS Configuration

SSL certificate verification and client certificate authentication options.

# SSL configuration parameters (available on all endpoint classes)
def __init__(self, url, verify=True, cert=None, **kwargs):
    """
    Parameters:
    - verify (bool or str): 
        - True: Verify SSL certificates using default CA bundle
        - False: Disable SSL verification (not recommended)
        - str: Path to custom CA bundle file
    - cert (str or tuple):
        - str: Path to client certificate file (.pem)
        - tuple: (cert_path, key_path) for separate cert and key files
    """

Proxy Configuration

HTTP and HTTPS proxy support for connections through corporate firewalls.

# Proxy configuration (available on all endpoint classes)
def __init__(self, url, proxies=None, **kwargs):
    """
    Parameters:
    - proxies (dict): Proxy configuration mapping
        - 'http': HTTP proxy URL
        - 'https': HTTPS proxy URL
    """

Connection Management

Connection pooling and session management with proper resource cleanup.

class _HTTPConnection:
    def __init__(self, user=None, password=None, verify=True, proxies=None,
                 user_agent=None, version="2.0", auth=None, cert=None):
        """
        HTTP connection session manager.
        
        Parameters:
        - user (str, optional): Username for basic auth
        - password (str, optional): Password for basic auth
        - verify (bool): SSL certificate verification
        - proxies (dict, optional): Proxy configuration
        - user_agent (str, optional): Custom User-Agent header
        - version (str): TAXII protocol version ("2.0" or "2.1")
        - auth (AuthBase, optional): Custom authentication object
        - cert (str or tuple, optional): SSL client certificate
        """
    
    @property
    def session(self) -> requests.Session:
        """Underlying requests Session object."""
    
    def close(self) -> None:
        """Close all connections and clean up resources."""

Usage Examples

Basic Authentication

from taxii2client import Server

# Username/password authentication
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    user="my_username",
    password="my_password"
)

print(f"Connected to: {server.title}")

# Basic auth is automatically applied to all requests
api_root = server.default
collections = api_root.collections

Token Authentication

from taxii2client import Server
from taxii2client.common import TokenAuth

# Create token authenticator
token_auth = TokenAuth("your-api-token-here")

# Use with server
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    auth=token_auth
)

# Token auth is inherited by child objects
api_root = server.default
collection = api_root.collections[0]
objects = collection.get_objects()  # Uses token auth automatically

Custom Authentication

import requests.auth
from taxii2client import Server

# HTTP Digest Authentication
digest_auth = requests.auth.HTTPDigestAuth("username", "password")
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    auth=digest_auth
)

# OAuth2 Bearer Token (custom implementation)
class BearerAuth(requests.auth.AuthBase):
    def __init__(self, token):
        self.token = token
    
    def __call__(self, r):
        r.headers["Authorization"] = f"Bearer {self.token}"
        return r

bearer_auth = BearerAuth("your-oauth2-token")
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    auth=bearer_auth
)

# API Key in custom header
class ApiKeyAuth(requests.auth.AuthBase):
    def __init__(self, api_key, header_name="X-API-Key"):
        self.api_key = api_key
        self.header_name = header_name
    
    def __call__(self, r):
        r.headers[self.header_name] = self.api_key
        return r

api_key_auth = ApiKeyAuth("your-api-key", "X-Custom-API-Key")
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    auth=api_key_auth
)

SSL Configuration

from taxii2client import Server

# Disable SSL verification (not recommended for production)
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    verify=False
)

# Use custom CA bundle
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    verify="/path/to/custom-ca-bundle.crt"
)

# Client certificate authentication (single file)
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    cert="/path/to/client-cert.pem"
)

# Client certificate authentication (separate cert and key)
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    cert=("/path/to/client-cert.crt", "/path/to/client-key.key")
)

# Combined SSL and basic auth
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    user="username",
    password="password",
    verify="/path/to/ca-bundle.crt",
    cert=("/path/to/client.crt", "/path/to/client.key")
)

Proxy Configuration

from taxii2client import Server

# HTTP and HTTPS proxy
proxies = {
    'http': 'http://proxy.company.com:8080',
    'https': 'https://proxy.company.com:8080'
}

server = Server(
    url="https://taxii-server.example.com/taxii2/",
    user="username",
    password="password",
    proxies=proxies
)

# Authenticated proxy
proxies = {
    'http': 'http://user:pass@proxy.company.com:8080',
    'https': 'https://user:pass@proxy.company.com:8080'
}

server = Server(
    url="https://taxii-server.example.com/taxii2/",
    proxies=proxies
)

# SOCKS proxy (requires requests[socks])
proxies = {
    'http': 'socks5://proxy.company.com:1080',
    'https': 'socks5://proxy.company.com:1080'
}

server = Server(
    url="https://taxii-server.example.com/taxii2/",
    proxies=proxies
)

Connection Sharing

from taxii2client import Server, ApiRoot, Collection
from taxii2client.common import _HTTPConnection

# Create shared connection
conn = _HTTPConnection(
    user="username",
    password="password",
    verify=True,
    proxies={'https': 'https://proxy.company.com:8080'}
)

# Use shared connection across multiple endpoints
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    conn=conn
)

# Direct API root with shared connection
api_root = ApiRoot(
    url="https://taxii-server.example.com/taxii2/api1/",
    conn=conn
)

# Direct collection with shared connection
collection = Collection(
    url="https://taxii-server.example.com/taxii2/api1/collections/indicators/",    
    conn=conn
)

# All endpoints share the same connection pool
objects = collection.get_objects()

# Close shared connection when done
conn.close()

Context Managers for Resource Management

# Automatic connection cleanup with context managers
with Server("https://taxii-server.example.com/taxii2/", user="user", password="pass") as server:
    print(f"Server: {server.title}")
    
    with server.default as api_root:
        print(f"API Root: {api_root.title}")
        
        for collection in api_root.collections:
            with collection:
                if collection.can_read:
                    objects = collection.get_objects()
                    print(f"Collection {collection.title}: {len(objects.get('objects', []))} objects")
# All connections automatically closed

Advanced Connection Configuration

from taxii2client.common import _HTTPConnection
import requests

# Custom session configuration
session = requests.Session()
session.timeout = 30  # 30 second timeout
session.stream = True  # Enable streaming for large responses

# Create connection with custom session
conn = _HTTPConnection(
    user="username",
    password="password",
    user_agent="MyApp/1.0 (Custom TAXII Client)"
)

# Modify session after creation
conn.session.timeout = 60
conn.session.headers.update({'X-Custom-Header': 'value'})

# Use connection
server = Server(
    url="https://taxii-server.example.com/taxii2/",
    conn=conn
)

Error Handling for Authentication

from taxii2client import Server
from taxii2client.exceptions import TAXIIServiceException
import requests

try:
    server = Server(
        url="https://taxii-server.example.com/taxii2/",
        user="wrong_user",
        password="wrong_password"
    )
    
    # This will trigger authentication
    title = server.title
    
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        print("Authentication failed: Invalid credentials")
    elif e.response.status_code == 403:
        print("Access forbidden: Insufficient permissions")
    else:
        print(f"HTTP error: {e}")
        
except requests.exceptions.SSLError as e:
    print(f"SSL error: {e}")
    print("Consider using verify=False for testing (not recommended for production)")
    
except requests.exceptions.ConnectionError as e:
    print(f"Connection error: {e}")
    print("Check proxy settings and network connectivity")
    
except TAXIIServiceException as e:
    print(f"TAXII service error: {e}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

Environment-Based Configuration

import os
from taxii2client import Server
from taxii2client.common import TokenAuth

# Load configuration from environment variables
def create_server_from_env():
    url = os.environ.get('TAXII_SERVER_URL')
    if not url:
        raise ValueError("TAXII_SERVER_URL environment variable required")
    
    # Check for different auth methods
    if os.environ.get('TAXII_API_TOKEN'):
        auth = TokenAuth(os.environ['TAXII_API_TOKEN'])
        return Server(url=url, auth=auth)
    elif os.environ.get('TAXII_USERNAME') and os.environ.get('TAXII_PASSWORD'):
        return Server(
            url=url,
            user=os.environ['TAXII_USERNAME'],
            password=os.environ['TAXII_PASSWORD']
        )
    else:
        # No authentication
        return Server(url=url)

# Usage
server = create_server_from_env()

# Additional environment-based configuration
verify_ssl = os.environ.get('TAXII_VERIFY_SSL', 'true').lower() == 'true'
ca_bundle = os.environ.get('TAXII_CA_BUNDLE')
client_cert = os.environ.get('TAXII_CLIENT_CERT')
client_key = os.environ.get('TAXII_CLIENT_KEY')

server = Server(
    url=os.environ['TAXII_SERVER_URL'],
    user=os.environ.get('TAXII_USERNAME'),
    password=os.environ.get('TAXII_PASSWORD'),
    verify=ca_bundle if ca_bundle else verify_ssl,
    cert=(client_cert, client_key) if client_cert and client_key else client_cert
)

Install with Tessl CLI