Registro inmutable de acciones de agentes
Registros de solo anexión, firmas criptográficas y pistas de auditoría que evidencian manipulaciones.
Registro inmutable de acciones de agentes es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 1 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
Por qué los agentes necesitan registros inmutables
Los agentes realizan acciones de forma autónoma, a menudo sin que una persona supervise cada paso. Cuando algo sale mal, necesita responder a estas preguntas: ¿qué hizo el agente, cuándo, en nombre de quién y por qué?
Un registro inmutable y resistente a manipulaciones hace imposible modificar retrospectivamente el historial de auditoría.
Esquema de una entrada de registro
Cada entrada de registro captura: marca de tiempo, identidad del agente, identidad del usuario, tipo de acción, parámetros, resultado y un hash que la vincula con la entrada anterior. Esta estructura permite tanto realizar búsquedas como detectar manipulaciones.
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)
Encadenamiento de hashes para detectar manipulaciones
El hash de cada entrada se calcula a partir de su contenido más el hash de la entrada anterior. Si alguna entrada se modifica posteriormente, su hash cambia, lo que invalida todos los hashes posteriores. Así se pueden detectar las manipulaciones.
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'])
Registro append-only en memoria
La implementación más sencilla es una lista en memoria con un método que solo permite añadir elementos. No se exponen métodos de eliminación ni actualización. En producción, esto se respalda con una base de datos o un almacén de objetos.
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 referenceEscritura en PostgreSQL con semántica append-only
En producción, escriba los registros en una tabla de base de datos para la que el usuario de la aplicación no tenga permisos UPDATE ni DELETE. Revoque estos privilegios en el nivel de la base de datos para conseguir un comportamiento 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()Escritura en S3 con Object Lock (WORM)
AWS S3 Object Lock, con el modo COMPLIANCE, impide que cualquier usuario, incluida la cuenta raíz, elimine o sobrescriba objetos durante el periodo de retención. Esto proporciona un almacenamiento de tipo write-once para los registros de auditoría.
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 keyConsulta del registro de auditoría
El registro de auditoría no sirve únicamente para el cumplimiento: también resulta útil desde el punto de vista operativo. Consúltelo para depurar el comportamiento del agente, reconstruir una sesión o responder a la pregunta «¿por qué hizo el agente 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))Depuración de datos sensibles antes del registro
Los registros de auditoría deben capturar lo que hizo el agente, no exponer datos sensibles. Elimine la PII (contraseñas, claves de API y números de tarjetas de crédito) de los parámetros y resultados antes de escribirlos.
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))
Rotación y archivado de registros
Los registros activos crecen indefinidamente. Rótelos: traslade a un almacenamiento en frío (por ejemplo, S3 Glacier) las entradas con más de 90 días y comprímalas. Mantenga intacta la cadena conservando el último hash de cada segmento rotado.
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}Alertas sobre patrones sospechosos
Supervise el registro de auditoría en tiempo real para detectar patrones inusuales: un aumento repentino de acciones de alto riesgo, acciones fuera del horario laboral o la repetición de una misma acción más de N veces en un intervalo corto.
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)
Verificación de la integridad del registro bajo demanda
Ejecute una comprobación de integridad como parte de su canalización de supervisión. Genere una alerta inmediatamente si se detecta un fallo de verificación: indica un error en el código de registro o un intento activo de manipulación.
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)¿Qué detecta el encadenamiento de hashes en un registro de auditoría?
El encadenamiento de hashes es el mecanismo criptográfico que proporciona al registro de auditoría su propiedad de detectar manipulaciones. Comprender qué detecta es fundamental para diseñar registros de auditoría.
Resumen del registro inmutable de acciones
Los registros inmutables de agentes utilizan: encadenamiento de hashes para detectar manipulaciones, almacenamiento append-only (revocar UPDATE/DELETE en el nivel de la base de datos), almacenamiento de objetos WORM (S3 Object Lock) para la retención normativa, sanitización de PII antes de escribir y alertas de anomalías en tiempo real sobre patrones de acciones sospechosos.
Preguntas frecuentes
¿La lección «Registro inmutable de acciones de agentes» es gratis?
Sí — el texto completo de «Registro inmutable de acciones de agentes» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Registro inmutable de acciones de agentes»?
Registros de solo anexión, firmas criptográficas y pistas de auditoría que evidencian manipulaciones. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 1 de 4.
¿Cuánto tiempo toma la lección «Registro inmutable de acciones de agentes»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Registro inmutable de acciones de agentes
- Aplicación de políticas a las acciones de agentes
- Cumplimiento normativo: GDPR y SOC2
- Puntos de aprobación con supervisión humana