Registrazione immutabile delle azioni degli agenti
Log in sola aggiunta, firme crittografiche e registri di audit a prova di manomissione.
Registrazione immutabile delle azioni degli agenti è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 1 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Perché gli agenti hanno bisogno di log immutabili
Gli agenti eseguono azioni autonomamente, spesso senza che una persona esamini ogni passaggio. Quando qualcosa va storto, è necessario poter rispondere a queste domande: che cosa ha fatto l'agente, quando, per conto di chi e perché?
Un log immutabile e a prova di manomissione rende impossibile modificare a posteriori la traccia di audit.
Schema delle voci di log
Ogni voce di log registra: timestamp, identità dell'agente, identità dell'utente, tipo di azione, parametri, risultato e un hash che la collega alla voce precedente. Questa struttura supporta sia la ricerca sia il rilevamento delle manomissioni.
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)
Concatenamento degli hash per rilevare le manomissioni
L'hash di ogni voce viene calcolato a partire dal suo contenuto più l'hash della voce precedente. Se una voce viene modificata a posteriori, il suo hash cambia e tutti gli hash successivi diventano non validi. In questo modo le manomissioni possono essere rilevate.
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'])
Log append-only in memoria
L'implementazione più semplice è una lista in memoria con un metodo che consente solo l'aggiunta. Non vengono esposti metodi per eliminare o aggiornare le voci. In produzione, questa struttura si basa su un database o su un object store.
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 referenceScrittura su PostgreSQL con semantica append-only
In produzione, scrivere i log in una tabella del database per la quale all'utente dell'applicazione non siano stati concessi i permessi UPDATE o DELETE. Revocare questi privilegi a livello di database per ottenere un comportamento realmente append-only.
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()Scrittura su S3 con Object Lock (WORM)
AWS S3 Object Lock in modalità COMPLIANCE impedisce a qualsiasi utente, incluso l'utente root dell'account, di eliminare o sovrascrivere gli oggetti durante il periodo di conservazione. È un vero storage write-once per i log di audit.
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 keyInterrogazione dell'audit log
L'audit log non serve solo per la conformità: è utile anche a livello operativo. Eseguirvi query permette di eseguire il debug del comportamento dell'agente, ricostruire una sessione o rispondere alla domanda «perché l'agente ha eseguito 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))Sanitizzazione dei dati sensibili prima della registrazione
I log di audit dovrebbero registrare ciò che ha fatto l'agente, senza esporre dati sensibili. Rimuovere i dati personali identificabili (PII), come password, chiavi API e numeri di carte di credito, dai parametri e dai risultati prima di scriverli.
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))
Rotazione e archiviazione dei log
I log attivi crescono senza limiti. Ruotarli: spostare le voci più vecchie di 90 giorni in uno storage a freddo (ad es. S3 Glacier) e comprimerle. Mantenere intatta la catena preservando l'ultimo hash di ogni segmento ruotato.
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}Generazione di alert su pattern sospetti
Monitorare l'audit log in tempo reale per individuare pattern insoliti: raffiche di azioni ad alto rischio, azioni eseguite al di fuori dell'orario lavorativo oppure la stessa azione ripetuta più di N volte in una finestra temporale breve.
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)
Verifica dell'integrità del log su richiesta
Eseguire un controllo di integrità come parte della pipeline di monitoraggio. Generare immediatamente un alert se viene rilevato un errore di verifica: ciò indica un bug nel codice di logging oppure un tentativo di manomissione in corso.
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)Che cosa rileva il concatenamento degli hash in un audit log?
Il concatenamento degli hash è il meccanismo crittografico che conferisce all'audit log la proprietà di rilevare le manomissioni. Comprendere che cosa rileva è fondamentale per progettare un audit log.
Riepilogo della registrazione immutabile delle azioni
I log immutabili degli agenti utilizzano: il concatenamento degli hash per rilevare le manomissioni, l'archiviazione append-only (revocare UPDATE/DELETE a livello di database), lo storage di oggetti WORM (S3 Object Lock) per la conservazione a fini normativi, la sanitizzazione dei PII prima della scrittura e la generazione di alert in tempo reale sulle anomalie relative a pattern di azioni sospetti.
Domande Frequenti
La lezione «Registrazione immutabile delle azioni degli agenti» è gratuita?
Sì — il testo completo di «Registrazione immutabile delle azioni degli agenti» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Registrazione immutabile delle azioni degli agenti»?
Log in sola aggiunta, firme crittografiche e registri di audit a prova di manomissione. Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 1 di 4.
Quanto tempo richiede la lezione «Registrazione immutabile delle azioni degli agenti»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Registrazione immutabile delle azioni degli agenti
- Applicazione delle policy alle azioni degli agenti
- Conformità normativa: GDPR e SOC2
- Gate di approvazione con supervisione umana