Filaments

Filaments are extensibility mechanism that allow you to write custom logic in Python and execute it on top of the live or replayed event stream. Filaments effectively turn Fibratus into a programmable security analytics engine, where you can build anything from simple event processors to complex detection pipelines.

Python is the lingua franca of penetration testers, threat hunters, and SecOps engineers. A vast ecosystem of security tooling already exists in Python. Filaments let you bring that ecosystem directly into Fibratus.

With filaments, you can reuse existing Python libraries and tooling, build custom detections beyond declarative rules, enrich events with external intelligence, or automate investigations.

+> Filaments principle is: If you can write it in Python, you can run it inside Fibratus.

Execution model

A filament is a Python script executed within Fibratus that processes the stream of events in real time or during replay. Filament has access to event data, process context, and internal Fibratus state.

Filaments are executed on top of the event stream. Each incoming event is passed to the filament. The filament can inspect, transform, correlate, or act on it. This makes filaments ideal for stateful analysis and complex heuristics that are difficult to express in rules.

Filament internals

From a technical standpoint each filament runs as a fully initialized Python interpreter instance. Fibratus interacts with the CPython API to bootstrap the interpreter, initialize the module from filament definition, declare functions, and other related tasks.

Event processing

Filament's backbone is the on_next_event function. This function is invoked for each event in the stream. The parameter of this function is the Python dictionary that contains event data. Here is the structure of such a dictionary object:

{
  'seq': 122344,
  'pid': 2034,
  'tid': 2453,
  'ppid': 45,
  'cwd': 'C:\Windows\system32',
  'exe': 'cmd.exe',
  'cmdline': 'cmd.exe rm /r',
  'sid': 'S-1-15-8',
  'cpu': 2,
  'name': 'CreateFile',
  'category': 'file',
  'timestamp': '2013-08-23 16:15:13.4323',
  'host': 'archrabbit',
  'description': 'Creates or opens a file or I/O device',
  'params': {
    'file_path': 'C:\WINDOWS\system32\config\systemprofile\AppData\WindowsApps\',
    'file_object': 'ffffa88c7ea077d0',
    'irp': 'ffffa88c746b2a88',
    'create_disposition': 'supersede',
    'share_mask': 'rw-',
    'type': 'directory'
  }
}

For a more convenient dictionary accesses, you can annotate the function with the dotdictify decorator.

from utils.dotdict import dotdictify

@dotdictify
def on_next_event(event):
    print(f'{event.name} generated by {event.exe}')

Initialization

If the on_init function is declared in the filament, any logic wrapped inside this function is executed prior to event processing. This is a convenient place for configuring the table columns or establishing the on_interval function triggering intervals among other initialization tasks.

def on_init():
    interval(1)

Termination

The on_stop function is called right before the Python interpreter is teared down. You can place any code you would like to get executed when the filament is stopped.

def on_stop():
    f.close()

Periodical actions

Filament has built-in support for scheduling timers. The timer, associated with the on_interval function, is fired after the interval specified by the interval function elapses. The minimum interval granularity is one second.

def on_interval():
    for ip, count in __connections__.copy().items():
      f.write(f'{ip.count}')

Filtering

The set_filter function defines a filter expression for the life span of a filament. Filter can be constructed dynamically, like in the following code snippet that defines a filter from the list:

set_filter("ps.name in (%s)" % (', '.join([f'\'{ps}\'' for ps in __procs__])))

Table rendering

Filaments are able to render tabular data on the console in a flicker-free fashion by using the frame buffers. To render a table, you start by defining the columns with the columns function. It's possible to sort the data by specifying the column via sort_by function. Finally, the add_row function appends rows to the table. When you're ready to draw the table, invoke the render_table function.

def on_init():
    columns(["Source", "Count"])
    sort_by('Count')
    interval(1)

def on_interval():
    for ip, count in __connections__.copy().items():
        add_row([ip, count])
    render_table()

Running filaments

Filaments are bootstrapped via the fibratus run command by specifying the filament name. Use the -f or --filament.name flags to indicate the filament you'd like to run.

$ fibratus run -f watch_files

The filament will keep running until the keyboard interrupt signal is received.

Filaments may require additional arguments to execute some conditional logic or set up a filter. Arguments are passed to a filament by specifying a list of comma-separated values after the filament name.

$ fibratus run -f "watch_files,powershell.exe"

This populates the sys.argv list with the provided arguments, where sys.argv[0] is the filament name.

Filters can be specified either via a command-line argument when running the filament or by calling the set_filter function during filament initialization. If a filter expression is provided in both, the one defined in set_filter takes precedence.

Listing filaments

By default, filaments are located in the %PROGRAMFILES%\Fibratus\Filaments directory. This location can be overridden by specifying an alternative path using the --filament.path flag or by modifying the configuration file.

To list available filaments, run the command below.

$ fibratus list filaments

Writing filaments

The most effective way to understand filaments is to build one from scratch. In this walkthrough, we’ll create a filament that retrieves an IP blacklist database and uses it to detect outbound and inbound connections to botnets, command-and-control (C&C) servers, and other suspicious endpoints.

We’ll start by defining a function that fetches the database and converts it into a list. Along the way, we’ll declare two regular expressions to validate standard IP addresses and those expressed in CIDR notation. The database is retrieved using the requests package.

def fetch_db(url):
    with requests.get(url) as r:
      ips = [ipaddress.ip_network(l) if '/' in l else \
            ipaddress.ip_address(l) \
          for l in r.text.splitlines() if IP_RE.match(l) \
            or IP_CIDR_RE.match(l)]
    return ips

Next, we invoke the fetch_db function during filament initialization and store the results in a global __fishy_ips__ list. To keep the data up to date, we schedule a periodic sync that refreshes the IP database every hour, ensuring we always have the latest indicators of spam networks and malicious hosts. Since our focus is on network activity, we also configure a filter to capture only network-related events.

def on_init():
    __fishy_ips__ = fetch_db(IP_DB_URL)
    interval(3600)
    set_filter("evt.category = 'net'")
    columns(["Source", "Destination", "Process"])

def on_interval():
    __fishy_ips__ = fetch_db(IP_DB_URL)

With the data in place, we can implement the core logic. This involves checking whether observed network flows match any entries in our blacklist, indicating potential communication with compromised infrastructure such as C&C servers. When a match is found, we add a new row and render a table displaying the source and destination IP addresses, along with the process responsible for initiating the connection.

@dotdictify
def on_next_event(event):
    sip = event.params.sip
    dip = event.params.dip
    if (sip in __fishy_ips__ or dip in __fishy_ips__) or \
        (sip or dip in net for net in __fishy_ips__ \
        if isinstance(net, ipaddress.IPv4Network)):
        add_row([sip, dip, event.exe])
    render_table()

In a more advanced setup, this logic could be extended to generate alerts and deliver them via channels such as Slack or email.

The complete filament implementation closely resembles the snippet shown above.

"""
Pinpoints network communications with botnets or C&C servers.
"""

import requests
import re
import ipaddress
from utils.dotdict import dotdictify

IP_RE      = re.compile(r"(?<!\d\.)(?<!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d|(?:\.\d))")
IP_CIDR_RE = re.compile(r"(?<!\d\.)(?<!\d)(?:\d{1,3}\.){3}\d{1,3}/\d{1,2}(?!\d|(?:\.\d))")
IP_DB_URL = 'http://rules.emergingthreats.net/fwrules/emerging-Block-IPs.txt'


def fetch_db(url):
    with requests.get(url) as r:
      ips = [ipaddress.ip_network(l) if '/' in l else ipaddress.ip_address(l) \
            for l in r.text.splitlines() if IP_RE.match(l) or IP_CIDR_RE.match(l)]
    return ips


def on_init():
    __fishy_ips__ = fetch_db(IP_DB_URL)
    interval(3600)
    set_filter("evt.category = 'net'")
    columns(["Source", "Destination", "Process"])


@dotdictify
def on_next_event(event):
    sip = event.params.sip
    dip = event.params.dip
    if (sip in __fishy_ips__ or dip in __fishy_ips__) or \
        (sip or dip in net for net in __fishy_ips__ if isinstance(net, ipaddress.IPv4Network)):
        add_row([sip, dip, event.exe])
    render_table()


def on_interval():
    __fishy_ips__ = fetch_db(IP_DB_URL)

Save it to, let's say, cc.py file inside the %PROGRAMFILES%\Fibratus\Filaments directory and you're ready to go. Run the filament with the following command.

$ fibratus run -f cc