we are sorry, we are open

Понадобилось собирать и обрабатывать netflow-статистику с циски на машине с FreeBSD. Задачи довольно простые — иметь возможность анализа сетевого трафика внутри корпоративной сети, например, с целью выявления адресов, генерирующих аномальный трафик, а также выявление других проблем, влияющих на пропускную способность каналов.

Каналы связи в моём случае — это VLAN L2 с подключениями по ADSL, на скоростях 64 и 256 Кбит/с, с вероятностью расширения до 512 Кбит/с, т.е. трафика будет немного.

В качестве сборщика выбран пакет flow-tools, последней на данный момент версии — 0.68, который был установлен из портов: «/usr/ports/net-mgmt/flow-tools».
В качестве хранилища взят MySQL версии 5.1.34.

Установка указанных пакетов не вызвала никаких проблем и описывать её я не буду )

На первом этапе надо наладить сбор данных и их экспорт в таблицу MySQL.

Мои знания о netflow вообще и о flow-tools в частности весьма поверхностны, так что просьба не пинать за не самые лучшие решения, а помочь советом в случае чего )

Запуск сборщика данных

Запускаем коллектор, который будет получать данные от циски и сохранять их в файлах:

/usr/local/bin/flow-capture 
 -n48 
 -N0 
 -w /data/flow 
 -E1G 
 0/10.0.1.13/2055

Опишу использованные мной параметры:

• -n48 — количество ротаций файлов за сутки, 48 раз — это каждые полчаса;
• -N0 — формат имён файлов со статистикой, 0 — просто файл в указанной папке;
• -w /data/flow — папка для хранения файлов;
• -E1G — размер файла, по достижении которого, файл со статистикой ротейтится — 1ГБ, если размер файла не достигает указанного, то файл ротейтится согласно параметра -n48;
• 10.0.1.5/10.0.1.13/2055 — принимать данные, отправленные на адрес 10.0.1.5/принимать данные только от экпортёра 10.0.1.13/порт для ожидания данных;

Сборщик будет висеть на указанных айпишнике и порту, принимать данные от указанного адреса и складывать их во временный файл в папке /data/flow, например с таким именем:

tmp-v05.2009-05-11.183644+0600

Через промежуток времени, согласно параметра -n48, а в нашем случае это полчаса, временный файл будет переименован в окончательный, например вот такой:

ft-v05.2009-05-11.183644+0600

И затем будет создан новый временный файл:

tmp-v05.2009-05-11.190609+0600

Пишем скриптик для запуска коллектора и помещаем его в «/usr/local/etc/rc.d», в простом случае он может быть вот такой:

#!/bin/sh

CMD="/usr/local/bin/flow-capture 
 -n48 
 -N0 
 -w /data/flow 
 -E5G 
 10.0.1.5/10.0.1.13/2055"

PID="/var/run/flow-capture.pid.2055"

case "$1" in
 start)
  echo "Start flow-capture"
  $CMD
 ;;
 stop)
  echo "Stop flow-capture"
  kill -TERM `cat $PID`
 ;;
 *)
  echo " Usage: `basename $0` { start | stop }"
  exit 1
 ;;
esac

exit 0

Ну и конечно надо разрешить в нашем файрволе входящие пакеты с адреса экспортёра на адрес коллектора по протоколу udp и на порт 2055.

Настройка экспорта данных с циски

Настраиваем циску для экспорта данных с нужного нам интерфейса:

cs#conf t
cs(config)#ip flow-export version 5
cs(config)#ip flow-export destination 10.0.1.5 2055
cs(config)#int bvi1
cs(config-if)#ip flow egress
cs(config-if)#ip flow ingress
cs(config-if)#^Z

Посмотреть что получилось на циске можно вот так:

cs#show ip flow interface
BVI1
  ip flow ingress
  ip flow egress
cs#show ip flow export
Flow export v5 is enabled for main cache
  Export source and destination details :
  VRF ID : Default
    Destination(1)  10.0.1.5 (2055)
  Version 5 flow records
  4084394 flows exported in 150001 udp datagrams
  0 flows failed due to lack of export packet
  0 export packets were sent up to process level
  0 export packets were dropped due to no fib
  0 export packets were dropped due to adjacency issues
  0 export packets were dropped due to fragmentation failures
  0 export packets were dropped due to encapsulation fixup failures

Экспорт данных из файла в текстовый вид

Для проверки наличия нужных нам данных в файлах, которые нам создал коллектор, можно экпортировать их в текстовый формат:

flow-export -f2 -mDPKTS,DOCTETS,SRCADDR,DSTADDR < raw_file

Ключ -f2 задаёт формат вывода - ASCII с полями, разделёнными запятой. Поля, которые вы хотите увидеть задаются параметром -m, список всех полей есть в мане к flow-export. При указании полей, насколько я понял, нужно перечислять их в порядке следования в списке в мане, например, указать поле DPKTS после поля DSTADD нельзя - это приведёт к ошибке:

# flow-export -f2 -mDOCTETS,SRCADDR,DSTADDR,DPKTS < raw_file
flow-export: Out of order field: DPKTS
flow-export: ftxfield_parse(): failed

Список полей для экспорта можно задать и при помощи масок, что будет короче в написании, но сложнее в восприятии )

Подготовка MySQL для сохранения данных

Понятно, что прежде чем записывать данные в MySQL, нужно создать пользователя, базу данных и таблицу с нужными нам полями.
Описывать создание пользователя и базы я не буду в виду тривиальности задачи )
Остановлюсь на создании таблицы, в моём случае это было примерно так:

create table export (
 unix_secs int(11) unsigned null,
 exaddr varchar(16) null,
 dpkts int(10) unsigned null,
 doctets int(10) unsigned null,
 first int(11) unsigned null,
 last int(11) unsigned null,
 srcaddr varchar(16) null,
 dstaddr varchar(16) null,
 srcport smallint(5) unsigned null,
 dstport smallint(5) unsigned null,
 prot tinyint(3) unsigned null,
 key (srcaddr)
);

Возможно, из-за отсутствия опыта работы с netflow, я мог собрать ненужные мне поля, а какие-то интересные, наоборот, пропустить. Это можно легко поправить в рабочем порядке )

Экспорт данных из файла в MySQL

Теперь, когда СУБД готова к приёму наших данных, можно их экспортировать:

/usr/local/bin/flow-export -f3 
 -mUNIX_SECS,EXADDR,DPKTS,DOCTETS,FIRST,
 LAST,SRCADDR,DSTADDR,SRCPORT,DSTPORT,PROT 
 -u "flow:DBPassword:localhost:/tmp/mysql.sock:flow:export" < raw_file

Параметр -f3 задаёт формат экспорта - MySQL, параметр -u задаёт данные для доступа к базе данных - "пользователь:пароль:хост:порт:база:таблица".

Так же не забываем о необходимости соблюдения соответствия очерёдности полей, экпортируемых из файлов, с очерёдности полей, созданных в таблице.

Периодичность запуска этой процедуры определите сами с учётом своих нужд. В моём случае довольно и одного раза в час.

Также, в зависимости от ваших потребностей и возможностей, решайте экспортировать ли все данные в одну таблицу или в разные таблицы, например по дате. В последнем случае, конечно, одной указанной выше командой не обойтись - нужно, как минимум, ещё и создавать таблицы по мере необходимости.
Так же нужно подумать об удалении уже обработанных файлов с netflow-данными, чтобы не занимали место на диске почём зря )

Я решил, что мне удобнее хранить данные посуточно в отдельных таблицах с названиями типа "export_0512". Для это я написал небольшой скрипт на perl, который проверяет наличие таблицы для данного дня, при необходимости создаёт её и экспортирует в неё данные указанной выше командой, а также удаляет таблицу за этот день, но прошлого месяца.

Выборки из MySQL

Как извлекать нужные данные из БД учить не буду, т.к. и сам не очень хорошо в этом разбираюсь. Приведу лишь несколько запросов, которые я использую.

Найти десятку самых активных хостов из сети "10.255.0" по количеству исходящих байт:

select srcaddr,sum(doctets) as traf_out from export_0511
 WHERE srcaddr like '10.255.0.%' group by srcaddr order by traf_out desc limit 10;

Найти десятку портов получателей с самым большим исходящим трафиком для указанного айпишника:

select srcaddr,sum(doctets) as traf_int,dstport,prot from export_0512
 WHERE srcaddr like '10.255.0.3' group by dstport  order by traf_int desc limit 10;

Найти десятку удалённых хостов, на которые был наибольший исходящий трафик с указанного айпишника:

select srcaddr,dstaddr,sum(doctets) as traf_int from export_0511
 WHERE srcaddr like '10.255.0.3' group by dstaddr  order by traf_int desc limit 10;

Заключение

Данный пример позволяет только лишь быстро запустить сбор данных и сделать какие-либо запросы из базы данных. Различные проверки ошибок, удобства и т.п. каждый реализует сам в меру своих требований и фантазии )
Кому-то, наверняка, потребуется и веб-интерфейс для запросов в базу, но лично для меня и данного случая такой необходимости пока нет.

Буду благодарен за конструктивную критику и советы )