Skip to content
Branch: master
Find file Copy path
Find file Copy path
1 contributor

Users who have contributed to this file

167 lines (134 sloc) 5.9 KB
# -*- coding: utf-8 -*-
# Copyright © 2019 The vt-py authors. All Rights Reserved.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
"""VT Intelligence searches to network IoCs.
This is a script to showcase how programmatic VT Intelligence searches can be
combined with file sandbox behaviour lookups in order to generate network
indicators of compromise that can be fed into network perimeter defenses.
Read more:
import argparse
import asyncio
from collections import defaultdict
import vt
class VTISearchToNetworkInfrastructureHandler:
"""Class for handling the process of analysing VTI search matches."""
def __init__(self, apikey):
self.apikey = apikey
self.queue = asyncio.Queue()
self.files_queue = asyncio.Queue()
self.networking_counters = {
'domains': defaultdict(lambda: 0),
'ips': defaultdict(lambda: 0),
'urls': defaultdict(lambda: 0)}
self.networking_infrastructure = defaultdict(
lambda: defaultdict(lambda: {}))
async def get_file_async(self, checksum, relationships=None):
"""Look up a file object."""
url = '/files/{}'
async with vt.Client(self.apikey) as client:
if isinstance(relationships, str) and relationships:
url += '?relationships={}'.format(relationships)
file_obj = await client.get_object_async(url.format(checksum))
return file_obj
async def get_matching_files(self, query, max_files):
"""Query intelligence for files matching the given criteria."""
if not isinstance(query, str):
raise ValueError('Search filter must be a string.')
async with vt.Client(self.apikey) as client:
query = query.lower()
url = '/intelligence/search'
print('Performing VT Intelligence search...')
files = client.iterator(url, params={'query': query}, limit=max_files)
async for matching_file in files:
await self.files_queue.put(matching_file.sha256)
print('Search concluded, waiting on network infrastructure retrieval...')
async def get_network(self):
"""Retrieve the network infrastructure related to matching files."""
while True:
checksum = await self.files_queue.get()
file_obj = await self.get_file_async(
checksum, 'contacted_domains,contacted_ips,contacted_urls')
relationships = file_obj.relationships
contacted_domains = relationships['contacted_domains']['data']
contacted_urls = relationships['contacted_urls']['data']
contacted_ips = relationships['contacted_ips']['data']
await self.queue.put(
{'contacted_addresses': contacted_domains,
'type': 'domains',
'file': checksum})
await self.queue.put(
{'contacted_addresses': contacted_ips,
'type': 'ips',
'file': checksum})
await self.queue.put(
{'contacted_addresses': contacted_urls,
'type': 'urls',
'file': checksum})
self.networking_infrastructure[checksum]['domains'] = contacted_domains
self.networking_infrastructure[checksum]['ips'] = contacted_ips
self.networking_infrastructure[checksum]['urls'] = contacted_urls
async def build_network(self):
"""Build the stats of the network infrastructure."""
while True:
item = await self.queue.get()
item_type = item['type']
for contacted_address in item['contacted_addresses']:
if item_type in ('domains', 'ips'):
address = contacted_address['id']
address = contacted_address['context_attributes']['url']
self.networking_counters[item_type][address] += 1
def print_results(self):
"""Pretty print network IoCs for the given VTI search query."""
print('\n\n=== Results: ===')
for item in self.networking_infrastructure.items():
contacted_addr = item[1].values()
if any(contacted_addr):
for inf in item[1].items():
for key in inf[1]:
print('{}: {}'.format(
key.get('context_attributes', {}).get('url') or key.get('id')))
async def main():
"""Perform a VTI search and extract IoCs for each of the matches."""
parser = argparse.ArgumentParser(
description='Generate network IoCs for files matching a VTI query.')
parser.add_argument('--apikey', required=True, help='your VirusTotal API key')
parser.add_argument('--query', required=True,
help='VT Intelligence search query')
parser.add_argument('--limit', default=10, help='Limit of files to process.')
args = parser.parse_args()
loop = asyncio.get_event_loop()
handler = VTISearchToNetworkInfrastructureHandler(args.apikey)
enqueue_files_task = loop.create_task(
handler.get_matching_files(args.query, int(args.limit)))
network_task = loop.create_task(handler.get_network())
build_network_task = loop.create_task(
await asyncio.gather(enqueue_files_task)
await handler.files_queue.join()
await handler.queue.join()
if __name__ == '__main__':
loop = asyncio.get_event_loop()
You can’t perform that action at this time.