エージェントの改ざん不能なアクションログ
追記専用ログ、暗号署名、改ざんの痕跡を検出できる監査証跡を実装します。
「エージェントの改ざん不能なアクションログ」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェントに不変ログが必要な理由
エージェントは、多くの場合、人間が各ステップを確認することなく自律的にアクションを実行します。問題が発生したときには、エージェントが何を、いつ、誰の代わりに、なぜ実行したのかを明らかにする必要があります。
不変で改ざん検知可能なログがあれば、監査証跡を後から変更することはできません。
ログエントリのスキーマ
すべてのログエントリには、タイムスタンプ、エージェントの識別情報、ユーザーの識別情報、アクション種別、パラメーター、結果、および直前のエントリと結び付けるハッシュを記録します。この構造により、検索と改ざん検知の両方が可能になります。
from dataclasses import dataclass
from typing import Any
@dataclass
class AuditEntry:
timestamp: float # Unix time (UTC)
agent_id: str
user_id: str
action_type: str
parameters: dict
result: dict
session_id: str
prev_hash: str # hash of previous entry
entry_hash: str = '' # computed after construction
if __name__ == '__main__':
entry = AuditEntry(
timestamp=1717000000.0, agent_id='agent-1', user_id='user-42',
action_type='send_email', parameters={'to': 'a@b.com'},
result={'status': 'sent'}, session_id='sess-1', prev_hash='0' * 64
)
print('Audit entry created:')
print(' agent_id:', entry.agent_id)
print(' action_type:', entry.action_type)
print(' prev_hash:', entry.prev_hash)
改ざん検知のためのハッシュチェーン
各エントリのハッシュは、その内容に加えて直前のエントリのハッシュから計算されます。後からエントリが変更されると、そのハッシュが変わり、それ以降のすべてのハッシュが無効になります。これにより、改ざんを検知できます。
import hashlib, json, time
def compute_entry_hash(entry: dict, prev_hash: str) -> str:
content = json.dumps(entry, sort_keys=True, default=str)
payload = f'{prev_hash}:{content}'
return hashlib.sha256(payload.encode()).hexdigest()
def create_log_entry(agent_id: str, user_id: str, action_type: str,
parameters: dict, result: dict,
session_id: str, prev_hash: str) -> dict:
entry = {
'timestamp': time.time(),
'agent_id': agent_id,
'user_id': user_id,
'action_type': action_type,
'parameters': parameters,
'result': result,
'session_id': session_id,
'prev_hash': prev_hash
}
entry['entry_hash'] = compute_entry_hash(
{k: v for k, v in entry.items() if k != 'entry_hash'},
prev_hash
)
return entry
if __name__ == '__main__':
entry = create_log_entry('agent-1', 'user-42', 'send_email',
{'to': 'a@b.com'}, {'status': 'sent'},
'sess-1', prev_hash='0' * 64)
print('New log entry hash:', entry['entry_hash'])
print('Chained from prev_hash:', entry['prev_hash'])
プロセス内の追記専用ログ
最も簡単な実装は、追記だけを許可するメソッドを備えたメモリ内リストです。削除や更新を行うメソッドは公開しません。実運用では、データベースまたはオブジェクトストアを利用します。
class AppendOnlyLog:
def __init__(self):
self._entries: list[dict] = []
self._last_hash = 'GENESIS'
def append(self, agent_id: str, user_id: str, action_type: str,
parameters: dict, result: dict, session_id: str) -> dict:
entry = create_log_entry(
agent_id, user_id, action_type,
parameters, result, session_id,
self._last_hash
)
self._entries.append(entry)
self._last_hash = entry['entry_hash']
return entry
def verify_integrity(self) -> bool:
running_hash = 'GENESIS'
for e in self._entries:
expected = compute_entry_hash(
{k: v for k, v in e.items() if k != 'entry_hash'},
running_hash
)
if expected != e['entry_hash']:
return False
running_hash = e['entry_hash']
return True
@property
def entries(self):
return list(self._entries) # return copy, not reference追記セマンティクスでPostgreSQLに書き込む
実運用では、アプリケーションユーザーにUPDATEまたはDELETE権限を付与していないデータベーステーブルにログを書き込みます。真の追記専用動作を実現するには、データベースレベルでこれらの権限を取り消してください。
import psycopg2, json
# Table DDL (run once):
# CREATE TABLE agent_audit_log (
# id BIGSERIAL PRIMARY KEY,
# timestamp DOUBLE PRECISION NOT NULL,
# agent_id TEXT NOT NULL,
# user_id TEXT NOT NULL,
# action_type TEXT NOT NULL,
# parameters JSONB NOT NULL,
# result JSONB NOT NULL,
# session_id TEXT NOT NULL,
# prev_hash TEXT NOT NULL,
# entry_hash TEXT NOT NULL UNIQUE
# );
# REVOKE UPDATE, DELETE ON agent_audit_log FROM app_user;
def write_to_db(conn, entry: dict):
with conn.cursor() as cur:
cur.execute(
'INSERT INTO agent_audit_log '
'(timestamp,agent_id,user_id,action_type,parameters,result,session_id,prev_hash,entry_hash) '
'VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)',
(entry['timestamp'], entry['agent_id'], entry['user_id'],
entry['action_type'], json.dumps(entry['parameters']),
json.dumps(entry['result']), entry['session_id'],
entry['prev_hash'], entry['entry_hash'])
)
conn.commit()Object Lock（WORM）を使用してS3に書き込む
COMPLIANCEモードのAWS S3 Object Lockを使用すると、アカウントのルートユーザーを含むすべてのユーザーが、保持期間中にオブジェクトを削除または上書きできなくなります。これは監査ログ向けの真のwrite-onceストレージです。
import boto3, json, time
s3 = boto3.client('s3', region_name='us-east-1')
BUCKET = 'your-audit-log-bucket-worm'
def write_to_s3_worm(entry: dict):
key = f'audit/{entry["agent_id"]}/{int(entry["timestamp"])}/{entry["entry_hash"][:16]}.json'
s3.put_object(
Bucket=BUCKET,
Key=key,
Body=json.dumps(entry).encode(),
ContentType='application/json',
ObjectLockMode='COMPLIANCE',
ObjectLockRetainUntilDate='2030-01-01T00:00:00Z' # 4+ year retention
)
return key監査ログのクエリ
監査ログはコンプライアンスのためだけのものではなく、運用上も役立ちます。エージェントの動作のデバッグ、セッションの再構成、「なぜエージェントはXを実行したのか」という疑問への回答に利用できます。
def query_session(log: AppendOnlyLog, session_id: str) -> list[dict]:
return [
e for e in log.entries
if e['session_id'] == session_id
]
def query_user_actions(log: AppendOnlyLog, user_id: str,
action_type: str = None) -> list[dict]:
entries = [e for e in log.entries if e['user_id'] == user_id]
if action_type:
entries = [e for e in entries if e['action_type'] == action_type]
return sorted(entries, key=lambda e: e['timestamp'])
def count_actions_by_type(log: AppendOnlyLog) -> dict:
from collections import Counter
return dict(Counter(e['action_type'] for e in log.entries))ログ記録前の機密データのサニタイズ
監査ログにはエージェントが行ったことを記録し、機密データを公開しないようにする必要があります。書き込む前に、パラメーターと結果からPII（パスワード、APIキー、クレジットカード番号）を除去してください。
import re
SENSITIVE_KEYS = {'password', 'api_key', 'secret', 'token', 'credit_card', 'ssn'}
def sanitize(obj, depth: int = 0) -> dict | list | str:
if depth > 5:
return '[MAX_DEPTH]'
if isinstance(obj, dict):
return {
k: '[REDACTED]' if k.lower() in SENSITIVE_KEYS
else sanitize(v, depth + 1)
for k, v in obj.items()
}
elif isinstance(obj, list):
return [sanitize(i, depth + 1) for i in obj]
elif isinstance(obj, str):
# Redact anything that looks like an API key
return re.sub(r'(sk-|Bearer\s)[A-Za-z0-9_-]{16,}', '[REDACTED]', obj)
return obj
if __name__ == '__main__':
record = {'user': 'alice', 'password': 'hunter2', 'note': 'call me at sk-abcdefghijklmnopqrstuv'}
print('Sanitized record:', sanitize(record))
ログのローテーションとアーカイブ
アクティブなログは無期限に増加します。ログをローテーションし、90日より古いエントリを低頻度アクセスストレージ（例: S3 Glacier）に移して圧縮してください。ローテーションした各セグメントの最後のハッシュを保持し、チェーンを維持します。
import time, json, gzip
def rotate_log(log: AppendOnlyLog, max_age_days: int = 90) -> dict:
cutoff = time.time() - max_age_days * 86400
archive = [e for e in log.entries if e['timestamp'] < cutoff]
remaining = [e for e in log.entries if e['timestamp'] >= cutoff]
if not archive:
return {'archived': 0, 'remaining': len(remaining)}
# Compress archive
archive_bytes = gzip.compress(json.dumps(archive).encode())
archive_file = f'/tmp/audit_archive_{int(time.time())}.json.gz'
with open(archive_file, 'wb') as f:
f.write(archive_bytes)
# Update in-memory log
log._entries = remaining
return {'archived': len(archive), 'remaining': len(remaining), 'file': archive_file}不審なパターンのアラート
監査ログをリアルタイムで監視し、通常とは異なるパターンを検出します。たとえば、高リスクのアクションが短時間に集中している、営業時間外にアクションが実行されている、同じアクションが短時間にN回を超えて繰り返されている、といったパターンです。
import time
from collections import defaultdict
HIGH_RISK_ACTIONS = {'delete_user', 'send_mass_email',
'transfer_funds', 'export_all_data'}
BURST_LIMIT = 5
BURST_WINDOW = 60 # seconds
action_timestamps: dict[str, list] = defaultdict(list)
def check_suspicious(entry: dict) -> list[str]:
alerts = []
action = entry['action_type']
if action in HIGH_RISK_ACTIONS:
alerts.append(f'HIGH_RISK_ACTION: {action} by {entry["agent_id"]}')
now = time.time()
action_timestamps[action].append(now)
recent = [t for t in action_timestamps[action] if now - t < BURST_WINDOW]
action_timestamps[action] = recent
if len(recent) > BURST_LIMIT:
alerts.append(f'ACTION_BURST: {action} called {len(recent)}x in {BURST_WINDOW}s')
return alerts
if __name__ == '__main__':
entry = {'action_type': 'delete_user', 'agent_id': 'agent-9'}
for _ in range(6):
alerts = check_suspicious(entry)
print('Alerts on 6th call:', alerts)
必要に応じたログ整合性の検証
監視パイプラインの一部として整合性チェックを実行します。検証の失敗を検出したら直ちにアラートを発生させてください。これは、ログ記録コードのバグまたは現在進行中の改ざんの試みを示しています。
def verify_and_alert(log: AppendOnlyLog) -> dict:
is_valid = log.verify_integrity()
total = len(log.entries)
result = {
'total_entries': total,
'integrity_ok': is_valid
}
if not is_valid:
import logging
alert_logger = logging.getLogger('agent.integrity_alert')
alert_logger.critical(
'AUDIT LOG INTEGRITY FAILURE: tampered or corrupted entries detected. '
'Total entries: %d. Initiating incident response.', total
)
result['alert_sent'] = True
return result
# Run periodically via cron or monitoring hook
# verify_and_alert(global_audit_log)監査ログのハッシュチェーンは何を検知するのか
ハッシュチェーンは、監査ログに改ざん検知可能という特性を与える暗号技術です。何を検知できるのかを理解することは、監査ログ設計の基本です。
不変アクションログまとめ
不変のエージェントログでは、改ざん検知のためのハッシュチェーン、追記専用ストレージ（データベースレベルでUPDATE/DELETE権限を取り消す）、規制上の保持要件に対応するWORMオブジェクトストレージ（S3 Object Lock）、書き込み前のPIIサニタイズ、および不審なアクションパターンに対するリアルタイムの異常アラートを使用します。
よくある質問
「エージェントの改ざん不能なアクションログ」レッスンは無料ですか？
はい。「エージェントの改ざん不能なアクションログ」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「エージェントの改ざん不能なアクションログ」で何を学びますか？
追記専用ログ、暗号署名、改ざんの痕跡を検出できる監査証跡を実装します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。
「エージェントの改ざん不能なアクションログ」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- エージェントの改ざん不能なアクションログ
- エージェントアクションへのポリシー適用
- 規制コンプライアンス：GDPR と SOC2
- Human-in-the-Loop 承認ゲート