Pencatatan Tindakan Agen yang Tidak Dapat Diubah

Mengapa Agen Memerlukan Catatan yang Tidak Dapat Diubah

Skema Entri Catatan

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)

Rantai Ringkasan Kriptografis untuk Mendeteksi Perubahan

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'])

Catatan Hanya-Penambahan dalam Proses

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

Menulis ke PostgreSQL dengan Perilaku Hanya-Penambahan

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()

Menulis ke S3 dengan Penguncian Objek (WORM)

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

Membuat Kueri pada Catatan Audit

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))

Membersihkan Data Sensitif Sebelum Pencatatan

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))

Rotasi dan Pengarsipan Catatan

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}

Memberi Peringatan tentang Pola Mencurigakan

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)

Memverifikasi Integritas Catatan Sesuai Kebutuhan

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)

Apa yang dideteksi rantai ringkasan kriptografis dalam catatan audit?

Ringkasan Pencatatan Tindakan yang Tidak Dapat Diubah