### 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 [] ```