PostgreSQLでデータ領域チェックサム

タイトルからすると、誤解というか違うものを想像される方が多数だと思いますが、ここでやりたいことは、データベースのテーブルが変更されたかどうかを検出したい話です。

PostgreSQLのファイルが破損を検出するチェックサムは実装されていますが、それとは違ってSQLでUPDATEやDELETEされているかどうかを検出します。

改ざんの検出を目的にするなら、データベース監査を導入すべきとなってしまうので、そこまで必要度が高い話ではないでしょう（なので、それほど事例が無いと思います）。

実は数年前に、出来そうと思いつつ何故か実現出来なかったネタを掘り起こしました。

対象テーブル

対象となるテーブルはなるべく何でも良いようにしたいのですが、領域が区切れる主キー、または領域が区切れる変更が許可されない列（日付等）があるテーブルが条件です。

データ領域を区切らずにテーブル全体を対象とする場合は、どんなテーブルでも可能なはずです。

ここでは、以下のようなデータがある3000件のサンプルデータでidが主キーになっているテーブルを対象にします。

id	first_name	last_name	email	gender	ip_address
1	Gerhard	Southcoat	[e-mail]	Male	71.29.39.91
2	Harold	Willbond	[e-mail]	Male	147.91.208.45
3	Brigitte	Deinert	[e-mail]	Female	39.15.232.73
4	Ethel	Matchitt	[e-mail]	Female	78.224.141.243
5	Amalea	Gotter	[e-mail]	Female	94.46.252.123
6	Gabbie	Durrans	[e-mail]	Male	151.216.236.193
7	Laughton	Grece	[e-mail]	Male	11.253.117.197
...	....	...	...	...	...

件数が3000ぐらいだと全件を対象にしても良いですが、もっと件数が多くなることを想定してハッシュ値を1000件程度に分割して計算します。
そうして出来た1〜999、1000〜1999...のハッシュ値をチェックサムとして保存しておくと、1000件単位で、変更があったかどうかを検出できるようになる。という仕組みです。

id範囲	hash値
0	ec0eb3df320b584d52698f17b654095c
1000	8eba1e086b46dc06f3a476f50d58a0f6
2000	851f53b891bea45f1677e852a691f8f4
3000	1abe5c87a3559fc8f06deb8a90116647

実現方法

まず、ハッシュ値を取る関数としては、md5()やPostgreSQL 11からは SHA224,sha256,sha384,sha512と標準で入ったので、これを使えば良いことにします。

続いて、これはつまづいたのですが、全ての列を対象としてハッシュ値をとるには、全ての列を連結して、文字列型またはbytea型にする必要があります。
キャストしつつ id || first_name || last_name と列を書いていけば出来ますが、列の追加があったときに対応が面倒です。

キャストに「テーブル.*」を指定できる構文があったので、これを使います（知ってました？私は知りませんでした。）

SELECT CAST(hash_user.* AS text) FROM hash_user;

                              hash_user
----------------------------------------------------------------------
 (1,Gerhard,Southcoat,[e-mail],Male,71.29.39.91)
 (2,Harold,Willbond,[e-mail],Male,147.91.208.45)
 ...

（他の方法としてto_jsonでJSON文字列にしてしまう

SELECT to_json(hash_user.*) from hash_user;

という手もあります）。

これをmd5()の関数に渡せば、ハッシュ値がとれます。

SELECT md5(CAST(hash_user.* AS text)) FROM hash_user;

               md5  
----------------------------------
 d999cd07b8126ae6df543c23ccefa8da
 c8c109da648013695fb9914e91ef1333
 3030c618924ba960916d605a3c746740
 a338c950c897ec452e0f5193a6b95bbf
 781ebdc3647fc51910b6244e6ef34857
 bd393207e938d1a7b43d782cb0eb119f
 ...

1行1行の変更を検出したいときは、これを保存しておけば良いことになります。しかし大きなテーブルに対して同じ行のハッシュ値を保存しておくのは難しいときがあります。

そこで、行自体をstring_agg()で繋いで、一つの文字列にします。この文字列はmd5に渡す前のデータ自体でも良いですが、テーブルの列の長さによって思わぬ影響が出るかもしれないので、ハッシュ値を繋いでいきます。

SELECT string_agg(md5(CAST(hash_user.* AS text)),',') FROM hash_user;

長い文字列ができるので、これのハッシュ値を取るとテーブルのハッシュ値が取れたことになります。

SELECT md5(string_agg(md5(CAST(hash_user.* AS text)),',')) FROM hash_user;
               md5                
----------------------------------
 7d0a17b86b6bcc0fa3413cf19a19585f

テーブルのハッシュ値は更新が頻繁なテーブルでは頻繁に変わるので、あまり意味がありません。
データ領域を区切ることで、本来あまり変わらない領域が変わった時に検出できるようになります。

idを1000単位でグループ化してハッシュ値を計算します。

SELECT round(id/1000) * 1000 AS i,
       md5(string_agg(md5(cast(g.* AS text)),'|' ORDER BY id)) 
  FROM hash_user AS g GROUP BY (i) ;

以下のような結果になりました。

  i   |               md5                
------+----------------------------------
    0 | ec0eb3df320b584d52698f17b654095c
 1000 | 9b33a6ded14d2649d4dadaeab8cb03b4
 2000 | 851f53b891bea45f1677e852a691f8f4
 3000 | 1abe5c87a3559fc8f06deb8a90116647

やっぱりテーブル全体のハッシュ値も欲しい場合は、ROLLUPを追加します。

SELECT round(id/1000) * 1000 AS i,
       md5(string_agg(md5(cast(g.* AS text)),'|' ORDER BY id)) 
  FROM hash_user AS g GROUP BY ROLLUP (i) ;
  i   |               md5                
------+----------------------------------
    0 | ec0eb3df320b584d52698f17b654095c
 1000 | 9b33a6ded14d2649d4dadaeab8cb03b4
 2000 | 851f53b891bea45f1677e852a691f8f4
 3000 | 1abe5c87a3559fc8f06deb8a90116647
      | 388c1e05b5488a8cb4df537d615c5d01

これをそのままテーブルに保存しておきます。

CREATE TABLE hash AS SELECT round(id/1000) * 1000 AS i,
       md5(string_agg(md5(cast(g.* AS text)),'|' ORDER BY id)) 
  FROM hash_user AS g GROUP BY ROLLUP (i) ;

これで完了です。

変更を検知する

1件追加してみます。

INSERT INTO hash_user (id,first_name,last_name, email, gender, ip_address) VALUES (3001,'noboru','saito', '[e-mail]', 'Male', '1,2,3,4');

先程と同じSQLを流します。

SELECT round(id/1000) * 1000 AS i,
       md5(string_agg(md5(cast(g.* AS text)),'|' ORDER BY id)) 
  FROM hash_user AS g GROUP BY ROLLUP (i) ;
  i   |               md5                
------+----------------------------------
    0 | ec0eb3df320b584d52698f17b654095c
 1000 | 9b33a6ded14d2649d4dadaeab8cb03b4
 2000 | 851f53b891bea45f1677e852a691f8f4
 3000 | 011fa3e65c7078774c4afe2afc8384f5
      | f0364faf6f9b49562b93c02322c8c16e
(5 rows)

3000の行と空欄（全件）のmd5の値が変わりました。

1件DELETEしてから計算し直すと

DELETE FROM hash_user WHERE id=342;

0の行と空欄が変わりました。

  i   |               md5                
------+----------------------------------
    0 | 792e28d64e40ef81214e9219c63b62c3
 1000 | 9b33a6ded14d2649d4dadaeab8cb03b4
 2000 | 851f53b891bea45f1677e852a691f8f4
 3000 | 011fa3e65c7078774c4afe2afc8384f5
      | 32acf24bc38686ccee64b67285cc6c8a

目でチェックは辛いので、EXCEPTで比較します。

SELECT round(id/1000) * 1000 AS i,
       md5(string_agg(md5(cast(g.* AS text)),'|' ORDER BY id)) 
  FROM hash_user AS g GROUP BY ROLLUP (i) 
  EXCEPT
SELECT * FROM hash;

  i   |               md5                
------+----------------------------------
      | 32acf24bc38686ccee64b67285cc6c8a
    0 | 792e28d64e40ef81214e9219c63b62c3
 3000 | 011fa3e65c7078774c4afe2afc8384f5
(3 rows)

DELETEした0の行と追加した3000の行と空欄（全件）が変更されているのがわかりました。

おわりに

わかった後だと、「そりゃ出来るでしょ」という内容ですが、前は何故か思いつかなかったのですよね。