e?grep 代わりに postgresql を使ってみよう
日々集まる /var/log/security のログ。 これを検索するのに、e?grep(1) や awk(1) でシコシコやる代わりに、 postgresql でデータベース化して検索してみようという話
。
たとえば port 445 へのスキャンを検索するのに、
# egrep ':445 ' /var/log/security Apr 4 00:05:01 fw /kernel: ipfw: 5300 Deny TCP 24.31.94.200:2312 xx.xx.xx.96:445 in via fxp0 Apr 4 00:05:01 fw /kernel: ipfw: 5300 Deny TCP 24.31.94.200:2314 xx.xx.xx.97:445 in via fxp0とやる代わりに
psql=# select * from ipfw where dport = 445;
date | deny | proto | sip | sport | dip | dport
-------------------------------+------+-------+----------------+-------+---------------+-------
2003-04-17 00:05:01.043242+09 | 1 | 6 | 24.31.94.200 | 2312 | xx.xx.xx.96 | 445
2003-04-17 00:05:01.094519+09 | 1 | 6 | 24.31.94.200 | 2314 | xx.xx.xx.97 | 445
となる、みたいな感じ。
なお、私自身 postgresql は全く使った事がない上に、 データベースの知識もゼロに近いので、 大した事はできないのであしからず。 なお、設定には、 日本PostgreSQLユーザー会 の PostgreSQL 7.2.x 付属ドキュメント を参考にさせて頂いた。
んでは、行ってみヨ!!
postgresql のインストール。
ports を使えば楽勝。 また、perl から postgresql を使えるように、p5-Pg もインストールする。# cd /usr/ports/databases/postgresql7 # make install # make clean # cd /usr/ports/databases/p5-Pg # make install # make clean
postgres の初期化。
インストール後一度だけ必要。# su -l pgsql -c initdb
postgres の起動。
# /usr/local/etc/rc.d/010.pgsql.sh start pgsql#
うまく行けば postmaster というプロセスがいるはず。# ps auxww | grep postmaster pgsql 62310 0.0 0.7 6948 1864 p1- I 2:44PM 0:03.84 /usr/local/bin/postmaster (postgres)
失敗したら、 /var/log/pgsql を調べる。| IpcSemaphoreCreate: semget(key=5432003, num=17, 03600) failed: No space left on device
というエラーだったら、カーネルオプションにoptions SEMMAP=256 options SEMMNI=256 options SEMMNS=512 options SEMMNU=256 options SHMMAXPGS=4096 options SHMSEG=256
を追加してビルドし直す。 参照: カーネルリソースの管理データベースの作成(createdb)。
データベース名は syslog とする。 なお、以下の作業は全てユーザ pgsql で行う事。# su -l pgsql pgsql$ createdb syslog CREATE DATABASE
ちなみに、削除する時は `dropdb syslog' を実行する。 参照: データベースの作成psql の起動。
まず psql (postgresql のフロントエンド) を立ち上げる。 データベースの操作は、この psql から対話的に行う。pgsql$ psql syslog Welcome to psql 7.3.2, the PostgreSQL interactive terminal.
終了は、Control-D か \q。テーブルの作成 (CREATE TABLE)。
テーブルとは表の事だね。 一つのデータベースには複数のテーブルを作成できるが、 今回作るのは一つだけ。 テーブル名は ipfw とする。 表の項目はこんな感じ。
以下のように SQL 文を入力する。 一行で入力してもよいし、 途中で改行をいれてもよい。 セミコロン ; を入力するまで、 実際には処理は行われない。 SQL 文の途中だとプロンプトが =# ではなく -# になっていることに注意。 大文字小文字は区別されない。名前 内容 型 date 日付 日付型 deny deny したか pass したか 整数型 proto プロトコル (TCP/UDP 等) 整数型 sip ソース IP アドレス IPv4 address 型 sport ソース port 整数型 dip デスティネーション IP アドレス IPv4 address 型 dport デスティネーション port 整数型 syslog=# CREATE TABLE ipfw ( syslog-# date timestamp with time zone, syslog-# deny integer, syslog-# proto integer, syslog-# sip inet, syslog-# sport integer, syslog-# dip inet, syslog-# dport integer); CREATE TABLE
参照: 新しいテーブルの作成 CREATE TABLE データ型テーブルの確認は、psql の \d コマンドで行える。(セミコロン不要)
syslog=# \d List of relations Schema | Name | Type | Owner --------+------+-------+------- public | ipfw | table | pgsql (1 row) syslog=# \d ipfw Table "public.ipfw" Column | Type | Modifiers --------+--------------------------+----------- date | timestamp with time zone | deny | integer | proto | integer | sip | inet | sport | integer | dip | inet | dport | integer |データの追加 (INSERT INTO)。
数値以外のデータは ' で囲む。 最後の ; を忘れない事。syslog=# INSERT INTO ipfw VALUES ('2003-04-16 00:00:00+9', 1, 6, '10.20.30.40', 12345, '192.168.0.1', 80); INSERT 16981 1 syslog=# INSERT INTO ipfw VALUES ('2003-04-16 00:01:23+9', 1, 17, '10.20.30.40', 137, '192.168.0.1', 137); INSERT 16982 1参照: テーブルに行を挿入 INSERTデータの検索 (SELECT)。
最後の ; を忘れない事。syslog=# SELECT * FROM ipfw; date | deny | proto | sip | sport | dip | dport ------------------------+------+-------+-------------+-------+-------------+------- 2003-04-16 00:00:00+09 | 1 | 6 | 10.20.30.40 | 12345 | 192.168.0.1 | 80 2003-04-16 00:01:23+09 | 1 | 17 | 10.20.30.40 | 137 | 192.168.0.1 | 137 (2 rows)条件付きの検索は次のように WHERE の後に条件式(結構直観的)を書けばよい。SELECT * FROM ipfw WHERE dport = 80 AND date >= '2003-04-16 14:00:00';
出力される順序はランダムなので、 時系列で見たい場合は、 ORDER BY で日付でソートして出力させる。SELECT * FROM ipfw WHERE dport = 80 ORDER BY date;
参照: テーブルへの問い合わせ SELECT 問い合わせ 関数と演算子データの削除(DELETE)。
全データを削除する。syslog=# DELETE FROM ipfw; DELETE 2 syslog=# SELECT * FROM ipfw; date | deny | sip | sport | dip | dport | proto ------+------+-----+-------+-----+-------+------- (0 rows)
SELECT と同様に、WHERE で条件を指定して削除することもできる。DELETE FROM ipfw WHERE dport = 80 ORDER BY date;
データを削除しても、その残骸は残り続ける。 それを消すには VACUUME を実行する。syslog=# VACUUME ipfw; VACUUM
参照: 削除 DELETE VACUUME以上で準備終り。
TCP 接続許可。
/var/log/security は各ホストに分散しているので、 それをデータベースに集中させるために、 postgres に TCP 経由で接続できるようにする。 ~pgsql/data/postgresql.conf を編集し、 次の行を追加する。(ユーザ pgsql で行う)tcpip_socket = true
編集したら、一旦 root になり、postgresql を再起動する。$ su # /usr/local/etc/rc.d/010.pgsql.sh stop # /usr/local/etc/rc.d/010.pgsql.sh start
認証設定。
~pgsql/data/pg_hba.conf を編集し、 接続許可するホストを登録する。 デフォルトでは自ホストからの接続は、 誰でも何にでもなれるというスカスカの状態(trust)なので、 パスワード認証を行う(password)ようにする。local all all password host all all 127.0.0.1 255.255.255.255 password host all all 192.168.0.0 255.255.255.0 password
編集したら、pg_ctl で設定の再読み込みを行う。pgsql$ pg_ctl -D ~pgsql/data reload postmaster successfully signaled
参照 クライアント認証注意
local に password を設定すると、OS 再起動時に Password: を聞いてきて、 そこから先のスタートアップスクリプトが全く動かないという事態に陥る。 そこで、少しまずい方法だが、 /usr/local/etc/rc.d/010.pgsql.sh を編集し、start) (中略) su -l pgsql -c \ "[ -d \${PGDATA} ] && exec ${PREFIX}/bin/pg_ctl start -s -l ${logfile}" sleep 5のように、-w オプションを外して sleep を入れ、 起動シーケンスをブロックしないようにする。 (参照: ほそかわさんのページ)ユーザを追加する(CREATE USER)。
ユーザ candy を追加する。 ついでに pgsql ユーザのパスワードも変更する。 これらのユーザ名やパスワードは postgresql が独自に管理するものなので、 UNIX passwd とは全く無関係に選んで良い。pgsql$ pgsql syslog syslog=# CREATE USER candy SYSID 10100 PASSWORD 'password'; syslog=# GRANT ALL ON ipfw TO candy; syslog=# \du List of database users User name | User ID | Attributes -----------+---------+---------------------------- candy | 10100 | pgsql | 1 | superuser, create database (2 rows) syslog=# SELECT * FROM pg_user ; usename | usesysid | usecreatedb | usesuper | usecatupd | passwd | valuntil | useconfig ---------+----------+-------------+----------+-----------+----------+----------+----------- pgsql | 1 | t | t | t | ******** | | candy | 10100 | f | f | f | ******** | | (2 rows) # ALTER USER pgsql PASSWORD 'password'; ALTER USER参照: CREATE USER ALTER USER GRANTリモートホストからアクセス。
リモートホスト(remote.example.com)から 自ホスト(sql.example.com)へアクセスしてみる。 (リモートホストにも postgresql と p5-Pg をインストールすること)remote.example.com$ psql -h sql.example.com -U candy syslog Password: Welcome to psql 7.3.2, the PostgreSQL interactive terminal. syslog=>
プロンプトが => になっていることに注意。 特権ユーザ(pgsql)の場合は =# になる。INSERT, SELECT, DELETE ができる事を確認する。 もし Permission deny の場合は、pgsql でログインし GRANT の設定を見直す。
/var/log/security* をデータベースに放り込む。
この ipfw2sql をダウンロードして chmod +x しておく。 psql の COPY を使って、一気にデータベース化する。 COPY はデータファイルを直接 postgresql に渡す方法で、 書式は 1 行 1 レコードで、各フィールドはタブで区切る。 数値以外でも ' で囲む必要はない。# cat /var/log/security | ./ipfw2sql -c | psql -h sql.example.com -U candy syslog -c 'COPY ipfw FROM stdin;' Password:
なぜか COPY は(特権ユーザ以外) stdin からしか受け付けない仕様らしい。 参照: COPYデータ自動収集
syslogd(8) から ipfw2sql を呼ぶようにし、自動でデータを収集できるようにする。 /etc/syslog.conf を編集し、security.* /var/log/security
の行の下に、security.* | /どこか/ipfw2sql -h sql.example.com -U candy -W password -d syslog
を追加する。 ipfw2sql はフルパスで記述すること。 編集したら、syslog.conf を再読み込みさせる。# kill -HUP `cat /var/run/syslogd.pid`
以上。
ポートスキャンを数えてみよう
postgresql に ipfw(8) のログを突っ込むことに成功したので、 そのデータの利用方法を考える。 とはいっても、egrep の代替以外に何ができるか分かってないので、 チュートリアル を読んでみると、 個数を数えることができる らしい。syslog=> SELECT dport, count(dport) FROM ipfw WHERE date >= '2003-04-18' GROUP BY dport HAVING count(dport) > 1;和訳すると、 「今日のデータから、dport (destination port) ごとにデータ数をカウントして、 データ数が 2 以上のものを表示せよ」 ということらしい。 要するに dport の度数分布。 その結果はこんな感じ。
dport | count
-------+-------
0 | 14
23 | 20
25 | 589
53 | 28
80 | 1251
111 | 3
135 | 16
137 | 216
139 | 55
443 | 59
445 | 1217
1080 | 96
1180 | 21
1226 | 5
1433 | 117
1434 | 149
3128 | 11
3389 | 57
3749 | 5
5497 | 18
6588 | 60
8080 | 10
15727 | 3
23518 | 17
26506 | 2
(25 rows)
これを見ると、
80 と 445 へのスキャンが多いなあ、とか、
port 6588 って何だろうとか思ったりするわけだ。
ほ〜〜、port 111 (sunrpc) なんてとこにスキャンがあるなあ、 どこのどいつだ!? そこで、その記録を調べる。
syslog=> SELECT * FROM ipfw WHERE date >= '2003-04-18' AND dport = 111;
date | deny | proto | sip | sport | dip | dport
-------------------------------+------+-------+----------------+-------+---------------+-------
2003-04-18 03:19:28.682577+09 | 1 | 6 | xx.197.165.164 | 3791 | xx.xxx.xx.105 | 111
2003-04-18 03:19:28.683185+09 | 1 | 6 | xx.197.165.164 | 3786 | xx.xxx.xx.100 | 111
2003-04-18 03:19:28.683531+09 | 1 | 6 | xx.197.165.164 | 3795 | xx.xxx.xx.109 | 111
(3 rows)
whois(1) してみるとどうも日本の会社らしい(アラアラ)。
んじゃ、このネットワークから他のスキャンがあったかな?
syslog=> SELECT * FROM ipfw WHERE sip >= 'xx.197.165.160' AND sip <= 'xx.197.165.167';
date | deny | proto | sip | sport | dip | dport
-------------------------------+------+-------+----------------+-------+---------------+-------
2003-04-13 07:53:38.606487+09 | 1 | 6 | xx.197.165.164 | 3713 | xx.xxx.xx.96 | 111
2003-04-13 22:35:48.887886+09 | 1 | 6 | xx.197.165.164 | 3529 | xx.xxx.xx.107 | 111
2003-04-13 22:35:48.888172+09 | 1 | 6 | xx.197.165.164 | 3533 | xx.xxx.xx.111 | 111
2003-04-14 18:59:46.621748+09 | 1 | 6 | xx.197.165.164 | 4059 | xx.xxx.xx.104 | 111
2003-04-14 18:59:46.622061+09 | 1 | 6 | xx.197.165.164 | 4055 | xx.xxx.xx.100 | 111
2003-04-14 18:59:46.622389+09 | 1 | 6 | xx.197.165.164 | 4051 | xx.xxx.xx.96 | 111
2003-04-14 18:59:46.622758+09 | 1 | 6 | xx.197.165.164 | 4064 | xx.xxx.xx.109 | 111
2003-04-15 00:54:07.159512+09 | 1 | 6 | xx.197.165.164 | 3783 | xx.xxx.xx.109 | 111
2003-04-18 03:19:28.682577+09 | 1 | 6 | xx.197.165.164 | 3791 | xx.xxx.xx.105 | 111
2003-04-18 03:19:28.683185+09 | 1 | 6 | xx.197.165.164 | 3786 | xx.xxx.xx.100 | 111
2003-04-18 03:19:28.683531+09 | 1 | 6 | xx.197.165.164 | 3795 | xx.xxx.xx.109 | 111
とまあこんな感じで、
これはかなり便利カモ
。
プロトコル名を表示してみよう
実は、当初の目標にしていたのは、 「一定時間ごとの、destination port ごとのアクセス数を調べる」 ということだったのだ。 もともと SELECT と WHERE しか知らなかったので、 それを dport ごとにループで回して……と考えていたのだが、 前回たった一行の SQLSELECT dport, count(dport) FROM ipfw WHERE date >= '2003-04-18' GROUP BY dport HAVING count(dport) > 1;でできてしまった。 あとはこれを cron で定時処理してグラフ化すれば、 この時 の「なんとか視覚化したい」という計画は完了。 でもその前に、 port 番号をプロトコル名で見たいという気持ちがムラムラと湧いて来たので、 それを実現してみる。
IP のプロトコル名は /etc/services (services(5)) に定義されているので、 これを COPY すればいいだろう。
- まずテーブルを作る。
$ psql syslog syslog=> CREATE TABLE services (port integer, name char(32));
- /etc/services を awk で整形し、COPY で流し込む。
TCP と UDP で別に管理したい気もするが、
とりあえずはごっちゃでいいことにしよう。
$ awk '/^[^#]/{ p = $2 + 0; if (x[p] == 0) printf("%s\t%s\n", p, $1); x[p] = 1;}' /etc/services | psql -h pgsql -U candy syslog -c 'COPY services FROM stdin;' -
チュートリアルの
「テーブル間を結合」
によると、
未定義の port も表示するためには、
OUTER JOIN を使うということらしい。
まず私が自分で考えた方法は、
例の度数分布表をビューに定義し、ビューと services を結合する。
(以下適宜改行を挿入している)
参照:ビュー
CREATE VIEW
SELECT (JOIN)
syslog=> CREATE VIEW myview as SELECT dport, count(dport) FROM ipfw WHERE date >= '2003-04-22' GROUP BY dport HAVING count(dport) > 1 ; syslog=> SELECT dport, count, name FROM myview LEFT OUTER JOIN services ON (myview.dport = services.port); # --- (1) dport | count | name -------+-------+---------------------------------- 21 | 30 | ftp 25 | 7 | smtp 80 | 475 | http 111 | 14 | sunrpc 137 | 151 | netbios-ns 139 | 7 | netbios-ssn 443 | 98 | https 445 | 233 | microsoft-ds 1024 | 2 | 1080 | 48 | socks [snip] - これで一応できたけど、
時間を指定するのにいちいち VIEW を CREATE/DROP する必要があるのは美しくない。
てなわけで、SQL の先生の O 君に聞いてみる。
すると、
「SELECT した結果を ( ) で括ってやれば、テーブルと同様に扱う事ができる」
とのこと!! 上(1)の SELECT 文の myview を (SELECT …) で置き換えるとsyslog=> SELECT dport, count, name FROM (SELECT dport, count(dport) FROM ipfw WHERE date >= '2003-04-22' GROUP BY dport HAVING count(dport) > 1) hist LEFT OUTER JOIN services ON (hist.dport = services.port); dport | count | name -------+-------+---------------------------------- 21 | 30 | ftp 25 | 7 | smtp 80 | 484 | http [snip] (ここで hist というのは、後で参照するために (SELECT …) に与える一時的な名称)なるほど〜〜〜〜〜。 こうやって帰納的(再帰的)にどんどんテーブルを結合して行けば、 任意の選択が可能になるわけか〜〜〜〜
。
よく見ると
「集約関数」
のとこに「副問い合わせ」という形で例示があった。
candy は SQL レベルが 1 上がった!!
ポートスキャンをグラフ化してみよう
てなわけで、御膳立ては整ったので、さっくりとグラフ化してみる。 目的は、新しいポートスキャンをいち早く検出することである。 そこで、1 日ごとの統計と 2 時間ごとの統計の 2 本立てで、 それぞれスキャン数の多い順にトップ N をグラフ化する。- gnuplot をインストールする。
# cd /usr/ports/math/gnuplot+ # make install # make clean
- ipfwcount と ipfwplot をダウンロードする。
- 過去データを集計し、1 日ごとのログと 2 時間ごとのログを作る。
$ ipfwcount -h pgsql.example.com -U candy -W candy -d syslog -s '2003-04-01' -i 86400 > log.daily # 4 月 1 日から 1 日ごと $ ipfwcount -h pgsql.example.com -U candy -W candy -d syslog -s '2003-04-30' -i 7200 > log.hourly # 4 月 30 日から 2 時間ごと オプション -h ホスト名 -U ユーザ名 -W パスワード -d データベース名 -s 開始日付。無指定なら期間秒前から。 -l 期間(秒)。無指定なら開始日付から現在まで。 -i インターバル(秒)
- プロットする。
$ ipfwplot -n 30 -x '%m/%d' log.daily > daily.png $ ipfwplot -n 24 -x '%H:%M' log.hourly > hourly.png
- cron(8) で定期的に実行する。
0 0 * * * cd /どこか/ ; ./ipfwcount -h pgsql.example.com -U candy -W candy -d syslog -l 86400 -i 86400 >> log.daily ; ipfwplot -n 30 -x '%m/%d' log.daily > daily.png 0 */2 * * * cd /どこか/ ; ./ipfwcount -h pgsql.example.com -U candy -W candy -d syslog -l 7200 -i 7200 >> log.hourly ; ipfwplot -n 24 -x '%H:%M' log.houry > hourly.png
- うまくいくと、 今日の port scan のようなグラフが随時更新されていくぞ。
関連リンク
ポートスキャンをグラフ化してみよう(part 2)
指定ポートのスキャン数をプロットする CGI を作った。 例えば port 80 なら こんな感じ。インストール方法。
- plot-cgi.txt を plot.cgi という名前で保存し、 chmod +x しておく。
- 前節で log.daily log.hourly を作ったが、 plot.cgi をそれと同じディレクトリに置く。
- CGI を実行するユーザ(*1)がそのディレクトリに書き込みできるようにしておく。
*1: apache だとデフォルトで nobody、 ports からインストールした場合は www、 SuEXEC を使う場合 ~user/ 以下ならその user。
便利な検索あれこれ
SQL でこんなことができるというサンプル。 何分 SQL 初心者なので、ダサイかもしれないが、許せ。 許せない場合は こっちへ。時間の指定
日付の書式は 'YYYY-mm-dd HH:MM:SS'。 シングルクォート ' を忘れないこと。 HH:MM:SS 省略時は 00:00:00 と解釈される。SELECT * FROM ipfw WHERE date >= 日付1 AND date < 日付2 ORDER BY date;
なお、SELECT * FROM ipfw WHERE date >= 日付1 AND date <= 日付2;
はSELECT * FROM ipfw WHERE date BETWEEN 日付1 AND 日付2;
とも書ける。特定のポートへのスキャンを検索する
SELECT * FROM ipfw WHERE (date >= 日付1 AND date < 日付2) AND dport = 指定ポート ORDER BY date;
特定のポートへのスキャンを検索する(複数 port)
SELECT * FROM ipfw WHERE (date >= 日付1 AND date < 日付2) AND (dport = 指定ポート1 or dport = 指定ポート2) ORDER BY date;
特定のネットワークからのアクセスを調べる
例: xx.xx.xx.xx/29 からのアクセスを調べるSELECT * FROM ipfw WHERE (date >= 日付1 AND date < 日付2) AND network(set_masklen(sip,29)) = 'xx.xx.xx.xx/29' ORDER by date;
指定 port へアクセスしてきた source IP 一覧
SELECT DISTINCT sip FROM ipfw WHERE (date >= 日付1 AND date < 日付2) AND dport = 指定ポート;
指定 port へアクセスしてきた source IP を数える
SELECT count(a.sip) FROM ( SELECT DISTINCT sip FROM ipfw WHERE (date >= 日付1 AND date < 日付2) AND dport = 指定ポート ) a;
destination port ごとの度数分布
SELECT dport, count(dport) FROM ipfw WHERE (date >= 日付1 AND date < 日付2) GROUP BY dport;
destination port ごとの度数分布 (2 回以上のものだけ)
SELECT dport, count(dport) FROM ipfw WHERE (date >= 日付1 AND date < 日付2) GROUP BY dport HAVING count(dport) >= 2;
指定 port へアクセスしてきた source IP を netmask 255.255.0.0 で分類し、度数分布を調べる
SELECT network(set_masklen(sip,16)), count(network(set_masklen(sip,16))) FROM ipfw WHERE (date >= 日付1 AND date < 日付2) AND dport = 17300 GROUP BY network(set_masklen(sip,16));