에이전트를 위한 변경 불가 작업 기록
추가 전용 기록, 암호화 서명, 변조 감지 감사 추적을 구현합니다.
에이전트를 위한 변경 불가 작업 기록은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 1번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 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은 계정 루트를 포함한 어떤 사용자도 보존 기간 동안 객체를 삭제하거나 덮어쓰지 못하게 합니다. 이는 감사 로그를 위한 진정한 일회성 기록 저장소입니다.
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감사 로그 조회하기
감사 로그는 COMPLIANCE를 위한 용도에만 쓰이지 않고 운영 측면에서도 유용합니다. 이를 조회하여 에이전트 동작을 디버깅하거나, 세션을 재구성하거나, ‘에이전트가 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)
필요할 때 로그 무결성 확인하기
모니터링 pipeline의 일부로 무결성 확인을 실행합니다. 확인 실패가 감지되면 즉시 경고를 발생시켜야 합니다. 이는 로깅 코드의 버그이거나 실제 변조 시도임을 나타냅니다.
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/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“에이전트를 위한 변경 불가 작업 기록”에서 뭘 배우나요?
추가 전용 기록, 암호화 서명, 변조 감지 감사 추적을 구현합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 1번째 강의입니다.
“에이전트를 위한 변경 불가 작업 기록” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 에이전트를 위한 변경 불가 작업 기록
- 에이전트 작업에 대한 정책 적용
- 규제 준수: GDPR 및 SOC2
- 사람 참여형 승인 관문