0
# Authentication & Connection
1

2
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.
3

4
## Capabilities
5

6
### Basic Authentication
7

8
HTTP Basic Authentication using username and password credentials.
9

10
```python { .api }
11
# Basic auth via constructor parameters
12
Server(url="https://taxii-server.example.com/taxii2/", 
13
       user="username", password="password")
14

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

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

22
### Token Authentication
23

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

26
```python { .api }
27
class TokenAuth:
28
    def __init__(self, key: str):
29
        """
30
        Create token-based authentication.
31
        
32
        Parameters:
33
        - key (str): Authentication token/API key
34
        
35
        Note:
36
        Adds 'Authorization: Token <key>' header to requests
37
        """
38
    
39
    def __call__(self, r):
40
        """Apply token authentication to request."""
41
```
42

43
### Custom Authentication
44

45
Support for custom authentication mechanisms using requests authentication objects.
46

47
```python { .api }
48
# Any requests.auth.AuthBase subclass can be used
49
Server(url="https://taxii-server.example.com/taxii2/", 
50
       auth=custom_auth_object)
51
```
52

53
### SSL/TLS Configuration
54

55
SSL certificate verification and client certificate authentication options.
56

57
```python { .api }
58
# SSL configuration parameters (available on all endpoint classes)
59
def __init__(self, url, verify=True, cert=None, **kwargs):
60
    """
61
    Parameters:
62
    - verify (bool or str): 
63
        - True: Verify SSL certificates using default CA bundle
64
        - False: Disable SSL verification (not recommended)
65
        - str: Path to custom CA bundle file
66
    - cert (str or tuple):
67
        - str: Path to client certificate file (.pem)
68
        - tuple: (cert_path, key_path) for separate cert and key files
69
    """
70
```
71

72
### Proxy Configuration
73

74
HTTP and HTTPS proxy support for connections through corporate firewalls.
75

76
```python { .api }
77
# Proxy configuration (available on all endpoint classes)
78
def __init__(self, url, proxies=None, **kwargs):
79
    """
80
    Parameters:
81
    - proxies (dict): Proxy configuration mapping
82
        - 'http': HTTP proxy URL
83
        - 'https': HTTPS proxy URL
84
    """
85
```
86

87
### Connection Management
88

89
Connection pooling and session management with proper resource cleanup.
90

91
```python { .api }
92
class _HTTPConnection:
93
    def __init__(self, user=None, password=None, verify=True, proxies=None,
94
                 user_agent=None, version="2.0", auth=None, cert=None):
95
        """
96
        HTTP connection session manager.
97
        
98
        Parameters:
99
        - user (str, optional): Username for basic auth
100
        - password (str, optional): Password for basic auth
101
        - verify (bool): SSL certificate verification
102
        - proxies (dict, optional): Proxy configuration
103
        - user_agent (str, optional): Custom User-Agent header
104
        - version (str): TAXII protocol version ("2.0" or "2.1")
105
        - auth (AuthBase, optional): Custom authentication object
106
        - cert (str or tuple, optional): SSL client certificate
107
        """
108
    
109
    @property
110
    def session(self) -> requests.Session:
111
        """Underlying requests Session object."""
112
    
113
    def close(self) -> None:
114
        """Close all connections and clean up resources."""
115
```
116

117
## Usage Examples
118

119
### Basic Authentication
120

121
```python
122
from taxii2client import Server
123

124
# Username/password authentication
125
server = Server(
126
    url="https://taxii-server.example.com/taxii2/",
127
    user="my_username",
128
    password="my_password"
129
)
130

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

133
# Basic auth is automatically applied to all requests
134
api_root = server.default
135
collections = api_root.collections
136
```
137

138
### Token Authentication
139

140
```python
141
from taxii2client import Server
142
from taxii2client.common import TokenAuth
143

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

147
# Use with server
148
server = Server(
149
    url="https://taxii-server.example.com/taxii2/",
150
    auth=token_auth
151
)
152

153
# Token auth is inherited by child objects
154
api_root = server.default
155
collection = api_root.collections[0]
156
objects = collection.get_objects()  # Uses token auth automatically
157
```
158

159
### Custom Authentication
160

161
```python
162
import requests.auth
163
from taxii2client import Server
164

165
# HTTP Digest Authentication
166
digest_auth = requests.auth.HTTPDigestAuth("username", "password")
167
server = Server(
168
    url="https://taxii-server.example.com/taxii2/",
169
    auth=digest_auth
170
)
171

172
# OAuth2 Bearer Token (custom implementation)
173
class BearerAuth(requests.auth.AuthBase):
174
    def __init__(self, token):
175
        self.token = token
176
    
177
    def __call__(self, r):
178
        r.headers["Authorization"] = f"Bearer {self.token}"
179
        return r
180

181
bearer_auth = BearerAuth("your-oauth2-token")
182
server = Server(
183
    url="https://taxii-server.example.com/taxii2/",
184
    auth=bearer_auth
185
)
186

187
# API Key in custom header
188
class ApiKeyAuth(requests.auth.AuthBase):
189
    def __init__(self, api_key, header_name="X-API-Key"):
190
        self.api_key = api_key
191
        self.header_name = header_name
192
    
193
    def __call__(self, r):
194
        r.headers[self.header_name] = self.api_key
195
        return r
196

197
api_key_auth = ApiKeyAuth("your-api-key", "X-Custom-API-Key")
198
server = Server(
199
    url="https://taxii-server.example.com/taxii2/",
200
    auth=api_key_auth
201
)
202
```
203

204
### SSL Configuration
205

206
```python
207
from taxii2client import Server
208

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

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

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

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

233
# Combined SSL and basic auth
234
server = Server(
235
    url="https://taxii-server.example.com/taxii2/",
236
    user="username",
237
    password="password",
238
    verify="/path/to/ca-bundle.crt",
239
    cert=("/path/to/client.crt", "/path/to/client.key")
240
)
241
```
242

243
### Proxy Configuration
244

245
```python
246
from taxii2client import Server
247

248
# HTTP and HTTPS proxy
249
proxies = {
250
    'http': 'http://proxy.company.com:8080',
251
    'https': 'https://proxy.company.com:8080'
252
}
253

254
server = Server(
255
    url="https://taxii-server.example.com/taxii2/",
256
    user="username",
257
    password="password",
258
    proxies=proxies
259
)
260

261
# Authenticated proxy
262
proxies = {
263
    'http': 'http://user:pass@proxy.company.com:8080',
264
    'https': 'https://user:pass@proxy.company.com:8080'
265
}
266

267
server = Server(
268
    url="https://taxii-server.example.com/taxii2/",
269
    proxies=proxies
270
)
271

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

278
server = Server(
279
    url="https://taxii-server.example.com/taxii2/",
280
    proxies=proxies
281
)
282
```
283

284
### Connection Sharing
285

286
```python
287
from taxii2client import Server, ApiRoot, Collection
288
from taxii2client.common import _HTTPConnection
289

290
# Create shared connection
291
conn = _HTTPConnection(
292
    user="username",
293
    password="password",
294
    verify=True,
295
    proxies={'https': 'https://proxy.company.com:8080'}
296
)
297

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

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

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

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

319
# Close shared connection when done
320
conn.close()
321
```
322

323
### Context Managers for Resource Management
324

325
```python
326
# Automatic connection cleanup with context managers
327
with Server("https://taxii-server.example.com/taxii2/", user="user", password="pass") as server:
328
    print(f"Server: {server.title}")
329
    
330
    with server.default as api_root:
331
        print(f"API Root: {api_root.title}")
332
        
333
        for collection in api_root.collections:
334
            with collection:
335
                if collection.can_read:
336
                    objects = collection.get_objects()
337
                    print(f"Collection {collection.title}: {len(objects.get('objects', []))} objects")
338
# All connections automatically closed
339
```
340

341
### Advanced Connection Configuration
342

343
```python
344
from taxii2client.common import _HTTPConnection
345
import requests
346

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

352
# Create connection with custom session
353
conn = _HTTPConnection(
354
    user="username",
355
    password="password",
356
    user_agent="MyApp/1.0 (Custom TAXII Client)"
357
)
358

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

363
# Use connection
364
server = Server(
365
    url="https://taxii-server.example.com/taxii2/",
366
    conn=conn
367
)
368
```
369

370
### Error Handling for Authentication
371

372
```python
373
from taxii2client import Server
374
from taxii2client.exceptions import TAXIIServiceException
375
import requests
376

377
try:
378
    server = Server(
379
        url="https://taxii-server.example.com/taxii2/",
380
        user="wrong_user",
381
        password="wrong_password"
382
    )
383
    
384
    # This will trigger authentication
385
    title = server.title
386
    
387
except requests.exceptions.HTTPError as e:
388
    if e.response.status_code == 401:
389
        print("Authentication failed: Invalid credentials")
390
    elif e.response.status_code == 403:
391
        print("Access forbidden: Insufficient permissions")
392
    else:
393
        print(f"HTTP error: {e}")
394
        
395
except requests.exceptions.SSLError as e:
396
    print(f"SSL error: {e}")
397
    print("Consider using verify=False for testing (not recommended for production)")
398
    
399
except requests.exceptions.ConnectionError as e:
400
    print(f"Connection error: {e}")
401
    print("Check proxy settings and network connectivity")
402
    
403
except TAXIIServiceException as e:
404
    print(f"TAXII service error: {e}")
405
    
406
except Exception as e:
407
    print(f"Unexpected error: {e}")
408
```
409

410
### Environment-Based Configuration
411

412
```python
413
import os
414
from taxii2client import Server
415
from taxii2client.common import TokenAuth
416

417
# Load configuration from environment variables
418
def create_server_from_env():
419
    url = os.environ.get('TAXII_SERVER_URL')
420
    if not url:
421
        raise ValueError("TAXII_SERVER_URL environment variable required")
422
    
423
    # Check for different auth methods
424
    if os.environ.get('TAXII_API_TOKEN'):
425
        auth = TokenAuth(os.environ['TAXII_API_TOKEN'])
426
        return Server(url=url, auth=auth)
427
    elif os.environ.get('TAXII_USERNAME') and os.environ.get('TAXII_PASSWORD'):
428
        return Server(
429
            url=url,
430
            user=os.environ['TAXII_USERNAME'],
431
            password=os.environ['TAXII_PASSWORD']
432
        )
433
    else:
434
        # No authentication
435
        return Server(url=url)
436

437
# Usage
438
server = create_server_from_env()
439

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

446
server = Server(
447
    url=os.environ['TAXII_SERVER_URL'],
448
    user=os.environ.get('TAXII_USERNAME'),
449
    password=os.environ.get('TAXII_PASSWORD'),
450
    verify=ca_bundle if ca_bundle else verify_ssl,
451
    cert=(client_cert, client_key) if client_cert and client_key else client_cert
452
)
453
```