🤑 ⚕️ 👼🏼 Mikrotikファイアウォールのログをデータベースに収集します 🎀 ➡️ ☃️

こんにちは

Mikrotikルーター用のネットワークトラフィックメタデータコレクションサーバーを簡単かつ自然に構成できることをお伝えしたいと思います。

目的：目標は、詳細な分析のために、「噛んだ」ファイアウォールログをデータベースに保存することです。

意味： rsyslogd v8以降の新しいLinuxディストリビューションはすべて実装に適しています。おそらく提案された構文はv7でも機能します。 DBMSも必要です。mariadbを選択しました。データベースの成長は、記録されたルールの数とは異なります。ドライブのサイズは任意であるため、私の場合、30〜40のルールが記録されます。これは1日あたり約12万行です。インデックスを含むデータベースを使用する月に、3.8 GBに増加しました。

メカニズム：ルーターはUDP経由でログをリモートサーバーに送信します。正規表現を使用して、rsyslogサーバーは不要な情報の文字列を削除し、SQL挿入を生成してDBMSに送信します。 DBMSは、挿入前にトリガーを使用して、rsyslogで解析できなかったフィールドの追加のクリーニングと分離を実行します。

RSYSLOGを構成する

ファイル/etc/rsyslog.confの編集
そこに次の行を追加します。

module(load="ommysql") module(load="imudp") input(type="imudp" port="514")

したがって、必要なモジュールをロードし、514 UDPポートを開きます。

Mikrotikからのログの行は次のようになります。

 20180927155341 BLOCKSMKNETS forward: in:ether6 - LocalTORF out:VLAN55 - RT_INET, src-mac 00:15:17:31:b8:d7, proto TCP (SYN), 192.168.0.234:2457->192.168.6.14:65535, len 60

ご覧のように、データベースに保存するための余分なものと明確な選択は困難です。
理論的には、このようなデータを追加する必要があります。

 20180927155341 ether6 VLAN5 192.168.0.234 2457 192.168.6.14 65535 00:15:17:31:b8:d7 TCP SYN forward BLOCKSMKNETS 60

rsyslogを1つだけ使用してこのような行を取得できませんでした。 RsyslogレギュラーはPOSIX ERE / BREを使用するため、lookaheadやlookbehindなどの機能を適用する方法はありません。

レギュラーをデバッグして試すことができるツールがあります。ポートをアドレスから分離したり、インターフェースの名前をin：やout：から分離したりできます。一部のスポーツおよびdportプロトコルが欠落していることに注意してください。

一般に、私の出力は次のようになりました。

 20180927155341 in:ether6 out:VLAN5 192.168.0.234:2457 192.168.6.14:65535 00:15:17:31:b8:d7 TCP (SYN) forward BLOCKSMKNETS 60

rsyslogの常連を調理する方法に関するドキュメントがあります。

最後のフォームでは、Mikrotik /etc/rsyslog.d/20-remote.confからログを受信するための構成ファイルは次のようになります。

 $template tpl_traflog,"insert into traflog.traffic (datetime, inif, outif, src, dst, smac, proto, flags, chain, logpref, len) values ('%timereported:::date-mysql%', '%msg:R,ERE,0,DFLT,0:in:[a-zA-Z]+[0-9]+|in:<[a-zA-Z]+-[a-zA-Z]+>--end%', '%msg:R,ERE,0,BLANK,0:out:[a-zA-Z]+[0-9]+|out:<[a-zA-Z]+-[a-zA-Z]+>--end%', '%msg:R,ERE,0,DFLT,0:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end%', '%msg:R,ERE,0,DFLT,1:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end%', '%msg:R,ERE,0,BLANK:([0-f]+:){5}[0-f]+--end%', '%msg:R,ERE,0,BLANK:\b[AX]{3,4}\b--end%', '%msg:R,ERE,0,BLANK:\([AZ]+\)|\(([AZ]+\,){1,3}[AZ]+\)--end%', '%msg:R,ERE,0,DFLT:[ax]+--end%', '%msg:F,32:2%', '%msg:R,ERE,0,DFLT:[0-9]+$--end%' )",SQL if ($fromhost-ip == '192.168.0.230') and ($syslogtag contains "firewall") then {action(type="ommysql" server="localhost" serverport="3306" db="traflog" uid="rsyslogger" pwd="rsyslogger" template="tpl_traflog") stop}

最初の行のテンプレート（テンプレート）の説明は、DBMSに転送するSQLコードの行です。
2行目は、アクションが発生する条件、つまりDBMSのレコードです。
条件は次のようになります：ログソース= 192.168.0.230（ if ($fromhost-ip == '192.168.0.230') ）そして、msg行に「firewall」が含まれる場合（および（$ syslogtagに「firewall」が含まれる））、モジュールを使用する場合接続パラメーターを使用したommysql（ then {action(type="ommysql" server="localhost" serverport="3306" db="traflog" uid="rsyslogger" pwd="..." ）テンプレートtpl_traflog（ template="tpl_traflog") ）、その後、行のさらなる処理を停止します（ stop} ）。

あなたの場合、何かがうまくいかない可能性があります。それは、インターフェースの名前またはログ接頭辞、おそらく何か他のものかもしれません。デバッグのために、次のことを行い、2行目にコメントを付け、新しいテンプレートと2つの新しい条件を追加します。

 $template tpl_traflog_test,"%timereported:::date-mysql% %msg:R,ERE,0,DFLT,0:in:[a-zA-Z]+[0-9]+|in:<[a-zA-Z]+-[a-zA-Z]+>--end% %msg:R,ERE,0,BLANK,0:out:[a-zA-Z]+[0-9]+|out:<[a-zA-Z]+-[a-zA-Z]+>--end% %msg:R,ERE,0,DFLT,0:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end% %msg:R,ERE,0,DFLT,1:([0-9]+\.){3}[0-9]+[:]?([0-9]+)?--end% %msg:R,ERE,0,BLANK:([0-f]+:){5}[0-f]+--end% %msg:R,ERE,0,BLANK:\b[AX]{3,4}\b--end% %msg:R,ERE,0,BLANK:\([AZ]+\)|\(([AZ]+\,){1,3}[AZ]+\)--end% %msg:R,ERE,0,DFLT:[ax]+--end% %msg:F,32:2% %msg:R,ERE,0,DFLT:[0-9]+$--end%\n" if ($fromhost-ip == '192.168.0.230') then {action(type="omfile" file="/var/log/remote/192.168.0.230.log" )} if ($fromhost-ip == '192.168.0.230') then {action(type="omfile" file="/var/log/remote/192.168.0.230.log" template="tpl_traflog_test" ) stop}

ロガーを再起動します。

tpl_traflog_testテンプレートはtpl_traflogに似ていますが、SQL INSERTはありません。

最初の条件は、テンプレートが指定されていないため、未処理の行％msg％をファイル/var/log/remote/192.168.0.230.logに追加します。

2番目の条件は、処理された行を同じファイルに追加します。
したがって、比較する方が便利です。
次に、データベースを準備します。

DBを準備します

DBMS設定を下げます。ここではすべてが標準です。

mysqlコンソールを起動して、次のコードを実行します。

 --   create database traflog character set utf8 collate utf8_bin; use traflog; --  create table traffic (id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, datetime DATETIME, inif VARCHAR(20), outif VARCHAR(20), src VARCHAR(21), sport INT(5), dst VARCHAR(21), dport INT(5), smac VARCHAR(17), proto VARCHAR(4), flags VARCHAR(11), chain VARCHAR(8), logpref VARCHAR(24), len INT(5)) ENGINE=MYISAM; --  create user rsyslogger@localhost identified by '...'; grant all privileges on traflog.* to rsyslogger@localhost;

ユーザーはテーブルの準備ができました。

ここでトリガーを追加します。ロガーがポートからアドレスを分離するのに失敗したことを実行し、インターフェイスの名前を消去し、フラグから角かっこを削除します。

 --   traffic DELIMITER // create TRIGGER delim_ip_port_flags BEFORE insert ON traffic FOR EACH ROW begin set NEW.inif = REGEXP_REPLACE ((NEW.inif), 'in:', '' ); set NEW.outif = REGEXP_REPLACE ((NEW.outif), 'out:', '' ); set NEW.sport = REGEXP_REPLACE ((NEW.src), '([0-9]+\.){3}[0-9]+:|([0-9]+\.){3}[0-9]+', '' ); set NEW.src = REGEXP_REPLACE ((NEW.src), ':[0-9]+', '' ); set NEW.dport = REGEXP_REPLACE ((NEW.dst), '([0-9]+\.){3}[0-9]+:|([0-9]+\.){3}[0-9]+', '' ); set NEW.dst = REGEXP_REPLACE ((NEW.dst), ':[0-9]+', '' ); set NEW.flags = REGEXP_REPLACE ((NEW.flags), '\\(|\\)', '' ); end // delimiter ;

REGEXP_REPLACEは、小数点の次の2番目のパラメーター（通常の季節）を検索し、それを3番目のパラメーターに置き換えます。この場合、引用符で囲まれていないため、検出されたものを単に削除します。

ロガーが行う方法と同様に、テスト挿入を作成しましょう。

 --   insert into traffic (datetime, inif, outif, src, dst, smac, proto, chain, logpref) values (20180730075437, 'in:ether6', 'out:VLAN55', '192.168.0.234:4997', '192.168.6.18:65535', '00:15:17:31:b8:d7', 'TCP', '(SYN)', 'forward', 'BLOCKSMKNETS');

何が起こったのか見てみましょう：

 select * from tarffic;

すべてが正しい場合は、次に進みます。そうでない場合は、間違いを探しています。

少なくとも1つのインデックスを追加します。私はインデックスを作成するマスターではありませんが、理解しているように、mysqlでは、異なるクエリに異なる結合フィールドを持つインデックスを使用する方が正しいです。なぜなら、1つのクエリは1つのインデックスしか使用できないからです（または間違っていますか？）。理解したら、あなたの裁量でそれをしてください。
私は頻繁に特定のプレフィックスでリクエストを作成する必要があるため、このインデックスを追加しました。

 --  create index traffic_index on traffic (datetime,logpref,src);

できた

ルーターでの送信を開始し、リモートログサーバーの設定とアクションを追加し、ファイアウォールルールの1つにログオプションを追加し、24文字以内のプレフィックスを追加する必要があります。

Mikrotikコンソールでは、次のようになります。

 /system logging action set 3 remote=192.168.0.94 src-address=192.168.0.230 add name=remote2 remote=192.168.0.19 syslog-facility=local6 target=remote /system logging add action=remote topics=error,account,critical,event,info add action=remote2 topics=firewall /ip firewall filter ... add action=drop chain=input comment="drop ssh brute forcers" dst-port=22,8291 log=yes log-prefix=DROP_SSH_BRUTE protocol=tcp src-address-list=ssh_blacklist ...

ここで、192.168.0.230はルーターのアドレス、192.168.0.19はファイアウォールログのログサーバーのアドレス、192.168.0.94は別のログサーバーです。Mikrotikシステムログがありますが、今は必要ありません。セットアップはremote2です。

次に、ファイルの内容を確認します。

 tail -f /var/log/remote/192.168.0.230.log

もちろん、ルールが非常に頻繁にトリガーされない限り、ルーターからの行はファイルに挿入する必要があります。

一部のフィールドが欠落している場合、つまり、シーケンスdatetime、inif、outif、src、dst、smac、proto、flags、chain、logpref、lenが後に続かない場合は、ロガーのデバッグテンプレートのパラメーターを変更して、BLANKをDLFTに置き換えることができます。次に、フィールドが空である代わりに、いくつかの文字が表示されます。どの文字がすでに記憶されているかは覚えていません。これが発生した場合、通常のスケジュールに何か問題があり、修正する必要があります。

すべてが正常に完了したら、テスト条件とテンプレートをオフにします。

また、以下の/etc/rsyslog.d/でデフォルトの設定を実行する必要があります。リモートログがシステムログ/ var / log / messageに注がれないように、50-default.confに名前を変更しました
ロガーを再起動します。

データベースがいっぱいになるまで少し待ちましょう。その後、選択を開始できます。

例のためのいくつかのクエリ：

データベースのサイズと行数を確認するには：

 MariaDB [traflog]> select table_schema as "database", round(sum(data_length + index_length)/1024/1024,2) as "size Mb", TABLE_ROWS as "count rows" from information_schema.tables group by table_schema; +--------------------+---------+------------+ | database | size Mb | count rows | +--------------------+---------+------------+ | information_schema | 0.17 | NULL | | traflog | 3793.39 | 21839553 | +--------------------+---------+------------+ 2 rows in set (0.48 sec)

1か月で約4GBが成長しましたが、ログに記録されたファイアウォールルールの数とプロパティに依存します

ログに記録されるプレフィックスの数

ログに記録されるプレフィックスの数はルールの数と等しくありません。一部のルールは1つのプレフィックスで機能しますが、それでも合計プレフィックスの数はいくつですか？そして、それらのためにいくつのルールが作成されましたか？：

 MariaDB [traflog]> select logpref,count(logpref) from traffic group by logpref order by count(logpref) desc; +----------------------+----------------+ | logpref | count(logpref) | +----------------------+----------------+ | ACCEPT_TORF_INET | 14582602 | | ACCEPT_SMK_PPP | 1085791 | | DROP_FORWARD_INVALID | 982374 | | REJECT_BNK01 | 961503 | | ACCEPT_MMAX_TORF | 802455 | | ACCEPT_TORF_PPP | 736803 | | SMTP_DNAT | 689533 | | ACCEPT_SMK_INET | 451411 | | ACCEPT_INET_TORF | 389857 | | BLOCK_SMKNETS | 335424 | | DROP_SMTP_BRUTE | 285850 | | ACCEPT_ROZN_TORF | 154811 | | ACCEPT_TORF_MMAX | 148393 | | DROP_ETHALL_ETHALL | 80679 | | ACCEPT_SMTP | 48921 | | DROP_SMTP_DDOS | 32190 | | RDP_DNAT | 28757 | | ACCEPT_TORF_ROZN | 18456 | | SIP_DNAT | 15494 | | 1CWEB_DNAT | 6406 | | BLOCKSMKNETS | 5789 | | DROP_SSH_BRUTE | 3162 | | POP_DNAT | 1997 | | DROP_RDP_BRUTE | 442 | | DROP_BNK01 | 291 | | DROPALL | 138 | | ACCEPT_RTP_FORWARD | 90 | | REJECT_SMTP_BRUTE | 72 | | L2TP_INPUT_ACCEPT | 33 | +----------------------+----------------+ 29 rows in set (2 min 51.03 sec)

ACCEPT_TORF_INETが先頭に立ちます。このプレフィックスにより、ローカルネットワークからインターネットにアクセスしたすべての人を見つけることができ、プロトコルとポートが記録され、時が来て一部のアクセスが閉じられます。バグに関する今後の作業の参照データがあります。

SMTPスピアリーダー

今日誰がsmtpサーバーにアクセスしようとしたかを見てみましょう。

 MariaDB [traflog]> select src,count(dport) from traffic where logpref='SMTP_DNAT' and datetime > '2018101600000000' group by src order by count(dport) desc limit 10; +----------------+--------------+ | src | count(dport) | +----------------+--------------+ | 191.96.249.92 | 12440 | | 191.96.249.24 | 4556 | | 191.96.249.61 | 4537 | | 185.255.31.122 | 3119 | | 178.57.79.250 | 226 | | 185.36.81.174 | 216 | | 185.234.219.32 | 211 | | 89.248.162.145 | 40 | | 45.125.66.157 | 32 | | 188.165.124.31 | 21 | +----------------+--------------+ 10 rows in set, 1 warning (21.36 sec)

今日、ノード191.96.249.92が勝者であることは明らかです。彼がまだ記録されているルールを見てみましょう：

 MariaDB [traflog]> select src,dport,count(dport),logpref from traffic where src='191.96.249.92' group by logpref order by count(dport) desc; +---------------+-------+--------------+-----------------+ | src | dport | count(dport) | logpref | +---------------+-------+--------------+-----------------+ | 191.96.249.92 | 25 | 226989 | SMTP_DNAT | | 191.96.249.92 | 25 | 170714 | DROP_SMTP_BRUTE | | 191.96.249.92 | 25 | 2907 | DROP_SMTP_DDOS | | 191.96.249.92 | 25 | 2061 | ACCEPT_SMTP | +---------------+-------+--------------+-----------------+ 4 rows in set (10 min 44.21 sec)

これはsmtpのみに特化しており、パスワードを推測しようとしたり、ゴミを送ろうとしたりしたヒットの1％程度で、残りは浴場に行きました。

リクエストには10分かかりました。これは非常に多く、現在のインデックスはそれに適していないか、リクエストを再定式化することはできますが、これについては説明しません。

将来的には、Webインターフェースを標準のクエリとフォームにねじ込むことが計画されています。
ベクトルが与えられています。この記事が役に立つことを願っています。

みんなありがとう！

参照：

Rsyslogドキュメント
MySQLドキュメント
Mikrotikロギングドキュメント

LORコミュニティのヒントをありがとう。

UPD.1
データベースにフラグフィールドが追加され、SYN、FINをキャッチすることで接続時間を追跡できるようになりました。
rsyslogレギュラーのいくつかのバグとmysqlトリガーを修正しました。

不思議なことに、デフォルトのdefconf：drop invalidルールはTCP接続のすべての最終パケットをドロップします。その結果、科学で接続を閉じようとするすべてのノードは失敗し、いくつかのFINを送信します。これは正しいですか？

ACK、FINフラグを使用したTCPトラバーサルを許可するルールを追加しました。

SQLスポイラーの下で、過去5分間のTCP接続の時間を示す手順

connections_list（）

 DROP PROCEDURE IF EXISTS connections_list; DELIMITER // CREATE PROCEDURE connections_list() BEGIN DECLARE logid BIGINT UNSIGNED; DECLARE done INT DEFAULT FALSE; DECLARE datefin DATETIME; DECLARE datesyn DATETIME; DECLARE conntime TIME; DECLARE connsport INT; DECLARE conndport INT; DECLARE connsrc VARCHAR(21); DECLARE conndst VARCHAR(21); DECLARE cur CURSOR FOR SELECT id,datetime,src,sport,dst,dport FROM conn_syn_fin WHERE flags='SYN'; DECLARE CONTINUE HANDLER FOR NOT FOUND SET done=TRUE; DROP TABLE IF EXISTS conn_syn_fin; DROP TABLE IF EXISTS connless; CREATE temporary TABLE connless(datestart DATETIME,dateend DATETIME,duration TIME,src VARCHAR(21),sport INT,dst VARCHAR(21),dport INT); CREATE temporary TABLE conn_syn_fin (SELECT * from traffic WHERE datetime > now() - interval 5 minute and src in (select src from traffic where datetime > now() - interval 5 minute and logpref='TCP_FIN' and flags like '%FIN%') and (flags like '%SYN%' or flags like '%FIN%') order by id); OPEN cur; read_loop: LOOP FETCH cur INTO logid,datesyn,connsrc,connsport,conndst,conndport; IF done THEN LEAVE read_loop; END IF; set datefin=(SELECT datetime FROM conn_syn_fin WHERE id>logid and src=connsrc and sport=connsport and flags like '%FIN%' and dst=conndst and dport=conndport limit 1); set conntime=(SELECT timediff(datefin,datesyn)); INSERT INTO connless (datestart,dateend,duration,src,sport,dst,dport) value (datesyn,datefin,conntime,connsrc,connsport,conndst,conndport); END LOOP; CLOSE cur; select * from connless; END; // DELIMITER ;

手順の結果、2つの一時テーブルが作成されます。
conn_syn_finテーブルには、SYNおよびFINフラグを持つログエントリが含まれ、このテーブルのカーソルを使用して検索が実行されます。
connlessテーブルには、接続のリストが含まれており、開いている状態と完了している状態、完了した状態の時間はそれぞれ開いている状態、ない
現在の時刻から5分を引いたサンプリング時間に注意してください。私の要求は遅いです。ゆっくりとカーソル検索を実行し、1秒あたり約10レコードを処理し、あらゆる方法で高速化しようとしましたが、実行時間は常にほぼ同じです。
また、この手順はデモンストレーションのみを目的としています。特定のsrc / sport / dst / dportを選択する必要がある場合は、これに似た別の手順を作成することをお勧めします。 SQLマスターの場合、クエリをより適切に作成できます。

connections_list（）を呼び出します。

 MariaDB [traflog]> call connections_list(); +---------------------+---------------------+----------+---------------+-------+-----------------+-------+ | datestart | dateend | duration | src | sport | dst | dport | +---------------------+---------------------+----------+---------------+-------+-----------------+-------+ | 2019-03-20 14:12:19 | 2019-03-20 14:13:14 | 00:00:55 | 192.168.0.81 | 41868 | 87.250.250.207 | 443 | | 2019-03-20 14:12:25 | NULL | NULL | 192.168.0.65 | 49311 | 52.5.23.125 | 443 | | 2019-03-20 14:12:31 | 2019-03-20 14:12:51 | 00:00:20 | 192.168.0.104 | 54433 | 217.69.139.42 | 443 | | 2019-03-20 14:12:31 | 2019-03-20 14:12:51 | 00:00:20 | 192.168.0.104 | 54434 | 217.69.139.42 | 443 | | 2019-03-20 14:12:32 | NULL | NULL | 192.168.0.119 | 37977 | 209.85.233.95 | 443 | ... | 2019-03-20 14:17:12 | NULL | NULL | 192.168.0.119 | 39331 | 91.213.158.131 | 443 | | 2019-03-20 14:17:13 | NULL | NULL | 192.168.0.90 | 63388 | 87.240.185.236 | 443 | +---------------------+---------------------+----------+---------------+-------+-----------------+-------+ 399 rows in set (33.17 sec) Query OK, 0 rows affected (33.18 sec)

手順が完了すると、一時テーブルconn_syn_finとconnlessが残ります。疑わしいものや信頼できないものが見つかった場合は、それらを詳細に確認できます。手順を開始すると、古いテーブルが削除され、新しいテーブルが表示されます。間違いを見つけたら書いてください。

Mikrotikファイアウォールのログをデータベースに収集します

RSYSLOGを構成する

DBを準備します

More articles: