Journalisation immuable des actions des agents
Journaux en ajout uniquement, signatures cryptographiques et pistes d’audit révélant toute altération.
Journalisation immuable des actions des agents est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 1 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.
Pourquoi les agents ont-ils besoin de journaux immuables ?
Les agents prennent des actions de manière autonome — souvent sans qu'une personne vérifie chaque étape. Lorsqu'un problème survient, vous devez pouvoir répondre aux questions suivantes : qu'a fait l'agent, quand, pour le compte de qui et pourquoi ?
Un journal immuable permettant de détecter toute altération rend impossible toute modification rétroactive de la trace d'audit.
Structure d'une entrée de journal
Chaque entrée de journal contient : un horodatage, l'identité de l'agent, l'identité de l'utilisateur, le type d'action, les paramètres, le résultat et une empreinte la reliant à l'entrée précédente. Cette structure permet à la fois la recherche et la détection des altérations.
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)
Chaînage d'empreintes pour détecter les altérations
L'empreinte de chaque entrée est calculée à partir de son contenu plus l'empreinte de l'entrée précédente. Si une entrée est modifiée après coup, son empreinte change, ce qui invalide toutes les empreintes suivantes. Toute altération devient ainsi détectable.
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'])
Journal en mémoire à ajouts seuls
L'implémentation la plus simple est une liste en mémoire avec une méthode qui autorise uniquement l'ajout. Aucune méthode de suppression ou de mise à jour n'est exposée. En production, cette structure repose sur une base de données ou un stockage d'objets.
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Écrire dans PostgreSQL avec une sémantique d'ajout uniquement
En production, écrivez les journaux dans une table de base de données pour laquelle l'utilisateur de l'application ne possède aucune autorisation UPDATE ou DELETE. Révoquez ces privilèges au niveau de la base de données pour garantir un véritable fonctionnement en ajout uniquement.
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()Écrire dans S3 avec Object Lock (WORM)
AWS S3 Object Lock en mode COMPLIANCE empêche tout utilisateur — y compris le compte racine — de supprimer ou de remplacer des objets pendant la période de conservation. Il s'agit d'un véritable stockage en écriture unique pour les journaux d'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 keyInterroger le journal d'audit
Le journal d'audit ne sert pas uniquement à la conformité — il est également utile sur le plan opérationnel. Interrogez-le pour déboguer le comportement de l'agent, reconstituer une session ou répondre à la question « pourquoi l'agent a-t-il fait 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))Assainir les données sensibles avant la journalisation
Les journaux d'audit doivent enregistrer ce que l'agent a fait, sans exposer de données sensibles. Supprimez les PII (mots de passe, clés API, numéros de carte bancaire) des paramètres et des résultats avant de les écrire.
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))
Rotation et archivage des journaux
Les journaux actifs croissent indéfiniment. Effectuez leur rotation : déplacez les entrées de plus de 90 jours vers un stockage froid (par exemple, S3 Glacier) et compressez-les. Préservez l'intégrité de la chaîne en conservant la dernière empreinte de chaque segment ayant fait l'objet d'une rotation.
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}Détecter les schémas suspects
Surveillez le journal d'audit en temps réel pour repérer les schémas inhabituels : une vague d'actions à haut risque, des actions effectuées en dehors des heures ouvrées ou la répétition d'une même action plus de N fois sur une courte période.
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)
Vérifier l'intégrité du journal à la demande
Exécutez une vérification d'intégrité dans le cadre de votre chaîne de traitement de supervision. Déclenchez immédiatement une alerte si un échec de vérification est détecté — cela indique soit un bogue dans le code de journalisation, soit une tentative active d'altération.
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)Que détecte le chaînage d'empreintes dans un journal d'audit ?
Le chaînage d'empreintes est le mécanisme cryptographique qui confère au journal d'audit sa capacité à révéler toute altération. Comprendre ce qu'il détecte est fondamental pour concevoir un journal d'audit.
Récapitulatif de la journalisation immuable des actions
Les journaux immuables des actions d'un agent utilisent : le chaînage d'empreintes pour détecter les altérations, le stockage en ajout uniquement (révoquer UPDATE/DELETE au niveau de la base de données), le stockage d'objets WORM (S3 Object Lock) pour la conservation réglementaire, l'assainissement des PII avant l'écriture et les alertes d'anomalie en temps réel en cas de schémas d'actions suspects.
Questions Fréquemment Posées
La leçon « Journalisation immuable des actions des agents » est-elle gratuite ?
Oui — le texte complet de « Journalisation immuable des actions des agents » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Journalisation immuable des actions des agents » ?
Journaux en ajout uniquement, signatures cryptographiques et pistes d’audit révélant toute altération. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Agents ?
Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 1 sur 4.
Combien de temps prend la leçon « Journalisation immuable des actions des agents » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?
Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Journalisation immuable des actions des agents
- Application des politiques aux actions des agents
- Conformité réglementaire : GDPR et SOC2
- Portes d’approbation avec intervention humaine