(self, tg_id)
Get Sample Network indicators for a given ThreatGRID id
Get Sample Network indicators for a given ThreatGRID id
Get Sample Network indicators for a given ThreatGRID id
def sample_network(self, tg_id):
        """
        Get Sample Network indicators for a given ThreatGRID id
        """
        indicators = []
        url = '/api/v2/samples/' + tg_id + '/analysis/network_streams'
        response = self.api_request(url, {}, 'get')
        if response:
            if response.get('data'):
                # DNS
                for num in response.get('data', {}).get('items'):
                    item = response['data']['items'][num]
                    if item.get('protocol') == 'DNS':
                        # Process DNS lookups
                        dns_objects = item.get('decoded')
                        for obj in dns_objects:
                            result = {
                                'dns_query':    dns_objects[obj].get('query', {}).get('query_data'),
                                'dns_type':     dns_objects[obj].get('query', {}).get('query_type'),
                                }
                            dns_qid = dns_objects[obj].get('query', {}).get('query_id')
                            # Find the answer for each DNS query by id, type
                            for answer in dns_objects[obj].get('answers', []):
                                if answer.get('answer_id', 0) == dns_qid:
                                    if answer.get('answer_type', '') == result['dns_type']:
                                        result['dns_answer'] = answer.get('answer_data')
                                        break
                            indicators.append([result.get('dns_query'), IndicatorTypes.DOMAIN])
                            indicators.append([result.get('dns_answer'), IndicatorTypes.IPV4_ADDRESS])
                            self._add_result('threatgrid_dns'.format(tg_id), result.pop('dns_query'), result)
                self._notify()
                # HTTP
                for num in response.get('data', {}).get('items'):
                    item = response['data']['items'][num]
                    if item.get('protocol') == 'HTTP':
                        for decode in item.get('decoded'):
                            for entry in decode:
                                # Only show HTTP requests
                                if entry.get('type') == 'request':
                                    result = {
                                        'host':         entry.get('host'),
                                        'method':       entry.get('method'),
                                        'url':          entry.get('url'),
                                        'ua':           entry.get('headers', {}).get('user-agent'),
                                        'referer':      entry.get('headers', {}).get('referer'),
                                        'dst':          item.get('dst'),
                                        'dst_port':     item.get('dst_port'),
                                        }
                                    indicators.append([result.get('host'), IndicatorTypes.DOMAIN])
                                    indicators.append([result.get('dst'), IndicatorTypes.IPV4_ADDRESS])
                                    self._add_result('threatgrid_http'.format(tg_id), result.pop('host'), result)
                self._notify()
                # IP/Other
                for num in response.get('data', {}).get('items'):
                    item = response['data']['items'][num]
                    if item.get('protocol') == None:
                        result = {
                                'transport':    item.get('transport'),
                                'src':          item.get('src'),
                                'src_port':     item.get('src_port'),
                                'dst':          item.get('dst'),
                                'dst_port':     item.get('dst_port'),
                                'bytes':        item.get('bytes'),
                                'packets':      item.get('packets'),
                                }
                        indicators.append([result.get('dst'), IndicatorTypes.IPV4_ADDRESS])
                        self._add_result('threatgrid_ip'.format(tg_id), result.pop('transport'), result)
                self._notify()

                # Enable user to add unique indicators for this sample
                added = []
                for item in indicators:
                    if item[0]:
                        indicator = item[0].lower()
                        if indicator not in added:
                            added.append(indicator)
                            tdict = {}
                            if item[1] in (IndicatorTypes.IPV4_ADDRESS, IndicatorTypes.DOMAIN):
                                tdict = {'Type': item[1]}
                                id_ = Indicator.objects(value=indicator).only('id').first()
                                if id_:
                                    tdict['exists'] = str(id_.id)
                            self._add_result('add_threatgrid_indicators', indicator, tdict)
def sample_network ( self , tg_id ) : indicators = [ ] url = '/api/v2/samples/' + tg_id + '/analysis/network_streams' response = self . api_request ( url , { } , 'get' ) if response : if response . get ( 'data' ) : # DNS for num in response . get ( 'data' , { } ) . get ( 'items' ) : item = response [ 'data' ] [ 'items' ] [ num ] if item . get ( 'protocol' ) == 'DNS' : # Process DNS lookups dns_objects = item . get ( 'decoded' ) for obj in dns_objects : result = { 'dns_query' : dns_objects [ obj ] . get ( 'query' , { } ) . get ( 'query_data' ) , 'dns_type' : dns_objects [ obj ] . get ( 'query' , { } ) . get ( 'query_type' ) , } dns_qid = dns_objects [ obj ] . get ( 'query' , { } ) . get ( 'query_id' ) # Find the answer for each DNS query by id, type for answer in dns_objects [ obj ] . get ( 'answers' , [ ] ) : if answer . get ( 'answer_id' , 0 ) == dns_qid : if answer . get ( 'answer_type' , '' ) == result [ 'dns_type' ] : result [ 'dns_answer' ] = answer . get ( 'answer_data' ) break indicators . append ( [ result . get ( 'dns_query' ) , IndicatorTypes . DOMAIN ] ) indicators . append ( [ result . get ( 'dns_answer' ) , IndicatorTypes . IPV4_ADDRESS ] ) self . _add_result ( 'threatgrid_dns' . format ( tg_id ) , result . pop ( 'dns_query' ) , result ) self . _notify ( ) # HTTP for num in response . get ( 'data' , { } ) . get ( 'items' ) : item = response [ 'data' ] [ 'items' ] [ num ] if item . get ( 'protocol' ) == 'HTTP' : for decode in item . get ( 'decoded' ) : for entry in decode : # Only show HTTP requests if entry . get ( 'type' ) == 'request' : result = { 'host' : entry . get ( 'host' ) , 'method' : entry . get ( 'method' ) , 'url' : entry . get ( 'url' ) , 'ua' : entry . get ( 'headers' , { } ) . get ( 'user-agent' ) , 'referer' : entry . get ( 'headers' , { } ) . get ( 'referer' ) , 'dst' : item . get ( 'dst' ) , 'dst_port' : item . get ( 'dst_port' ) , } indicators . append ( [ result . get ( 'host' ) , IndicatorTypes . DOMAIN ] ) indicators . append ( [ result . get ( 'dst' ) , IndicatorTypes . IPV4_ADDRESS ] ) self . _add_result ( 'threatgrid_http' . format ( tg_id ) , result . pop ( 'host' ) , result ) self . _notify ( ) # IP/Other for num in response . get ( 'data' , { } ) . get ( 'items' ) : item = response [ 'data' ] [ 'items' ] [ num ] if item . get ( 'protocol' ) == None : result = { 'transport' : item . get ( 'transport' ) , 'src' : item . get ( 'src' ) , 'src_port' : item . get ( 'src_port' ) , 'dst' : item . get ( 'dst' ) , 'dst_port' : item . get ( 'dst_port' ) , 'bytes' : item . get ( 'bytes' ) , 'packets' : item . get ( 'packets' ) , } indicators . append ( [ result . get ( 'dst' ) , IndicatorTypes . IPV4_ADDRESS ] ) self . _add_result ( 'threatgrid_ip' . format ( tg_id ) , result . pop ( 'transport' ) , result ) self . _notify ( ) # Enable user to add unique indicators for this sample added = [ ] for item in indicators : if item [ 0 ] : indicator = item [ 0 ] . lower ( ) if indicator not in added : added . append ( indicator ) tdict = { } if item [ 1 ] in ( IndicatorTypes . IPV4_ADDRESS , IndicatorTypes . DOMAIN ) : tdict = { 'Type' : item [ 1 ] } id_ = Indicator . objects ( value = indicator ) . only ( 'id' ) . first ( ) if id_ : tdict [ 'exists' ] = str ( id_ . id ) self . _add_result ( 'add_threatgrid_indicators' , indicator , tdict )