Supercharging VT API With vt-py ​

>>> client = vt.Client(apikey="dummy")
>>> obj = client.get_object("/files/44d88612fea8a8f36de82e1278abb02f")
>>> exit()
Unclosed connector

# /// script
# dependencies = [
#   "vt-py==0.19.0",
# ]
# ///

import vt

print(vt.__version__)
print(vt.__file__)

import vt

with vt.Client(apikey="...") as client:
    obj = client.get_object("/files/44d88612fea8a8f36de82e1278abb02f")

# vt-py/vt/client.py
class Client:
    ...

  def get_object(
      self,
      path: str,
      *path_args: typing.Any,
      params: typing.Optional[typing.Dict] = None,
  ) -> Object:
    ...
    return make_sync(self.get_object_async(path, *path_args, params=params))

# vt-py/vt/utils.py
import asyncio
import typing


def make_sync(future: typing.Coroutine):
  """Utility function that waits for an async call, making it sync."""
  try:
    event_loop = asyncio.get_event_loop()
  except RuntimeError:
    # Generate an event loop if there isn't any.
    event_loop = asyncio.new_event_loop()
    asyncio.set_event_loop(event_loop)
  return event_loop.run_until_complete(future)

async with vt.Client(apikey="...") as client:
    obj = await client.get_object_async("/files/44d88612fea8a8f36de82e1278abb02f")

hashes: list[str] = ["..."]

async with vt.Client(apikey="...") as client:
    coroutines = [client.get_object_async(f"/files/{h}") for h in hashes]
    objects: list[vt.Object] = await asyncio.gather(*coroutines)

for obj in objects:
    print(obj.id, obj.type)

# vt-py/client.py
class Client:
  ...

  async def __aenter__(self):
    return self

  async def __aexit__(self, item_type, value, traceback):
    await self.close_async()

  def __enter__(self):
    return self

  def __exit__(self, item_type, value, traceback):
    self.close()

  async def close_async(self) -> None:
    """Like :func:`close` but returns a coroutine."""
    # Using getattr(self, '_session', None) instead of self._session because
    # close_async can be called from __del__ when the object is not yet
    # inialized and therefore the object doesn't have a _session. Calling
    # self._session in that case would raise AttributeError. See:
    # https://github.com/VirusTotal/vt-py/issues/125#issue-1449917146
    session = getattr(self, "_session", None)
    if session:
      await session.close()
      self._session = None

  def close(self) -> None:
    """Closes the client.

    When the client is not needed anymore it must be closed for releasing
    resources like TCP connections.

    Not closing the client after it's used might show error tracebacks
    indicating it was not properly closed.
    """
    return make_sync(self.close_async())

# Intelligence search
for obj in client.iterator(
    "/intelligence/search", params={"query": "..."},  batch_size=10, limit=100
):
    print(obj)

# IoC stream
for obj in client.iterator("/ioc-stream"):
    print(obj)

# async ver.
async for obj in client.iterator(
    "/intelligence/search", params={"query": "..."}, batch_size=10, limit=100
):
    print(obj)

for obj in client.feed(vt.FeedType.URLS):
    print(obj)

# async ver.
async for obj in client.feed(vt.FeedType.URLS):
    print(obj)

# https://www.virustotal.com/api/v3/{collection name}/{object id}?relationships={relationship 1},{relationship 2}
client.get_object(
    "/ip_addresses/...",
    params={"relationships": "communicating_files,downloaded_files"},
)

>>> obj = client.get_object("/files/275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f")
>>> obj.type
file
>>> obj.id
275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f
>>> obj.context_attributes
{}
>>> obj.relationships
{}

>>> obj.reputation
0
>>> obj.get("reputation")
0

>>> obj.first_seen_itw_date
datetime.datetime(2020, 2, 24, 23, 9, 20)
>>> obj.get("first_seen_itw_date")
1582585760

>>> obj.last_analysis_stats
{'malicious': 0, ...}
>>> obj.last_analysis_stats.malicious
AttributeError: 'WhistleBlowerDict' object has no attribute 'malicious'

>>> obj.to_dict()
{
    'type': 'file',
    'id': '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f',
    'attributes': { ... },
}

import json

import vt
from vt.object import UserDictJsonEncoder

obj = vt.Object(...)

json_str = json.dumps(obj.to_dict(), cls=UserDictJsonEncoder)

# can have AttributeError and KeyError
obj.pe_info["imphash"]
# redundant
obj.get("pe_info", {}).get("imphash")

# "pip install jsonpath-ng" is needed
>>> import dictpath
>>> obj = vt.Object(...)
>>> data = obj.to_dict()
>>> dictpath.get_all(data, "$.attributes.pe_info.sections[*].md5")
['d41d8cd98f00b204e9800998ecf8427e', ...]
>>> dictpath.get_all(data, "$.attributes.sandbox_verdicts.*.sandbox_name")
['Zenbox', ...]

import vt


with vt.Client(apikey="...") as client:
    for obj in client.iterator(
        "/intelligence/search", params={"query": "entity:url ...."}
    ):
        url: str | None = obj.context_attributes.get("url")
        if url is None:
            continue

        do_something(url)

# /// script
# dependencies = [
#   "arakawa",
#   "starlette",
#   "vt-py>=0.20.0",
# ]
# ///


import collections

import arakawa as ar
import pandas as pd
import vt
from starlette.config import Config
from starlette.datastructures import Secret

config = Config(".env")

VT_API_KEY: Secret = config("VT_API_KEY", cast=Secret, default="")  # type: ignore

with vt.Client(str(VT_API_KEY)) as client:
    objects = list(
        client.iterator(
            "/intelligence/search", params={"query": "entity:url p:10+"}, limit=10
        )
    )


def extract_attributes_with_url(obj: vt.Object) -> dict:
    # extract attributes and add url in context_attributes
    attributes = obj.to_dict()["attributes"]
    attributes["url"] = obj.context_attributes.get("url")
    return attributes


data = [extract_attributes_with_url(obj) for obj in objects]

df = pd.json_normalize(data)

ar.Report(
    ar.DataTable(df),
    df["title"].value_counts().to_frame(),
    df["favicon.raw_md5"].value_counts().to_frame(),
).save("sample.html", open=True)

# /// script
# dependencies = [
#   "loguru",
#   "starlette",
#   "typer",
#   "vt-py",
# ]
# ///

# Forked from https://github.com/VirusTotal/vt-py/blob/master/examples/intelligence_search_to_network_infrastructure.py
#
# Copyright © 2019 The vt-py authors under the Apache License version 2.0.
"""VT Intelligence searches to network IoCs.

This is a script to showcase how programmatic VT Intelligence searches can be
combined with file sandbox behavior lookups in order to generate network
indicators of compromise that can be fed into network perimeter defenses.

Read more:
https://www.virustotal.com/gui/intelligence-overview
https://docs.virustotal.com/reference/search
https://docs.virustotal.com/docs/virustotal-intelligence-introduction
"""

import asyncio
import re
from collections import defaultdict
from typing import Annotated, Any, DefaultDict, Literal, TypedDict, cast

import typer
import vt
from loguru import logger
from starlette.config import Config
from starlette.datastructures import Secret

config = Config(".env")

VT_API_KEY: Secret | None = config("VT_API_KEY", cast=Secret, default=None)

NetworkSingularType = Literal["domain", "ip", "url"]


class ContactedAddress(TypedDict):
    type: NetworkSingularType
    id: str
    context_attributes: dict[str, str] | None


NetworkPluralType = Literal["domains", "ips", "urls"]


class NetworkContainer(TypedDict):
    contacted_addresses: list[ContactedAddress]
    type: NetworkPluralType
    file: str


class VTISearchToNetworkInfrastructureHandler:
    """Class for handling the process of analyzing VTI search matches."""

    _SEARCH_ENTITY_REGEX = re.compile(r"entity: (\w+)")

    def __init__(self, apikey: Secret):
        self.apikey = apikey

        self.queue: asyncio.Queue[NetworkContainer] = asyncio.Queue()
        self.files_queue: asyncio.Queue[str] = asyncio.Queue()

        self.networking_counters: dict[NetworkPluralType, defaultdict[str, int]] = {
            "domains": defaultdict(lambda: 0),
            "ips": defaultdict(lambda: 0),
            "urls": defaultdict(lambda: 0),
        }
        self.networking_infrastructure: DefaultDict[
            str, DefaultDict[NetworkPluralType, Any]
        ] = defaultdict(lambda: defaultdict(lambda: {}))

    async def get_file_async(self, checksum: str, *, relationships: str | None = None):
        """Look up a file object."""
        async with vt.Client(str(self.apikey)) as client:
            return await client.get_object_async(
                f"/files/{checksum}", params={"relationships": relationships}
            )

    async def get_matching_files(self, query: str, max_files: int):
        """Query intelligence for files matching the given criteria."""
        entity_match = self._SEARCH_ENTITY_REGEX.match(query.lower())
        if entity_match and entity_match.group(1) != "file":
            raise ValueError("Only file search queries are valid in this example.")

        async with vt.Client(str(self.apikey)) as client:
            query = query.lower()
            url = "/intelligence/search"

            logger.info("Performing VT Intelligence search...")

            files = client.iterator(url, params={"query": query}, limit=max_files)
            async for matching_file in files:
                sha256 = cast(str, matching_file.sha256)
                await self.files_queue.put(sha256)

            logger.info(
                "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, relationships="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
            self.files_queue.task_done()

    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"]
                else:
                    # url type should have url in context_attributes
                    address = contacted_address["context_attributes"]["url"]  # type: ignore

                self.networking_counters[item_type][address] += 1

            self.queue.task_done()

    def print_results(self):
        print("=== 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]:
                        key = cast(ContactedAddress, key)
                        k = key["type"].upper()
                        context_attributes = key.get("context_attributes") or {}
                        v = context_attributes.get("url") or key["id"]
                        print(f"{k}: {v}")


def main(
    query: str,
    *,
    apikey: Annotated[
        str | None,
        typer.Option(help="Your VirusTotal API key. Defaults to VT_API_KEY ENV."),
    ] = None,
    limit: Annotated[int, typer.Option(help="Limit of files to process.")] = 10,
):
    key = apikey or VT_API_KEY
    assert key, "API key is required."

    async def inner():
        handler = VTISearchToNetworkInfrastructureHandler(Secret(str(key)))

        enqueue_files_task = asyncio.create_task(
            handler.get_matching_files(query, limit)
        )
        _ = asyncio.create_task(handler.get_network())  # noqa: RUF006
        _ = asyncio.create_task(handler.build_network())  # noqa: RUF006

        await asyncio.gather(enqueue_files_task)

        await handler.files_queue.join()
        await handler.queue.join()

        handler.print_results()

    asyncio.run(inner())


if __name__ == "__main__":
    typer.run(main)