0
# Collection Operations
1

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

4
## Capabilities
5

6
### Collection Connection
7

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

10
```python { .api }
11
class Collection:
12
    def __init__(self, url, conn=None, user=None, password=None, verify=True,
13
                 proxies=None, collection_info=None, auth=None, cert=None):
14
        """
15
        Create a TAXII collection endpoint connection.
16
        
17
        Parameters:
18
        - url (str): URL of TAXII collection endpoint
19
        - conn (_HTTPConnection, optional): Reuse existing connection
20
        - user (str, optional): Username for HTTP basic authentication
21
        - password (str, optional): Password for HTTP basic authentication
22
        - verify (bool): Validate SSL certificates (default: True)
23
        - proxies (dict, optional): HTTP/HTTPS proxy settings
24
        - collection_info (dict, optional): Pre-loaded collection metadata
25
        - auth (requests.auth.AuthBase, optional): Custom authentication object
26
        - cert (str or tuple, optional): SSL client certificate path or (cert, key) tuple
27
        """
28
```
29

30
### Collection Information
31

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

34
```python { .api }
35
@property
36
def id(self) -> str:
37
    """Collection identifier (required)."""
38

39
@property
40
def title(self) -> str:
41
    """Collection title (required)."""
42

43
@property
44
def description(self) -> str:
45
    """Collection description (optional)."""
46

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

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

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

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

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

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

71
@property
72
def _raw(self) -> dict:
73
    """Raw collection information response (parsed JSON)."""
74
```
75

76
### Object Retrieval
77

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

80
```python { .api }
81
def get_objects(self, accept=None, **filter_kwargs) -> dict:
82
    """
83
    Retrieve objects from the collection.
84
    
85
    Parameters:
86
    - accept (str, optional): Media type for Accept header
87
    - **filter_kwargs: Filter parameters (added_after, match[type], match[id], 
88
                       match[version], limit, next for TAXII 2.1; 
89
                       start, per_request for TAXII 2.0)
90
    
91
    Returns:
92
    dict: Response envelope/bundle containing objects
93
    
94
    Raises:
95
    AccessError: If collection doesn't allow reading
96
    """
97

98
def get_object(self, obj_id, accept=None, **filter_kwargs) -> dict:
99
    """
100
    Retrieve a specific object by ID.
101
    
102
    Parameters:
103
    - obj_id (str): STIX object identifier
104
    - accept (str, optional): Media type for Accept header
105
    - **filter_kwargs: Filter parameters (match[version] for version filtering)
106
    
107
    Returns:
108
    dict: Response envelope/bundle containing the object
109
    
110
    Raises:
111
    AccessError: If collection doesn't allow reading
112
    """
113

114
def get_manifest(self, accept=None, **filter_kwargs) -> dict:
115
    """
116
    Retrieve object manifests (metadata without full objects).
117
    
118
    Parameters:
119
    - accept (str, optional): Media type for Accept header
120
    - **filter_kwargs: Filter parameters (same as get_objects)
121
    
122
    Returns:
123
    dict: Response containing object manifests
124
    
125
    Raises:
126
    AccessError: If collection doesn't allow reading
127
    """
128
```
129

130
### Object Management
131

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

134
```python { .api }
135
def add_objects(self, envelope, wait_for_completion=True, poll_interval=1,
136
               timeout=60, accept=None, content_type=None) -> Status:
137
    """
138
    Add objects to the collection.
139
    
140
    Parameters:
141
    - envelope (dict|str|bytes): STIX envelope/bundle containing objects to add
142
    - wait_for_completion (bool): Whether to wait for async completion
143
    - poll_interval (int): Polling interval in seconds for completion
144
    - timeout (int): Maximum wait time in seconds (0 or negative for no limit)
145
    - accept (str, optional): Media type for Accept header
146
    - content_type (str, optional): Media type for Content-Type header
147
    
148
    Returns:
149
    Status: Status object tracking the operation
150
    
151
    Raises:
152
    AccessError: If collection doesn't allow writing
153
    TypeError: If envelope type is not supported
154
    """
155

156
def delete_object(self, obj_id, accept=None, **filter_kwargs) -> dict:
157
    """
158
    Delete a specific object by ID (TAXII 2.1 only).
159
    
160
    Parameters:
161
    - obj_id (str): STIX object identifier to delete
162
    - accept (str, optional): Media type for Accept header
163
    - **filter_kwargs: Filter parameters (match[version] for version filtering)
164
    
165
    Returns:
166
    dict: Response confirming deletion
167
    
168
    Raises:
169
    AccessError: If collection doesn't allow writing
170
    """
171

172
def object_versions(self, obj_id, accept=None, **filter_kwargs) -> dict:
173
    """
174
    Get all versions of a specific object (TAXII 2.1 only).
175
    
176
    Parameters:
177
    - obj_id (str): STIX object identifier
178
    - accept (str, optional): Media type for Accept header
179
    - **filter_kwargs: Filter parameters for version filtering
180
    
181
    Returns:
182
    dict: Response containing object versions
183
    
184
    Raises:
185
    AccessError: If collection doesn't allow reading
186
    """
187
```
188

189
### Collection Operations
190

191
Refresh collection information and manage the collection connection.
192

193
```python { .api }
194
def refresh(self, accept=None) -> None:
195
    """
196
    Update collection information.
197
    
198
    Parameters:
199
    - accept (str, optional): Media type for Accept header
200
    """
201

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

205
def __enter__(self):
206
    """Context manager entry."""
207

208
def __exit__(self, exc_type, exc_val, exc_tb):
209
    """Context manager exit."""
210
```
211

212
## Usage Examples
213

214
### Basic Object Retrieval
215

216
```python
217
from taxii2client import Server
218

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

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

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

234
# Print first object
235
if objects:
236
    obj = objects[0]
237
    print(f"First object: {obj.get('type')} - {obj.get('id')}")
238
```
239

240
### Filtered Object Retrieval
241

242
```python
243
from datetime import datetime, timezone
244

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

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

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

256
# Filter by specific object IDs
257
specific_ids = [
258
    "indicator--12345678-1234-5678-9012-123456789012",
259
    "malware--87654321-4321-8765-2109-876543210987"
260
]
261
specific_objects = collection.get_objects(id=specific_ids)
262
```
263

264
### Paginated Retrieval
265

266
```python
267
from taxii2client import as_pages
268

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

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

283
### Object Manifest Retrieval
284

285
```python
286
# Get manifests instead of full objects (lighter weight)
287
manifest_response = collection.get_manifest()
288
manifests = manifest_response.get('objects', [])
289

290
print(f"Collection contains {len(manifests)} objects:")
291
for manifest in manifests:
292
    print(f"  {manifest.get('id')}")
293
    print(f"    Versions: {manifest.get('versions', [])}")
294
    print(f"    Media Types: {manifest.get('media_types', [])}")
295
```
296

297
### Adding Objects
298

299
```python
300
# Create STIX envelope/bundle
301
stix_envelope = {
302
    "objects": [
303
        {
304
            "type": "indicator",
305
            "id": "indicator--12345678-1234-5678-9012-123456789012",
306
            "created": "2023-01-01T00:00:00.000Z",
307
            "modified": "2023-01-01T00:00:00.000Z",
308
            "pattern": "[file:hashes.MD5 = 'd41d8cd98f00b204e9800998ecf8427e']",
309
            "labels": ["malicious-activity"],
310
            "spec_version": "2.1"
311
        },
312
        {
313
            "type": "malware",
314
            "id": "malware--87654321-4321-8765-2109-876543210987",
315
            "created": "2023-01-01T00:00:00.000Z",
316
            "modified": "2023-01-01T00:00:00.000Z",
317
            "name": "BadStuff",
318
            "labels": ["trojan"],
319
            "spec_version": "2.1"
320
        }
321
    ]
322
}
323

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

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

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

341
# Poll manually for completion
342
import time
343
while status.status != "complete":
344
    time.sleep(1)
345
    status.refresh()
346
    print(f"Status: {status.status} ({status.success_count}/{status.total_count})")
347
```
348

349
### TAXII 2.1 Specific Operations
350

351
```python
352
# These operations only work with TAXII 2.1 collections
353
from taxii2client.v21 import Collection
354

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

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

367
# Delete an object
368
delete_response = collection.delete_object(obj_id)
369
print(f"Delete response: {delete_response}")
370
```
371

372
### Error Handling
373

374
```python
375
from taxii2client.exceptions import AccessError, ValidationError, TAXIIServiceException
376

377
try:
378
    # Attempt to read from collection
379
    objects = collection.get_objects()
380
except AccessError as e:
381
    print(f"Access denied: {e}")
382
except ValidationError as e:
383
    print(f"Validation error: {e}")
384
except TAXIIServiceException as e:
385
    print(f"TAXII service error: {e}")
386
except Exception as e:
387
    print(f"Unexpected error: {e}")
388

389
try:
390
    # Attempt to add objects
391
    status = collection.add_objects(malformed_envelope)
392
except TypeError as e:
393
    print(f"Invalid envelope format: {e}")
394
except AccessError as e:
395
    print(f"Write access denied: {e}")
396
```
397

398
### Context Manager Usage
399

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