main.py
def poll_indicators_from_ctix(taxii_discovery_url: str,
                              taxii_username: str,
                              taxii_password: str,
                              taxii_collection_id: str,
                              from_date: str,

                              misp_url: str,
                              misp_api_key: str):
    """
    This method is used to poll indicators from CTIX, package these indicators into a report,
    and push those packaged indicator to MISP

    :param taxii_discovery_url: Enter the TAXII discovery URL to connect to
    :param taxii_username:  Enter the TAXII username to authenticate with
    :param taxii_password: Enter the TAXII password to authenticate with
    :param taxii_collection_id: Enter the TAXII collection ID which houses the indicators
    :param misp_url: Enter the MISP URL to push the data to
    :param misp_api_key: Enter the MISP API key to authenticate to MISP with
    :param from_date: Enter the date, to poll data from. Form YYYY-MM-DD
    :return: Collection of MISP responses
    """
    try:
        response = []

        taxii_object = CyTaxii(discovery_url=taxii_discovery_url,
                               username=taxii_username,
                               password=taxii_password)
        misp_object = MISP(base_url=misp_url, api_key=misp_api_key)

        initial_taxii_response = taxii_object.get_feeds(collection_id=taxii_collection_id,
                                                        object_type="indicator",
                                                        added_after=from_date)
        from_list = []
        to_list = []

        if initial_taxii_response["status"] and len(initial_taxii_response["response"]["objects"]) != 0:

            # Flush the existing initial objects, to avoid duplications while sequentially overwriting pagination
            initial_taxii_response["response"]["objects"] = []

            to_range = initial_taxii_response["headers"].get("Content-Range")
            to_range = int(str(to_range).split("/")[1])

            if to_range > 1000:
                chunk_number = to_range / 1000
                chunk_number = math.ceil(chunk_number)
                from_no = 0
                tmp = 0

                for number in range(1, chunk_number + 1):
                    to_number = number * 1000
                    to_list.append(to_number)
                    from_no = tmp
                    from_list.append(from_no)
                    tmp = to_number

                for from_range, to_range in zip(from_list, to_list):
                    paginated_taxii_response = taxii_object.get_feeds(collection_id=taxii_collection_id,
                                                                      object_type="indicator",
                                                                      added_after=from_date,
                                                                      range_from=from_range,
                                                                      range_to=to_range)

                    if paginated_taxii_response["status"]:
                        initial_taxii_response["response"]["objects"].extend(
                            paginated_taxii_response["response"]["objects"])

                    else:
                        raise SyntaxError(paginated_taxii_response)
            else:
                paginated_taxii_response = taxii_object.get_feeds(collection_id=taxii_collection_id,
                                                                  object_type="indicator",
                                                                  added_after=from_date,
                                                                  range_from=0,
                                                                  range_to=1000)

                if paginated_taxii_response["status"]:
                    initial_taxii_response["response"]["objects"].extend(
                        paginated_taxii_response["response"]["objects"])

                else:
                    raise SyntaxError(paginated_taxii_response)

        else:
            raise SyntaxError(initial_taxii_response)

        for indicator in initial_taxii_response["response"]["objects"]:

            indicator_id = indicator.get("id", None)
            indicator_name = indicator.get("name", None)
            created_date = indicator.get("created", None)

            if created_date:
                created_date = parser.parse(created_date)

            if not created_date:
                created_date = datetime.datetime.now()

            report = Report(type="report", name=indicator_name, object_refs=[indicator_id], published=created_date,
                            labels=["indicator"], allow_custom=True)

            init_bundle = Bundle(indicator, report, allow_custom=True).serialize(pretty=True)

            misp_response = misp_object.upload_stix(stix_data=init_bundle)
            response.append(misp_response)

    except Exception as e:
        response = str(e)

    return response

    This method is used to poll indicators from CTIX, package these indicators into a report,
    and push those packaged indicator to MISP

    :param taxii_discovery_url: Enter the TAXII discovery URL to connect to
    :param taxii_username:  Enter the TAXII username to authenticate with
    :param taxii_password: Enter the TAXII password to authenticate with
    :param taxii_collection_id: Enter the TAXII collection ID which houses the indicators
    :param misp_url: Enter the MISP URL to push the data to
    :param misp_api_key: Enter the MISP API key to authenticate to MISP with
    :param from_date: Enter the date, to poll data from. Form YYYY-MM-DD
    :return: Collection of MISP responses
    
This method is used to poll indicators from CTIX, package these indicators into a report,
and push those packaged indicator to MISP
This method is used to poll indicators from CTIX package these indicators into a report and push those packaged indicator to MISP
def poll_indicators_from_ctix(taxii_discovery_url: str,
                              taxii_username: str,
                              taxii_password: str,
                              taxii_collection_id: str,
                              from_date: str,
                              misp_url: str,
                              misp_api_key: str):
    try:
        response = []
        taxii_object = CyTaxii(discovery_url=taxii_discovery_url,
                               username=taxii_username,
                               password=taxii_password)
        misp_object = MISP(base_url=misp_url, api_key=misp_api_key)
        initial_taxii_response = taxii_object.get_feeds(collection_id=taxii_collection_id,
                                                        object_type="indicator",
                                                        added_after=from_date)
        from_list = []
        to_list = []
        if initial_taxii_response["status"] and len(initial_taxii_response["response"]["objects"]) != 0:
            initial_taxii_response["response"]["objects"] = []
            to_range = initial_taxii_response["headers"].get("Content-Range")
            to_range = int(str(to_range).split("/")[1])
            if to_range > 1000:
                chunk_number = to_range / 1000
                chunk_number = math.ceil(chunk_number)
                from_no = 0
                tmp = 0
                for number in range(1, chunk_number + 1):
                    to_number = number * 1000
                    to_list.append(to_number)
                    from_no = tmp
                    from_list.append(from_no)
                    tmp = to_number
                for from_range, to_range in zip(from_list, to_list):
                    paginated_taxii_response = taxii_object.get_feeds(collection_id=taxii_collection_id,
                                                                      object_type="indicator",
                                                                      added_after=from_date,
                                                                      range_from=from_range,
                                                                      range_to=to_range)
                    if paginated_taxii_response["status"]:
                        initial_taxii_response["response"]["objects"].extend(
                            paginated_taxii_response["response"]["objects"])
                    else:
                        raise SyntaxError(paginated_taxii_response)
            else:
                paginated_taxii_response = taxii_object.get_feeds(collection_id=taxii_collection_id,
                                                                  object_type="indicator",
                                                                  added_after=from_date,
                                                                  range_from=0,
                                                                  range_to=1000)
                if paginated_taxii_response["status"]:
                    initial_taxii_response["response"]["objects"].extend(
                        paginated_taxii_response["response"]["objects"])
                else:
                    raise SyntaxError(paginated_taxii_response)
        else:
            raise SyntaxError(initial_taxii_response)
        for indicator in initial_taxii_response["response"]["objects"]:
            indicator_id = indicator.get("id", None)
            indicator_name = indicator.get("name", None)
            created_date = indicator.get("created", None)
            if created_date:
                created_date = parser.parse(created_date)
            if not created_date:
                created_date = datetime.datetime.now()
            report = Report(type="report", name=indicator_name, object_refs=[indicator_id], published=created_date,
                            labels=["indicator"], allow_custom=True)
            init_bundle = Bundle(indicator, report, allow_custom=True).serialize(pretty=True)
            misp_response = misp_object.upload_stix(stix_data=init_bundle)
            response.append(misp_response)
    except Exception as e:
        response = str(e)
    return response
def poll_indicators_from_ctix ( taxii_discovery_url : str , taxii_username : str , taxii_password : str , taxii_collection_id : str , from_date : str , misp_url : str , misp_api_key : str ) : try : response = [ ] taxii_object = CyTaxii ( discovery_url = taxii_discovery_url , username = taxii_username , password = taxii_password ) misp_object = MISP ( base_url = misp_url , api_key = misp_api_key ) initial_taxii_response = taxii_object . get_feeds ( collection_id = taxii_collection_id , object_type = "indicator" , added_after = from_date ) from_list = [ ] to_list = [ ] if initial_taxii_response [ "status" ] and len ( initial_taxii_response [ "response" ] [ "objects" ] ) != 0 : initial_taxii_response [ "response" ] [ "objects" ] = [ ] to_range = initial_taxii_response [ "headers" ] . get ( "Content-Range" ) to_range = int ( str ( to_range ) . split ( "/" ) [ 1 ] ) if to_range > 1000 : chunk_number = to_range / 1000 chunk_number = math . ceil ( chunk_number ) from_no = 0 tmp = 0 for number in range ( 1 , chunk_number + 1 ) : to_number = number * 1000 to_list . append ( to_number ) from_no = tmp from_list . append ( from_no ) tmp = to_number for from_range , to_range in zip ( from_list , to_list ) : paginated_taxii_response = taxii_object . get_feeds ( collection_id = taxii_collection_id , object_type = "indicator" , added_after = from_date , range_from = from_range , range_to = to_range ) if paginated_taxii_response [ "status" ] : initial_taxii_response [ "response" ] [ "objects" ] . extend ( paginated_taxii_response [ "response" ] [ "objects" ] ) else : raise SyntaxError ( paginated_taxii_response ) else : paginated_taxii_response = taxii_object . get_feeds ( collection_id = taxii_collection_id , object_type = "indicator" , added_after = from_date , range_from = 0 , range_to = 1000 ) if paginated_taxii_response [ "status" ] : initial_taxii_response [ "response" ] [ "objects" ] . extend ( paginated_taxii_response [ "response" ] [ "objects" ] ) else : raise SyntaxError ( paginated_taxii_response ) else : raise SyntaxError ( initial_taxii_response ) for indicator in initial_taxii_response [ "response" ] [ "objects" ] : indicator_id = indicator . get ( "id" , None ) indicator_name = indicator . get ( "name" , None ) created_date = indicator . get ( "created" , None ) if created_date : created_date = parser . parse ( created_date ) if not created_date : created_date = datetime . datetime . now ( ) report = Report ( type = "report" , name = indicator_name , object_refs = [ indicator_id ] , published = created_date , labels = [ "indicator" ] , allow_custom = True ) init_bundle = Bundle ( indicator , report , allow_custom = True ) . serialize ( pretty = True ) misp_response = misp_object . upload_stix ( stix_data = init_bundle ) response . append ( misp_response ) except Exception as e : response = str ( e ) return response
This method is used to poll indicators from CTIX, package these indicators into a report,
and push those packaged indicator to MISP
This method is used to poll indicators from CTIX package these indicators into a report and push those packaged indicator to MISP
"""
    This method is used to poll indicators from CTIX, package these indicators into a report,
    and push those packaged indicator to MISP

    :param taxii_discovery_url: Enter the TAXII discovery URL to connect to
    :param taxii_username:  Enter the TAXII username to authenticate with
    :param taxii_password: Enter the TAXII password to authenticate with
    :param taxii_collection_id: Enter the TAXII collection ID which houses the indicators
    :param misp_url: Enter the MISP URL to push the data to
    :param misp_api_key: Enter the MISP API key to authenticate to MISP with
    :param from_date: Enter the date, to poll data from. Form YYYY-MM-DD
    :return: Collection of MISP responses
    """ # Flush the existing initial objects, to avoid duplications while sequentially overwriting pagination
Collection of MISP responses
Collection of MISP responses
taxii_discovery_url
str
Enter the TAXII discovery URL to connect to
Enter the TAXII discovery URL to connect to
taxii_username
str
Enter the TAXII username to authenticate with
Enter the TAXII username to authenticate with
taxii_password
str
Enter the TAXII password to authenticate with
Enter the TAXII password to authenticate with
taxii_collection_id
str
Enter the TAXII collection ID which houses the indicators
Enter the TAXII collection ID which houses the indicators
from_date
str
Enter the date, to poll data from. Form YYYY-MM-DD
Enter the date to poll data from . Form YYYY - MM - DD
misp_url
str
Enter the MISP URL to push the data to
Enter the MISP URL to push the data to
misp_api_key
str
Enter the MISP API key to authenticate to MISP with
Enter the MISP API key to authenticate to MISP with