tcex.tcex_ti_batch module

ThreatConnect Batch Import Module.

class tcex.tcex_ti_batch.TcExBatch(tcex, owner, action=None, attribute_write_type=None, halt_on_error=True, playbook_triggers_enabled=None)[source]

Bases: object

ThreatConnect Batch Import Module

__len__()[source]

Return the number of groups and indicators.

__str__()[source]

Return string represtentation of object.

_gen_indicator_class()[source]

Generate Custom Indicator Classes.

_gen_indicator_method(name, custom_class, value_count)[source]

Dynamically generate custom Indicator methods.

Parameters:
  • name (str) – The name of the method.
  • custom_class (object) – The class to add.
  • value_count (int) – The number of value parameters to support.
_group(group_data)[source]

Return previously stored group or new group.

Parameters:group_data (dict|obj) – An Group dict or instance of Group object.
Returns:The new Group dict/object or the previously stored dict/object.
Return type:dict|obj
_indicator(indicator_data)[source]

Return previously stored indicator or new indicator.

Parameters:indicator_data (dict|obj) – An Indicator dict or instance of Indicator object.
Returns:The new Indicator dict/object or the previously stored dict/object.
Return type:dict|obj
action

Return batch action.

add_group(group_data)[source]

Add a group to Batch Job.

{
    "name": "Example Incident",
    "type": "Incident",
    "attribute": [{
        "type": "Description",
        "displayed": false,
        "value": "Example Description"
    }],
    "xid": "e336e2dd-5dfb-48cd-a33a-f8809e83e904",
    "associatedGroupXid": [
        "e336e2dd-5dfb-48cd-a33a-f8809e83e904:58",
    ],
    "tag": [{
        "name": "China"
    }]
}
Parameters:group_data (dict) – The full Group data including attributes, labels, tags, and associations.
add_indicator(indicator_data)[source]

Add an indicator to Batch Job.

{
    "type": "File",
    "rating": 5.00,
    "confidence": 50,
    "summary": "53c3609411c83f363e051d455ade78a7
                : 57a49b478310e4313c54c0fee46e4d70a73dd580
                : db31cb2a748b7e0046d8c97a32a7eb4efde32a0593e5dbd58e07a3b4ae6bf3d7",
    "associatedGroups": [
        {
            "groupXid": "e336e2dd-5dfb-48cd-a33a-f8809e83e904"
        }
    ],
    "attribute": [{
        "type": "Source",
        "displayed": true,
        "value": "Malware Analysis provided by external AMA."
    }],
    "fileOccurrence": [{
        "fileName": "drop1.exe",
        "date": "2017-03-03T18:00:00-06:00"
    }],
    "tag": [{
        "name": "China"
    }],
    "xid": "e336e2dd-5dfb-48cd-a33a-f8809e83e904:170139"
}
Parameters:indicator_data (dict) – The Full Indicator data including attributes, labels, tags, and associations.
address(ip, **kwargs)[source]

Add Address data to Batch object.

Parameters:
  • ip (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of Address.

Return type:

obj

adversary(name, **kwargs)[source]

Add Adversary data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Adversary.

Return type:

obj

asn(as_number, **kwargs)[source]

Add ASN data to Batch object.

Parameters:
  • as_number (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of ASN.

Return type:

obj

attribute_write_type

Return batch attribute write type.

campaign(name, **kwargs)[source]

Add Campaign data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • first_seen (str, kwargs) – The first seen datetime expression for this Group.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Campaign.

Return type:

obj

cidr(block, **kwargs)[source]

Add CIDR data to Batch object.

Parameters:
  • block (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of CIDR.

Return type:

obj

close()[source]

Cleanup batch job.

data

Return the batch data to be sent to the ThreatConnect API.

Processing Order: * Process groups in memory up to max batch size. * Process groups in shelf to max batch size. * Process indicators in memory up to max batch size. * Process indicators in shelf up to max batch size.

This method will remove the group/indicator from memory and/or shelf.

data_group_association(xid)[source]

Return group dict array following all associations.

Parameters:xid (str) – The xid of the group to retrieve associations.
Returns:A list of group dicts.
Return type:list
data_group_type(group_data)[source]

Return dict representation of group data.

Parameters:group_data (dict|obj) – The group data dict or object.
Returns:The group data in dict format.
Return type:dict
data_groups(groups, entity_count)[source]

Process Group data.

Parameters:groups (list) – The list of groups to process.
Returns:A list of groups including associations
Return type:list
data_indicators(indicators, entity_count)[source]

Process Indicator data.

debug

Return debug setting

document(name, file_name, **kwargs)[source]

Add Document data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • file_name (str) – The name for the attached file for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • file_content (str;method, kwargs) – The file contents or callback method to retrieve file content.
  • malware (bool, kwargs) – If true the file is considered malware.
  • password (bool, kwargs) – If malware is true a password for the zip archive is
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Document.

Return type:

obj

email(name, subject, header, body, **kwargs)[source]

Add Email data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • subject (str) – The subject for this Email.
  • header (str) – The header for this Email.
  • body (str) – The body for this Email.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • from_addr (str, kwargs) – The from address for this Email.
  • to_addr (str, kwargs) – The to address for this Email.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Email.

Return type:

obj

email_address(address, **kwargs)[source]

Add Email Address data to Batch object.

Parameters:
  • address (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of EmailAddress.

Return type:

obj

errors(batch_id, halt_on_error=True)[source]

Retrieve Batch errors to ThreatConnect API.

[{
    "errorReason": "Incident incident-001 has an invalid status.",
    "errorSource": "incident-001 is not valid."
}, {
    "errorReason": "Incident incident-002 has an invalid status.",
    "errorSource":"incident-002 is not valid."
}]
Parameters:
  • batch_id (str) – The ID returned from the ThreatConnect API for the current batch job.
  • (bool, default (halt_on_error) – True): If True any exception will raise an error.
event(name, **kwargs)[source]

Add Event data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • event_date (str, kwargs) – The event datetime expression for this Group.
  • status (str, kwargs) – The status for this Group.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Event.

Return type:

obj

file(md5=None, sha1=None, sha256=None, **kwargs)[source]

Add File data to Batch object.

Note

A least one file hash value must be specified.

Parameters:
  • md5 (str, optional) – The md5 value for this Indicator.
  • sha1 (str, optional) – The sha1 value for this Indicator.
  • sha256 (str, optional) – The sha256 value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • size (str, kwargs) – The file size for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of File.

Return type:

obj

file_len

Return the number of current indicators.

files

Return dictionary containing all of the file content or callbacks.

static generate_xid(identifier=None)[source]

Generate xid from provided identifiers.

Important

If no identifier is provided a unique xid will be returned, but it will not be reproducible. If a list of identifiers are provided they must be in the same order to generate a reproducible xid.

Parameters:identifier (list|str) – Optional string value(s) to be used to make a unique and reproducible xid.
group(group_type, name, **kwargs)[source]

Add Group data to Batch object.

Parameters:
  • group_type (str) – The ThreatConnect define Group type.
  • name (str) – The name for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Group.

Return type:

obj

group_len

Return the number of current groups.

group_shelf_fqfn

Return groups shelf fully qualified filename.

For testing/debugging a previous shelf file can be copied into the tc_temp_path directory instead of creating a new shelf file.

groups

Return dictionary of all Groups data.

groups_shelf

Return dictionary of all Groups data.

halt_on_batch_error

Return halt on batch error value.

halt_on_error

Return batch halt on error setting.

halt_on_file_error

Return halt on file post error value.

halt_on_poll_error

Return halt on poll error value.

host(hostname, **kwargs)[source]

Add Email Address data to Batch object.

Parameters:
  • hostname (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • dns_active (bool, kwargs) – If True DNS active is enabled for this indicator.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • whois_active (bool, kwargs) – If True WhoIs active is enabled for this indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of Host.

Return type:

obj

incident(name, **kwargs)[source]

Add Incident data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • event_date (str, kwargs) – The event datetime expression for this Group.
  • status (str, kwargs) – The status for this Group.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Incident.

Return type:

obj

indicator(indicator_type, summary, **kwargs)[source]

Add Indicator data to Batch object.

Parameters:
  • indicator_type (str) – The ThreatConnect define Indicator type.
  • summary (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of Indicator.

Return type:

obj

indicator_len

Return the number of current indicators.

indicator_shelf_fqfn

Return indicator shelf fully qualified filename.

For testing/debugging a previous shelf file can be copied into the tc_temp_path directory instead of creating a new shelf file.

indicators

Return dictionary of all Indicator data.

indicators_shelf

Return dictionary of all Indicator data.

intrusion_set(name, **kwargs)[source]

Add Intrusion Set data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of IntrusionSet.

Return type:

obj

mutex(mutex, **kwargs)[source]

Add Mutex data to Batch object.

Parameters:
  • mutex (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of Mutex.

Return type:

obj

poll(batch_id, retry_seconds=None, back_off=None, timeout=None, halt_on_error=True)[source]

Poll Batch status to ThreatConnect API.

{
    "status": "Success",
    "data": {
        "batchStatus": {
            "id":3505,
            "status":"Completed",
            "errorCount":0,
            "successCount":0,
            "unprocessCount":0
        }
    }
}
Parameters:
  • batch_id (str) – The ID returned from the ThreatConnect API for the current batch job.
  • retry_seconds (int) – The base number of seconds used for retries when job is not completed.
  • back_off (float) – A multiplier to use for backing off on each poll attempt when job has not completed.
  • timeout (int, optional) – The number of seconds before the poll should timeout.
  • (bool, default (halt_on_error) – True): If True any exception will raise an error.
Returns:

The batch status returned from the ThreatConnect API.

Return type:

dict

poll_timeout

Return current poll timeout value.

registry_key(key_name, value_name, value_type, **kwargs)[source]

Add Registry Key data to Batch object.

Parameters:
  • key_name (str) – The key_name value for this Indicator.
  • value_name (str) – The value_name value for this Indicator.
  • value_type (str) – The value_type value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of Registry Key.

Return type:

obj

report(name, **kwargs)[source]

Add Report data to Batch object.

Parameters:
  • name (str) – The name for this Group.
  • file_name (str) – The name for the attached file for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • file_content (str;method, kwargs) – The file contents or callback method to retrieve file content.
  • publish_date (str, kwargs) – The publish datetime expression for this Group.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Report.

Return type:

obj

save(resource)[source]

Save group|indicator dict or object to shelve.

Best effort to save group/indicator data to disk. If for any reason the save fails the data will still be accessible from list in memory.

Parameters:resource (dict|obj) – The Group or Indicator dict or object.
saved_groups

Return True if saved group files exits, else False.

saved_indicators

Return True if saved indicators files exits, else False.

saved_xids

Return previously saved xids.

settings

Return batch job settings.

signature(name, file_name, file_type, file_text, **kwargs)[source]

Add Signature data to Batch object.

Valid file_types: + Snort ® + Suricata + YARA + ClamAV ® + OpenIOC + CybOX ™ + Bro + Regex + SPL - Splunk ® Search Processing Language

Parameters:
  • name (str) – The name for this Group.
  • file_name (str) – The name for the attached signature for this Group.
  • file_type (str) – The signature type for this Group.
  • file_text (str) – The signature content for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Signature.

Return type:

obj

submit(poll=True, errors=True, process_files=True, halt_on_error=True)[source]

Submit Batch request to ThreatConnect API.

By default this method will submit the job request and data and if the size of the data is below the value synchronousBatchSaveLimit set in System Setting it will process the request synchronously and return the batch status. If the size of the batch is greater than the value set the batch job will be queued. Errors are not retrieve automatically and need to be enabled.

If any of the submit, poll, or error methods fail the entire submit will halt at the point of failure. The behavior can be changed by setting halt_on_error to False.

Each of these methods can also be called on their own for greater control of the submit process.

Parameters:
  • (bool, default (halt_on_error) – True): Poll for status.
  • (bool, default – True): Retrieve any batch errors (only if poll is True).
  • (bool, default – True): Send any document or report attachments to the API.
  • (bool, default – True): If True any exception will raise an error.
Returns.
dict: The Batch Status from the ThreatConnect API.
submit_all(poll=True, errors=True, process_files=True, halt_on_error=True)[source]

Submit Batch request to ThreatConnect API.

By default this method will submit the job request and data and if the size of the data is below the value synchronousBatchSaveLimit set in System Setting it will process the request synchronously and return the batch status. If the size of the batch is greater than the value set the batch job will be queued. Errors are not retrieve automatically and need to be enabled.

If any of the submit, poll, or error methods fail the entire submit will halt at the point of failure. The behavior can be changed by setting halt_on_error to False.

Each of these methods can also be called on their own for greater control of the submit process.

Parameters:
  • (bool, default (halt_on_error) – True): Poll for status.
  • (bool, default – True): Retrieve any batch errors (only if poll is True).
  • (bool, default – True): Send any document or report attachments to the API.
  • (bool, default – True): If True any exception will raise an error.
Returns.
dict: The Batch Status from the ThreatConnect API.
submit_create_and_upload(halt_on_error=True)[source]

Submit Batch request to ThreatConnect API.

Returns.
dict: The Batch Status from the ThreatConnect API.
submit_data(batch_id, halt_on_error=True)[source]

Submit Batch request to ThreatConnect API. :param batch_id: The batch id of the current job. :type batch_id: string

submit_file_content(method, url, data, headers, params, halt_on_error=True)[source]

Submit File Content for Documents and Reports to ThreatConnect API.

Parameters:
  • method (str) – The HTTP method for the request (POST, PUT).
  • url (str) – The URL for the request.
  • data (str;bytes;file) – The body (data) for the request.
  • headers (dict) – The headers for the request.
  • params (dict) – The query string parameters for the request.
  • (bool, default (halt_on_error) – True): If True any exception will raise an error.
Returns:

The response from the request.

Return type:

requests.models.Response

submit_files(halt_on_error=True)[source]

Submit Files for Documents and Reports to ThreatConnect API.

Critical Errors

  • There is insufficient document storage allocated to this account.
Parameters:(bool, default (halt_on_error) – True): If True any exception will raise an error.
Returns:The upload status for each xid.
Return type:dict
submit_job(halt_on_error=True)[source]

Submit Batch request to ThreatConnect API.

threat(name, **kwargs)[source]

Add Threat data to Batch object

Parameters:
  • name (str) – The name for this Group.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • xid (str, kwargs) – The external id for this Group.
Returns:

An instance of Threat.

Return type:

obj

url(text, **kwargs)[source]

Add URL Address data to Batch object.

Parameters:
  • text (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of URL.

Return type:

obj

user_agent(text, **kwargs)[source]

Add User Agent data to Batch object

Parameters:
  • text (str) – The value for this Indicator.
  • confidence (str, kwargs) – The threat confidence for this Indicator.
  • date_added (str, kwargs) – The date timestamp the Indicator was created.
  • last_modified (str, kwargs) – The date timestamp the Indicator was last modified.
  • rating (str, kwargs) – The threat rating for this Indicator.
  • xid (str, kwargs) – The external id for this Indicator.
Returns:

An instance of UserAgent.

Return type:

obj

write_batch_json(content)[source]

Write batch json data to a file.