SIEMの相関分析
原記録から検索結果、相関の根拠までをたどり、欠落や誤った結合を確認します。
説明
複数製品の記録を結ぶには、同じ人物が別のアカウント名を使う場合や、別人が同じ名前を使う場合を扱う必要があります。相関結果には元イベントの集合を残します。
認証失敗の後の成功とクラウド操作を、時刻と利用者対応表で結びます。名前だけの結合や窓の変更によって、結果がどう変わるかを確認します。
原記録から検索結果、相関の根拠までをたどり、欠落や誤った結合を確認します。
1認証とクラウド操作を保存する
認証3行に、同じ人物と別人の操作を一つずつ追加します。
from sp07 import *
from sp02 import show認証とクラウド操作を保存する
db=connect()
for i,r in enumerate(RAW[:3],1): ingest(db,r,i)
base=db.execute("SELECT MIN(t) FROM event").fetchone()[0]
db.executemany("INSERT INTO event VALUES (?,?,?,?,?)",[('cloud','c1',base+150,'a.sato','export'),('other','o1',base+160,'alice','export')])
show(db,"SELECT source,id,user,kind,t-? AS seconds FROM event ORDER BY t",(base,))authのaliceとcloudのa.satoは別名です。otherのaliceが同じ人物とは限りません。
2出所を含む対応表を読む
sourceとaccountを組で検索します。
show出所を含む対応表を読む(db,"SELECT * FROM identity ORDER BY source")auth/aliceとcloud/a.satoはp7、other/aliceはp8です。表示名だけをキーにしません。
3同じ人の失敗と成功を時間順に結ぶ
同じ認証システムで成功の前120秒にある失敗を抽出します。
query="SELECT f.id AS fail_id,s.id AS success_id,s.t-f.t AS seconds FROM event f JOIN event s ON f.source=s.source AND f.user=s.user WHERE f.kind='fail' AND s.kind='success' AND s.t-f.t BETWEEN 0 AND ? ORDER BY f.id"
show同じ人の失敗と成功を時間順に結ぶ(db,query,(120,))e1とe2がe3の前にあります。成功後の失敗や別人の失敗は含みません。
4時間幅の境界を変える
同じ検索を60秒幅へ変えます。
show時間幅の境界を変える(db,query,(60,))e2だけになります。境界を含むBETWEENなので、60秒差のe2は残ります。
5人物IDで異なるシステムの操作を結ぶ
正規化した人物IDをイベントへ付けます。
db.execute("CREATE VIEW人物IDで異なるシステムの操作を結ぶ person_event AS SELECT e.*,i.person FROM event e JOIN identity i ON e.source=i.source AND e.user=i.account")
show(db,"SELECT person,source,id,kind FROM person_event ORDER BY person,t")p7には認証3件とcloudの1件が結び付きます。otherのaliceはp8へ分かれます。
6成功後の持出し操作を検索する
同じ人物で、成功から60秒以内のexportを検索します。
show成功後の持出し操作を検索する(db,"SELECT a.id AS auth_id,x.id AS export_id,x.t-a.t AS seconds FROM person_event a JOIN person_event x USING(person) WHERE a.kind='success' AND x.kind='export' AND x.t-a.t BETWEEN 0 AND 60")e3とc1が30秒差で該当します。o1は別人なので結び付きません。検索への一致は調査候補であり、業務上の正当性は別途確認します。
7根拠のイベント集合を取得する
p7に関係する原記録へのキーを残します。
show根拠のイベント集合を取得する(db,"SELECT source,id FROM person_event WHERE person='p7' ORDER BY t")4件の出所とIDが残ります。アラートを一つの案件にまとめても、元記録や時系列を失わずに調査できます。
まとめ
主体の対応には出所とアカウントを使います。
順序と境界を含む時間条件を確認します。
集約した案件にも根拠イベントのキーを残します。
操作履歴と解説を振り返れます。
Python 3.15 / SQLite / 固定の認証ログ
1
from sp07 import *
from sp02 import show
db=connect()
for i,r in enumerate(RAW[:3],1): ingest(db,r,i)
base=db.execute("SELECT MIN(t) FROM event").fetchone()[0]
db.executemany("INSERT INTO event VALUES (?,?,?,?,?)",[('cloud','c1',base+150,'a.sato','export'),('other','o1',base+160,'alice','export')])
show(db,"SELECT source,id,user,kind,t-? AS seconds FROM event ORDER BY t",(base,))source | id | user | kind | seconds auth | e1 | alice | fail | 0 auth | e2 | alice | fail | 60 auth | e3 | alice | success | 120 cloud | c1 | a.sato | export | 150 other | o1 | alice | export | 160
2
show(db,"SELECT * FROM identity ORDER BY source")source | account | person auth | alice | p7 cloud | a.sato | p7 other | alice | p8
3
query="SELECT f.id AS fail_id,s.id AS success_id,s.t-f.t AS seconds FROM event f JOIN event s ON f.source=s.source AND f.user=s.user WHERE f.kind='fail' AND s.kind='success' AND s.t-f.t BETWEEN 0 AND ? ORDER BY f.id"
show(db,query,(120,))fail_id | success_id | seconds e1 | e3 | 120 e2 | e3 | 60
4
show(db,query,(60,))fail_id | success_id | seconds e2 | e3 | 60
5
db.execute("CREATE VIEW person_event AS SELECT e.*,i.person FROM event e JOIN identity i ON e.source=i.source AND e.user=i.account")
show(db,"SELECT person,source,id,kind FROM person_event ORDER BY person,t")person | source | id | kind p7 | auth | e1 | fail p7 | auth | e2 | fail p7 | auth | e3 | success p7 | cloud | c1 | export p8 | other | o1 | export
6
show(db,"SELECT a.id AS auth_id,x.id AS export_id,x.t-a.t AS seconds FROM person_event a JOIN person_event x USING(person) WHERE a.kind='success' AND x.kind='export' AND x.t-a.t BETWEEN 0 AND 60")auth_id | export_id | seconds e3 | c1 | 30
7
show(db,"SELECT source,id FROM person_event WHERE person='p7' ORDER BY t")source | id auth | e1 auth | e2 auth | e3 cloud | c1
演習完了