### Install libtaxii using setup.py
Source: https://libtaxii.readthedocs.io/en/stable/installation.html
Navigate to the extracted directory and run the installation script. This is for manual installation.
```bash
$ cd libtaxii-1.1.119
$ python setup.py install
```
--------------------------------
### Discovery Client Example
Source: https://libtaxii.readthedocs.io/en/stable/scripts.html
Example of how to call the discovery_client with various common command-line arguments.
```bash
discovery_client --url http://hailataxii.com/taxii-discovery-service --cert MyCert.crt --key MyKey.key --username foo --pass bar --proxy http://myproxy.example.com:80 --xml-output
```
--------------------------------
### Verify libtaxii installation
Source: https://libtaxii.readthedocs.io/en/stable/installation.html
Start a Python interpreter and attempt to import the libtaxii library to confirm successful installation.
```python
Python 2.7.6 (default, Mar 22 2014, 22:59:56)
[GCC 4.8.2] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import libtaxii
>>>
```
--------------------------------
### Inbox Client Example
Source: https://libtaxii.readthedocs.io/en/stable/scripts.html
Example of calling the inbox_client with content binding and content file specified.
```bash
inbox_client --content-binding urn:stix.mitre.org:xml:1.1 --content-file stix_file.xml
```
--------------------------------
### DefaultQueryInfo Example
Source: https://libtaxii.readthedocs.io/en/stable/api/query.html
Example demonstrating the creation of a DefaultQueryInfo object using TargetingExpressionInfo objects.
```APIDOC
## DefaultQueryInfo Example
### Description
Example demonstrating the creation of a DefaultQueryInfo object using TargetingExpressionInfo objects.
### Code Example
```python
import libtaxii.taxii_default_query as tdq
from libtaxii.taxii_default_query import TargetingExpressionInfo
from libtaxii.constants import *
# This example has no preferred scope, and allows any scope
tei_01 = TargetingExpressionInfo(
targeting_expression_id=CB_STIX_XML_111,
preferred_scope=[],
allowed_scope=['**'])
# This example prefers the Indicator scope and allows no other scope
tei_02 = TargetingExpressionInfo(
targeting_expression_id=CB_STIX_XML_111,
preferred_scope=['STIX_Package/Indicators/Indicator/**'],
allowed_scope=[])
# DefaultQueryInfo describes the TAXII Default Queries that are supported
tdqi1 = tdq.DefaultQueryInfo(
targeting_expression_infos=[tei_01, tei_02],
capability_modules=[CM_CORE])
```
```
--------------------------------
### Install libtaxii using pip
Source: https://libtaxii.readthedocs.io/en/stable/installation.html
Use this command to install the libtaxii library with pip. Ensure you have pip installed.
```bash
$ pip install libtaxii
```
--------------------------------
### Query Client Command Line Example
Source: https://libtaxii.readthedocs.io/en/stable/scripts.html
Example usage of the Query Client with a specified collection and IP address.
```bash
query_client --collection MyQueryCollection --ip 10.0.0.0
```
--------------------------------
### TargetingExpressionInfo Example
Source: https://libtaxii.readthedocs.io/en/stable/api/query.html
Examples demonstrating the creation of TargetingExpressionInfo objects with different preferred and allowed scopes.
```APIDOC
## TargetingExpressionInfo Examples
### Description
Examples demonstrating the creation of TargetingExpressionInfo objects with different preferred and allowed scopes.
### Code Example
```python
import libtaxii.taxii_default_query as tdq
from libtaxii.taxii_default_query import TargetingExpressionInfo
from libtaxii.constants import *
# This example has no preferred scope, and allows any scope
tei_01 = TargetingExpressionInfo(
targeting_expression_id=CB_STIX_XML_111,
preferred_scope=[],
allowed_scope=['**'])
# This example prefers the Indicator scope and allows no other scope
tei_02 = TargetingExpressionInfo(
targeting_expression_id=CB_STIX_XML_111,
preferred_scope=['STIX_Package/Indicators/Indicator/**'],
allowed_scope=[])
```
```
--------------------------------
### Poll Client Example
Source: https://libtaxii.readthedocs.io/en/stable/scripts.html
Example of calling the poll_client to request data from a specific collection.
```bash
poll_client --collection MyCollection
```
--------------------------------
### Poll Client 1.0 Command Line Example
Source: https://libtaxii.readthedocs.io/en/stable/scripts.html
Example usage of the Poll Client 1.0 with specified feed and subscription ID.
```bash
poll_client_10 --feed MyFeedName --subscription-id SomeSubscriptionId
```
--------------------------------
### PollRequest Example
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Example demonstrating how to construct a PollRequest object with PollParameters, including delivery parameters and content bindings.
```python
import datetime
from libtaxii.messages_11 import PollRequest, PollParameters, DeliveryParameters, ContentBinding
from libtaxii.constants import VID_TAXII_HTTPS_10, VID_TAXII_XML_11, CB_STIX_XML_11, RT_COUNT_ONLY
from dateutil.tz import tzutc
delivery_parameters1 = DeliveryParameters(
inbox_protocol=VID_TAXII_HTTPS_10,
inbox_address='https://example.com/inboxAddress/',
delivery_message_binding=VID_TAXII_XML_11)
poll_params1 = PollParameters(
allow_asynch=False,
response_type=RT_COUNT_ONLY,
content_bindings=[ContentBinding(binding_id=CB_STIX_XML_11)],
#query=query1,
delivery_parameters=delivery_parameters1)
poll_req3 = PollRequest(
message_id='PollReq03',
collection_name='collection100',
exclusive_begin_timestamp_label=datetime.datetime.now(tzutc()),
inclusive_end_timestamp_label=datetime.datetime.now(tzutc()),
poll_parameters=poll_params1)
```
--------------------------------
### Poll Fulfillment Client Example
Source: https://libtaxii.readthedocs.io/en/stable/scripts.html
Example of calling the fulfillment_client with collection, result ID, and result part number.
```bash
fulfillment_client --collection MyCollection --result_id someId --result_part_number 1
```
--------------------------------
### ContentBlock Example
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Demonstrates how to create a ContentBlock instance with content binding, content data, timestamp, padding, and an associated message.
```python
cb001 = tm11.ContentBlock(
content_binding=tm11.ContentBinding(CB_STIX_XML_11),
content='',
timestamp_label=datetime.datetime.now(tzutc()),
message='Hullo!',
padding='The quick brown fox jumped over the lazy dogs.')
```
--------------------------------
### Example Usage of TAXII HttpClient
Source: https://libtaxii.readthedocs.io/en/stable/getting_started.html
Demonstrates how to instantiate and use the HttpClient to make a TAXII discovery request. This client supports HTTP Basic and TLS Certificate authentication.
```python
import libtaxii as t
import libtaxii.clients as tc
import libtaxii.messages_11 as tm11
from libtaxii.constants import *
client = tc.HttpClient()
client.set_auth_type(tc.HttpClient.AUTH_BASIC)
client.set_use_https(True)
client.set_auth_credentials({'username': 'MyUsername', 'password': 'MyPassword'})
discovery_request = tm11.DiscoveryRequest(tm11.generate_message_id())
discovery_xml = discovery_request.to_xml()
http_resp = client.call_taxii_service2('example.com', '/pollservice/', VID_TAXII_XML_11, discovery_xml)
taxii_message = t.get_message_from_http_response(http_resp, discovery_request.message_id)
print taxii_message.to_xml()
```
--------------------------------
### Create a Manage Collection Subscription Request
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Example of creating a Manage Collection Subscription Request message to subscribe to a collection. Requires setting up subscription and push parameters.
```python
subscription_parameters1 = tm11.SubscriptionParameters()
push_parameters1 = tm11.PushParameters("", "", "")
subs_req1 = tm11.ManageCollectionSubscriptionRequest(
message_id='SubsReq01',
action=ACT_SUBSCRIBE,
collection_name='collection1',
subscription_parameters=subscription_parameters1,
push_parameters=push_parameters1)
```
--------------------------------
### Create a Manage Collection Subscription Response
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Example of constructing a Manage Collection Subscription Response message. This includes details about subscription instances and poll instances.
```python
subscription_parameters1 = tm11.SubscriptionParameters()
push_parameters1 = tm11.PushParameters("", "", "")
poll_instance1 = tm11.PollInstance(
poll_protocol=VID_TAXII_HTTPS_10,
poll_address='https://example.com/poll1/',
poll_message_bindings=[VID_TAXII_XML_11])
subs1 = tm11.SubscriptionInstance(
subscription_id='Subs001',
status=SS_ACTIVE,
subscription_parameters=subscription_parameters1,
push_parameters=push_parameters1,
poll_instances=[poll_instance1])
subs_resp1 = tm11.ManageCollectionSubscriptionResponse(
message_id='SubsResp01',
in_response_to='xyz',
collection_name='abc123',
message='Hullo!',
subscription_instances=[subs1])
```
--------------------------------
### TAXII 1.1 Discovery Response
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Discovery Response message in TAXII 1.1.
```python
from libtaxii.messages_11 import DiscoveryResponse, DiscoveryResponseServiceInstance
service_instance = DiscoveryResponseServiceInstance(
service_name='TAXII Service',
service_version='1.0',
service_type='COLLECTION_MANAGEMENT',
message_need_to_process=True,
path='https://example.com/taxii/services/collection/1.0'
)
discovery_response = DiscoveryResponse(service_instances=[service_instance])
print(discovery_response.to_xml())
```
--------------------------------
### TAXII 1.1 Discovery Request
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Discovery Request message in TAXII 1.1.
```python
from libtaxii.messages_11 import DiscoveryRequest
discovery_request = DiscoveryRequest()
print(discovery_request.to_xml())
```
--------------------------------
### Extract tar.gz archive
Source: https://libtaxii.readthedocs.io/en/stable/installation.html
Extract the downloaded libtaxii tar.gz archive. This is part of the manual installation process.
```bash
$ tar -zxf libtaxii-1.1.119.tar.gz
$ ls
libtaxii-1.1.119 libtaxii-1.1.119.tar.gz
```
--------------------------------
### HttpClient with TLS Certificate Authentication
Source: https://libtaxii.readthedocs.io/en/stable/api/clients.html
Initializes an HttpClient with TLS Certificate authentication, using HTTPS, and demonstrates calling a TAXII discovery service. Note: This example requires valid certificate files and a server that supports TLS authentication.
```python
client = tc.HttpClient()
client.set_use_https(True)
client.set_auth_type(tc.HttpClient.AUTH_CERT)
client.set_auth_credentials({'key_file': '../PATH_TO_KEY_FILE.key', 'cert_file': '../PATH_TO_CERT_FILE.crt'})
discovery_request = tm11.DiscoveryRequest(generate_message_id())
discovery_xml = discovery_request.to_xml(pretty_print=True)
http_resp = client.call_taxii_service2('hailataxii.com', '/taxii-discovery-service/', VID_TAXII_XML_11, discovery_xml)
taxii_message = t.get_message_from_http_response(http_resp, discovery_request.message_id)
print taxii_message.to_xml(pretty_print=True)
```
--------------------------------
### Python Example for Collection Information Response
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Demonstrates how to construct a TAXII Collection Information Response object in Python, including defining push methods, polling services, subscription methods, and receiving inbox services.
```python
push_method1 = tm11.PushMethod(
push_protocol=VID_TAXII_HTTP_10,
push_message_bindings=[VID_TAXII_XML_11])
poll_service1 = tm11.PollingServiceInstance(
poll_protocol=VID_TAXII_HTTPS_10,
poll_address='https://example.com/PollService1',
poll_message_bindings=[VID_TAXII_XML_11])
poll_service2 = tm11.PollingServiceInstance(
poll_protocol=VID_TAXII_HTTPS_10,
poll_address='https://example.com/PollService2',
poll_message_bindings=[VID_TAXII_XML_11])
subs_method1 = tm11.SubscriptionMethod(
subscription_protocol=VID_TAXII_HTTPS_10,
subscription_address='https://example.com/SubscriptionService',
subscription_message_bindings=[VID_TAXII_XML_11])
inbox_service1 = tm11.ReceivingInboxService(
inbox_protocol=VID_TAXII_HTTPS_10,
inbox_address='https://example.com/InboxService',
inbox_message_bindings=[VID_TAXII_XML_11],
supported_contents=None)
collection1 = tm11.CollectionInformation(
collection_name='collection1',
collection_description='This is a collection',
supported_contents=[tm11.ContentBinding(CB_STIX_XML_101)],
available=False,
push_methods=[push_method1],
polling_service_instances=[poll_service1, poll_service2],
subscription_methods=[subs_method1],
collection_volume=4,
collection_type=CT_DATA_FEED,
receiving_inbox_services=[inbox_service1])
collection_response1 = tm11.CollectionInformationResponse(
message_id='CIR01',
in_response_to='0',
collection_informations=[collection1])
```
--------------------------------
### Get XML Parser
Source: https://libtaxii.readthedocs.io/en/stable/api/common.html
Retrieves the currently configured XML parser. If none is set, it initializes a new etree.XMLParser with specific network and tree size configurations.
```python
libtaxii.common.get_xml_parser()
```
--------------------------------
### Instantiate PollResponse Message
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Example of creating a PollResponse message with content blocks, record count, and other optional parameters. This is useful for constructing responses to poll requests.
```python
cb1 = tm11.ContentBlock(CB_STIX_XML_11, "")
cb2 = tm11.ContentBlock(CB_STIX_XML_11, "")
count = tm11.RecordCount(record_count=22, partial_count=False)
poll_resp1 = tm11.PollResponse(
message_id='PollResp1',
in_response_to='tmp',
collection_name='blah',
exclusive_begin_timestamp_label=datetime.datetime.now(tzutc()),
inclusive_end_timestamp_label=datetime.datetime.now(tzutc()),
subscription_id='24',
message='This is a test message',
content_blocks=[cb1, cb2],
more=True,
result_id='123',
result_part_number=1,
record_count=count)
```
--------------------------------
### TAXII 1.1 Collection Management Request
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Collection Management Request message in TAXII 1.1.
```python
from libtaxii.messages_11 import CollectionManagementRequest
collection_management_request = CollectionManagementRequest()
print(collection_management_request.to_xml())
```
--------------------------------
### HttpClient Initialization and No Authentication
Source: https://libtaxii.readthedocs.io/en/stable/api/clients.html
Initializes an HttpClient with no authentication and demonstrates calling a TAXII discovery service.
```python
client = tc.HttpClient()
client.set_auth_type(tc.HttpClient.AUTH_NONE)
client.set_use_https(False)
discovery_request = tm11.DiscoveryRequest(generate_message_id())
discovery_xml = discovery_request.to_xml(pretty_print=True)
http_resp = client.call_taxii_service2('hailataxii.com', '/taxii-discovery-service', VID_TAXII_XML_11, discovery_xml)
taxii_message = t.get_message_from_http_response(http_resp, discovery_request.message_id)
print taxii_message.to_xml(pretty_print=True)
```
--------------------------------
### SubscriptionInformation Subscription ID Getter
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Gets the subscription ID for SubscriptionInformation.
```python
@property
def subscription_id(self):
return self._subscription_id
```
--------------------------------
### HttpClient with HTTP Basic Authentication
Source: https://libtaxii.readthedocs.io/en/stable/api/clients.html
Initializes an HttpClient with HTTP Basic authentication and demonstrates calling a TAXII discovery service.
```python
client = tc.HttpClient()
client.set_auth_type(tc.HttpClient.AUTH_BASIC)
client.set_auth_credentials({'username': 'guest', 'password': 'guest'})
discovery_request = tm11.DiscoveryRequest(generate_message_id())
discovery_xml = discovery_request.to_xml(pretty_print=True)
http_resp = client.call_taxii_service2('hailataxii.com', '/taxii-discovery-service', VID_TAXII_XML_11, discovery_xml)
taxii_message = t.get_message_from_http_response(http_resp, discovery_request.message_id)
print taxii_message.to_xml(pretty_print=True)
```
--------------------------------
### TAXII 1.1 Status Message
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Status Message in TAXII 1.1.
```python
from libtaxii.messages_11 import StatusMessage, StatusDetail
status_detail = StatusDetail(code=200, message='Operation successful')
status_message = StatusMessage(status_type='SUCCESS', status_details=[status_detail])
print(status_message.to_xml())
```
--------------------------------
### TAXII 1.1 Poll Response
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Poll Response message in TAXII 1.1.
```python
from libtaxii.messages_11 import PollResponse, ContentBlock, DiscoveryResponseServiceInstance
content_block = ContentBlock(
content_binding_id='urn:taxii.mitre.org:granularity:1.0',
content='Sample content'
)
poll_response = PollResponse(
collection_name='my_collection',
feed_name='my_feed',
version='1.1',
content_blocks=[content_block]
)
print(poll_response.to_xml())
```
--------------------------------
### TAXII 1.1 Poll Request
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Poll Request message in TAXII 1.1.
```python
from libtaxii.messages_11 import PollRequest
poll_request = PollRequest(
collection_name='my_collection',
feed_name='my_feed',
start_date='2023-01-01T00:00:00Z',
end_date='2023-01-31T23:59:59Z',
version='1.1'
)
print(poll_request.to_xml())
```
--------------------------------
### TAXII 1.1 GetCollections Response
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a GetCollections Response message in TAXII 1.1.
```python
from libtaxii.messages_11 import GetCollectionsResponse, CollectionDescription
collection_description = CollectionDescription(
collection_name='my_collection',
display_name='My Collection',
description='A sample collection.',
is_exclusive=False,
is_full_archive=False,
is_limited=False,
max_content_blocks_per_message=10,
max_events_per_message=100,
preferred_content_types=['application/xml']
)
get_collections_response = GetCollectionsResponse(collections=[collection_description])
print(get_collections_response.to_xml())
```
--------------------------------
### SubscriptionMethod Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_10.html
Initializes a SubscriptionMethod object with protocol, address, and message bindings. These parameters are required for defining a subscription service.
```python
def __init__(self, subscription_protocol, subscription_address,
subscription_message_bindings):
self.subscription_protocol = subscription_protocol
self.subscription_address = subscription_address
self.subscription_message_bindings = subscription_message_bindings
```
--------------------------------
### TAXII 1.1 GetCollections Request
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a GetCollections Request message in TAXII 1.1.
```python
from libtaxii.messages_11 import GetCollectionsRequest
get_collections_request = GetCollectionsRequest()
print(get_collections_request.to_xml())
```
--------------------------------
### Creating ContentBlock Instance
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_10.html
Initializes a ContentBlock object. The `content_binding` is a URI, and `content` can be a string or an etree element. `timestamp_label` and `padding` are optional.
```python
from libtaxii.messages_10 import ContentBlock
# Example with string content
content_block_str = ContentBlock(content_binding='urn:taxii.mitre.org:Content:1.0', content='This is some content.')
# Example with XML content (assuming etree_element is an lxml etree element)
# from lxml import etree
# etree_element = etree.Element('MyContent')
# content_block_xml = ContentBlock(content_binding='urn:taxii.mitre.org:Content:1.1', content=etree_element)
```
--------------------------------
### ManageFeedSubscriptionRequest Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_10.html
Initializes a ManageFeedSubscriptionRequest object. Use this to create a new request for managing feed subscriptions.
```python
def __init__(self, message_id, extended_headers=None,
feed_name=None, action=None, subscription_id=None,
delivery_parameters=None):
super(ManageFeedSubscriptionRequest, self).__init__(message_id, extended_headers=extended_headers)
self.feed_name = feed_name
self.action = action
self.subscription_id = subscription_id
self.delivery_parameters = delivery_parameters
```
--------------------------------
### TAXII 1.1 Message Not Handled Exception
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Message Not Handled Exception in TAXII 1.1.
```python
from libtaxii.messages_11 import MessageNotHandledException
exception = MessageNotHandledException(message='Message type not supported')
print(exception.to_xml())
```
--------------------------------
### Constructing a TAXII 1.0 DiscoveryResponse
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_10.html
Demonstrates how to create a DiscoveryResponse message, including adding ServiceInstance details. Service instances can be appended to an existing response or provided during initialization.
```python
discovery_response = tm10.DiscoveryResponse(
message_id=tm10.generate_message_id(),
in_response_to=discovery_request.message_id)
service_instance = tm10.ServiceInstance(
service_type=SVC_INBOX,
services_version=VID_TAXII_SERVICES_10,
protocol_binding=VID_TAXII_HTTPS_10,
service_address='https://example.com/inbox/',
message_bindings=[VID_TAXII_XML_10],
inbox_service_accepted_content=[CB_STIX_XML_10],
available=True,
message='This is a sample inbox service instance')
discovery_response.service_instances.append(service_instance)
# Alternatively, you could define the service instance(s) first and use the
# following:
service_instance_list = [service_instance]
discovery_response = tm10.DiscoveryResponse(
message_id=tm10.generate_message_id(),
in_response_to=discovery_request.message_id,
service_instances=service_instance_list)
```
--------------------------------
### Extract zip archive
Source: https://libtaxii.readthedocs.io/en/stable/installation.html
Extract the downloaded libtaxii zip archive. This is an alternative to using tar.gz for manual installation.
```bash
$ unzip libtaxii-1.1.119.zip
$ ls
libtaxii-1.1.119 libtaxii-1.1.119.zip
```
--------------------------------
### Describing TAXII Default Query Capabilities with DefaultQueryInfo
Source: https://libtaxii.readthedocs.io/en/stable/api/query.html
Illustrates the creation of a DefaultQueryInfo object, which aggregates TargetingExpressionInfo objects and capability modules to describe the default query capabilities supported by a TAXII service.
```python
import libtaxii.taxii_default_query as tdq
from libtaxii.constants import CM_CORE
# Assuming tei_01 and tei_02 are defined as above
tdqi1 = tdq.DefaultQueryInfo(
targeting_expression_infos=[tei_01, tei_02],
capability_modules=[CM_CORE])
```
--------------------------------
### Constructing TAXII Default Queries with Criteria
Source: https://libtaxii.readthedocs.io/en/stable/api/query.html
Demonstrates how to build Criteria objects with AND/OR operators and nested criteria, which are then used to construct DefaultQuery objects for TAXII requests.
```python
from libtaxii.constants import OP_AND, OP_OR, CB_STIX_XML_111
import libtaxii.taxii_default_query as tdq
criteria1 = tdq.Criteria(operator=OP_AND,
criterion=[criterion1])
criteria2 = tdq.Criteria(operator=OP_OR,
criterion=[criterion1, criterion2, criterion3])
criteria3 = tdq.Criteria(operator=OP_AND,
criterion=[criterion1, criterion3],
criteria=[criteria2])
query1 = tdq.DefaultQuery(targeting_expression_id=CB_STIX_XML_111,
criteria=criteria1)
query2 = tdq.DefaultQuery(targeting_expression_id=CB_STIX_XML_111,
criteria=criteria3)
query3 = tdq.DefaultQuery(targeting_expression_id=CB_STIX_XML_111,
criteria=criteria2)
```
--------------------------------
### TAXII 1.1 Feed Management Response
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Feed Management Response message in TAXII 1.1.
```python
from libtaxii.messages_11 import FeedManagementResponse, FeedDescription
feed_description = FeedDescription(
feed_name='my_feed',
display_name='My Feed',
description='A sample feed.',
collection_names=['my_collection']
)
feed_management_response = FeedManagementResponse(feeds=[feed_description])
print(feed_management_response.to_xml())
```
--------------------------------
### PushMethod from_dict() Method
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_10.html
Creates a PushMethod object from a dictionary. The dictionary keys must match the PushMethod constructor arguments.
```python
@staticmethod
def from_dict(d):
return PushMethod(**d)
```
--------------------------------
### TAXII 1.1 Feed Management Request
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Feed Management Request message in TAXII 1.1.
```python
from libtaxii.messages_11 import FeedManagementRequest
feed_management_request = FeedManagementRequest()
print(feed_management_request.to_xml())
```
--------------------------------
### TAXII 1.1 Collection Management Response
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Example of creating a Collection Management Response message in TAXII 1.1.
```python
from libtaxii.messages_11 import CollectionManagementResponse, CollectionDescription
collection_description = CollectionDescription(
collection_name='my_collection',
display_name='My Collection',
description='A sample collection.',
is_exclusive=False,
is_full_archive=False,
is_limited=False,
max_content_blocks_per_message=10,
max_events_per_message=100,
preferred_content_types=['application/xml']
)
collection_management_response = CollectionManagementResponse(collections=[collection_description])
print(collection_management_response.to_xml())
```
--------------------------------
### ManageCollectionSubscriptionRequest Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Initializes a ManageCollectionSubscriptionRequest object with essential parameters like message ID, collection name, and action. Optional parameters for subscription and push details can also be provided.
```python
def __init__(self, message_id, collection_name, action, subscription_id=None,
subscription_parameters=None, push_parameters=None):
super(ManageCollectionSubscriptionRequest, self).__init__(message_id, extended_headers=extended_headers)
self.collection_name = collection_name
self.action = action
self.subscription_id = subscription_id
self.subscription_parameters = subscription_parameters or SubscriptionParameters()
self.push_parameters = push_parameters
```
--------------------------------
### Create and Populate DiscoveryResponse Message
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Demonstrates how to create a DiscoveryResponse message and add ServiceInstance objects to it. Use this when constructing a response to a DiscoveryRequest.
```python
discovery_response = tm11.DiscoveryResponse(
message_id=tm11.generate_message_id(),
in_response_to=discovery_request.message_id)
service_instance = tm11.ServiceInstance(
service_type=SVC_POLL,
services_version=VID_TAXII_SERVICES_11,
protocol_binding=VID_TAXII_HTTP_10,
service_address='http://example.com/poll/',
message_bindings=[VID_TAXII_XML_11],
available=True,
message='This is a message.',
#supported_query=[tdq1],
)
discovery_response.service_instances.append(service_instance)
```
```python
service_instance_list = [service_instance]
discovery_response = tm11.DiscoveryResponse(
message_id=tm11.generate_message_id(),
in_response_to=discovery_request.message_id,
service_instances=service_instance_list)
```
--------------------------------
### Get Required XML Element
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/common.html
Retrieves a required XML element using XPath. Raises a ValueError if the element is not found.
```python
def get_required(etree_xml, xpath, ns_map):
elements = etree_xml.xpath(xpath, namespaces=ns_map)
if len(elements) == 0:
raise ValueError('Element "%s" is required' % xpath)
return elements[0]
```
--------------------------------
### ServiceInstance Class Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Initializes a ServiceInstance object with required and optional parameters. The `message_bindings` list must contain at least one value, and `supported_query` is optional for POLL services.
```python
def __init__(self, service_type, services_version, protocol_binding,
service_address, message_bindings,
inbox_service_accepted_content=None, available=None,
message=None, supported_query=None):
self.service_type = service_type
self.services_version = services_version
self.protocol_binding = protocol_binding
self.service_address = service_address
self.message_bindings = message_bindings
self.inbox_service_accepted_content = inbox_service_accepted_content or []
self.available = available
self.message = message
self.supported_query = supported_query or []
```
--------------------------------
### Get TAXII Message from Dictionary
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_10.html
Creates a TAXII Message object from a dictionary. The message type is determined by the 'message_type' key within the dictionary.
```python
message_dict = message.to_dict()
new_message = tm10.get_message_from_dict(message_dict)
```
--------------------------------
### SubscriptionInformation Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_10.html
Initializes a SubscriptionInformation object with feed name, subscription ID, and timestamp labels. Requires feed_name and subscription_id.
```python
def __init__(self, feed_name, subscription_id,
inclusive_begin_timestamp_label,
inclusive_end_timestamp_label):
self.feed_name = feed_name
self.subscription_id = subscription_id
self.inclusive_begin_timestamp_label = inclusive_begin_timestamp_label
self.inclusive_end_timestamp_label = inclusive_end_timestamp_label
```
--------------------------------
### Get Optional XML Element
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/common.html
Retrieves an optional XML element using XPath. Returns None if the element is not found, suppressing ValueError.
```python
def get_optional(etree_xml, xpath, ns_map):
try:
return get_required(etree_xml, xpath, ns_map)
except ValueError:
pass
```
--------------------------------
### Creating TAXII 1.0 Feed Information Components
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_10.html
Demonstrates how to instantiate PushMethod, PollingServiceInstance, and SubscriptionMethod objects for use within a FeedInformation response. These objects define the communication protocols and addresses for push, polling, and subscription services.
```python
push_method1 = tm10.PushMethod(
push_protocol=VID_TAXII_HTTP_10,
push_message_bindings=[VID_TAXII_XML_10])
poll_service1 = tm10.PollingServiceInstance(
poll_protocol=VID_TAXII_HTTP_10,
poll_address='http://example.com/PollService/',
poll_message_bindings=[VID_TAXII_XML_10])
subscription_service1 = tm10.SubscriptionMethod(
subscription_protocol=VID_TAXII_HTTP_10,
subscription_address='http://example.com/SubsService/',
subscription_message_bindings=[VID_TAXII_XML_10])
feed1 = tm10.FeedInformation(
feed_name='Feed1',
feed_description='Description of a feed',
supported_contents=[CB_STIX_XML_10],
available=True,
push_methods=[push_method1]
```
--------------------------------
### DefaultQueryInfo Class Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/taxii_default_query.html
Initializes a `DefaultQueryInfo` object, which describes the TAXII Default Queries supported by a service. It requires a list of supported targeting expression infos and a list of capability module identifiers.
```python
class DefaultQueryInfo(tm11.SupportedQuery):
""" Used to describe the TAXII Default Queries that are supported.
:param targeting_expression_infos: Describe the supported targeting expressions
:type targeting_expression_infos: :class:`list` of :class:`TargetingExpressionInfo` objects
:param capability_modules: Indicate the supported capability modules
:type capability_modules: :class:`list` of :class:`str`
"""
def __init__(self, targeting_expression_infos, capability_modules):
super(DefaultQueryInfo, self).__init__(FID_TAXII_DEFAULT_QUERY_10)
self.targeting_expression_infos = targeting_expression_infos
self.capability_modules = capability_modules
```
--------------------------------
### Create InboxMessage with Subscription Information
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_11.html
Demonstrates how to construct an InboxMessage including content blocks, subscription information, record count, and destination collections. Use this when sending content in response to a subscription.
```python
cb1 = tm11.ContentBlock(CB_STIX_XML_11, "")
cb2 = tm11.ContentBlock(CB_STIX_XML_11, "")
subs_info1 = tm11.SubscriptionInformation(
collection_name='SomeCollectionName',
subscription_id='SubsId021',
exclusive_begin_timestamp_label=datetime.datetime.now(tzutc()),
inclusive_end_timestamp_label=datetime.datetime.now(tzutc()))
inbox1 = tm11.InboxMessage(
message_id='Inbox1',
result_id='123',
destination_collection_names=['collection1','collection2'],
message='Hello!',
subscription_information=subs_info1,
record_count=tm11.RecordCount(22, partial_count=True),
content_blocks=[cb1, cb2])
```
--------------------------------
### Getting a Query Deserializer
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Retrieves a registered deserializer for a given query format ID and type ('query' or 'query_info'). Raises an exception if the format ID is not registered.
```Python
def get_deserializer(format_id, type):
do_check(type, 'type', value_tuple=('query', 'query_info'))
if format_id not in query_deserializers:
raise UnsupportedQueryException('A deserializer for the query format \'%s\' is not registered.' % format_id)
return query_deserializers[format_id][type]
```
--------------------------------
### LIBTAXII URL Construction
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/clients.html
This code constructs the final URL for a TAXII request, including the scheme (HTTP/HTTPS), host, port, path, and optional GET parameters.
```python
opener = urllib.request.build_opener(*handler_list)
urllib.request.install_opener(opener)
if port is None: # If the caller did not specify a port, use the default
if self.use_https:
port = 443
else:
port = 80
if self.use_https:
scheme = 'https://'
else:
scheme = 'http://'
url = scheme + host + ':' + str(port) + path
if get_params_dict is not None:
url += '?' + urllib.parse.urlencode(get_params_dict)
```
--------------------------------
### Module Header
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/clients.html
This is the header for the libtaxii.clients module, including copyright information.
```python
# Copyright (c) 2017, The MITRE Corporation
```
--------------------------------
### Get Optional XML Element Text
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/common.html
Retrieves the text content of an optional XML element using XPath. Returns None if the element is not found or has no text.
```python
def get_optional_text(etree_xml, xpath, ns_map):
try:
return get_required(etree_xml, xpath, ns_map).text
except ValueError:
pass
```
--------------------------------
### Create Subscription Method from Dictionary
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_10.html
Instantiates a Subscription Method object directly from a dictionary. Assumes dictionary keys match constructor arguments.
```python
@classmethod
def from_dict(cls, d):
return cls(**d)
```
--------------------------------
### SubscriptionInformation Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Initializes a SubscriptionInformation object with collection name, subscription ID, and optional timestamp labels for exclusive begin and inclusive end.
```python
def __init__(self, collection_name, subscription_id, exclusive_begin_timestamp_label=None, inclusive_end_timestamp_label=None):
self.collection_name = collection_name
self.subscription_id = subscription_id
self.exclusive_begin_timestamp_label = exclusive_begin_timestamp_label
self.inclusive_end_timestamp_label = inclusive_end_timestamp_label
```
--------------------------------
### TAXII Default Query Tests
Source: https://libtaxii.readthedocs.io/en/stable/api/query.html
Demonstrates how to create different types of tests for TAXII Default Queries, including 'equals', 'matches' (regex), and 'greater_than' (timestamp) relationships. Ensure the appropriate capability ID and parameters are used for each test.
```python
import libtaxii.taxii_default_query as tdq
from libtaxii.taxii_default_query import Test
from libtaxii.taxii_default_query import Criterion
from libtaxii.taxii_default_query import Criteria
from libtaxii.constants import *
import datetime
##############################################################################
# A Taxii Default Query *Test* gives the consumer granular control over the
# Target of a Query by applying unambiguos relationship requirements specified
# using a standardized vocabulary.
# Each Relationship (e.g. equals, greater_than, etc.) in a Capability
# Module defines a set of paramater fields, capable of expressing that
# relation.
# The *equals* relationship, of the Core Capability Module, returns True if
# the target matches the value exactly. If the target merely contains the
# value (but does not match exactly) the relationship Test returns False.
test_equals = Test(capability_id=CM_CORE,
relationship='equals',
parameters={'value': 'Test value',
'match_type': 'case_sensitive_string'})
# The *matches* relationship, in the context of the Regular Expression
# Capability Module, returns true if the target matches the regular expression
# contained in the value.
test_matches = Test(capability_id=CM_REGEX,
relationship='matches',
parameters={'value': '[A-Z]*',
'case_sensitive': True})
# The *greater than* relationship, in the context of the Timestamp Capability
# Module returns True if the target's timestamp indicates a later time than
# that specified by this value. This relationship is only valid for timestamp
# comparisons.
test_timestamp = Test(capability_id=CM_TIMESTAMP,
relationship='greater_than',
parameters={'value': datetime.datetime.now()})
##############################################################################
# A *Criterion* specifies how a Target is evaluated against a Test. Within a
# Criterion, the Target is used to identify a specific region of a record to
# which the Test should be applied. Slash Notation Targeting Expression syntax,
# in conjunction with a Targeting Expression Vocabulary, are used to form a
# Targeting Expression
# A Multi-field Wildcard (**). This indicates any Node or series of Nodes,
# specified by double asterisks.
criterion1 = Criterion(target='**',
test=test_equals)
# Indicates that *id* fields in the STIX Indicator construct are in scope
criterion2 = Criterion(target='STIX_Package/Indicators/Indicator/@id',
test=test_matches)
# Indicates that all STIX Description fields are in scope
criterion3 = Criterion(target='**/Description',
test=test_timestamp)
##############################################################################
# *Criteria* consist of a logical operator (and/or) that should be applied to
```
--------------------------------
### Get TAXII Message from JSON String
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_10.html
Parses a JSON string into a TAXII Message object. The function automatically detects the message type based on the JSON content.
```python
new_message = tm10.get_message_from_json(json_string)
```
--------------------------------
### PushMethod from_etree() Method
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_10.html
Creates a PushMethod object from an ElementTree XML element. Requires the XML to contain 'Protocol_Binding' and 'Message_Binding' tags.
```python
@staticmethod
def from_etree(etree_xml):
kwargs = {}
kwargs['push_protocol'] = get_required(etree_xml, './taxii:Protocol_Binding', ns_map).text
kwargs['push_message_bindings'] = []
for message_binding in etree_xml.xpath('./taxii:Message_Binding', namespaces=ns_map):
kwargs['push_message_bindings'].append(message_binding.text)
return PushMethod(**kwargs)
```
--------------------------------
### Get TAXII Message from XML String
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_10.html
Parses an XML string into a TAXII Message object. The function automatically detects the message type based on the XML content.
```python
message_xml = message.to_xml()
new_message = tm10.get_message_from_xml(message_xml)
```
--------------------------------
### Initialize TAXII 1.1 Subscription Instance
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/messages_11.html
Initializes a Subscription object with an ID, status, and optional parameters. Defaults status to 'ACTIVE'.
```python
def __init__(self, subscription_id, status=SS_ACTIVE,
subscription_parameters=None, push_parameters=None,
poll_instances=None):
self.subscription_id = subscription_id
self.status = status
self.subscription_parameters = subscription_parameters
self.push_parameters = push_parameters
self.poll_instances = poll_instances or []
```
--------------------------------
### Creating a TAXII 1.0 FeedInformationRequest
Source: https://libtaxii.readthedocs.io/en/stable/api/messages_10.html
Shows how to instantiate a FeedInformationRequest message, optionally including extended headers for additional metadata.
```python
ext_headers = {'name1': 'val1', 'name2': 'val2'}
feed_information_request= tm10.FeedInformationRequest(
message_id=tm10.generate_message_id(),
extended_headers=ext_headers)
```
--------------------------------
### Defining Supported Targeting Expressions with TargetingExpressionInfo
Source: https://libtaxii.readthedocs.io/en/stable/api/query.html
Shows how to create TargetingExpressionInfo objects to describe available targeting vocabularies and their permissible query scopes. This is used by DefaultQueryInfo to indicate service capabilities.
```python
from libtaxii.taxii_default_query import TargetingExpressionInfo
from libtaxii.constants import CB_STIX_XML_111
# Example with no preferred scope, allowing any scope
tei_01 = TargetingExpressionInfo(
targeting_expression_id=CB_STIX_XML_111,
preferred_scope=[],
allowed_scope=['**'])
# Example preferring Indicator scope and allowing no other scope
tei_02 = TargetingExpressionInfo(
targeting_expression_id=CB_STIX_XML_111,
preferred_scope=['STIX_Package/Indicators/Indicator/**'],
allowed_scope=[])
```
--------------------------------
### Set Custom XML Parser
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/common.html
Allows setting a custom etree.XMLParser for libtaxii. This is useful for advanced configurations or when integrating with existing XML parsing setups. The provided parser will be used for all subsequent TAXII XML processing.
```python
def set_xml_parser(xml_parser=None):
"""Set the libtaxii.messages XML parser.
Args:
xml_parser (etree.XMLParser): The parser to use to parse TAXII XML.
"""
global _XML_PARSER
_XML_PARSER = xml_parser
```
--------------------------------
### TargetingExpressionInfo Class Initialization
Source: https://libtaxii.readthedocs.io/en/stable/_modules/libtaxii/taxii_default_query.html
Initializes a `TargetingExpressionInfo` object, which describes a supported targeting expression. It includes the expression ID and optional preferred and allowed scopes for queries.
```python
class TargetingExpressionInfo(TAXIIBase):
"""This class describes supported Targeting Expressions
:param string targeting_expression_id: The supported targeting expression ID
:param preferred_scope: Indicates the preferred scope of queries
:type preferred_scope: :class:`list` of :class:`string`
:param allowed_scope: Indicates the allowed scope of queries
:type allowed_scope: :class:`list` of :class:`string`
"""
def __init__(self, targeting_expression_id, preferred_scope=None, allowed_scope=None):
self.targeting_expression_id = targeting_expression_id
self.preferred_scope = preferred_scope or []
self.allowed_scope = allowed_scope or []
```