SDK Python Natif (boundry-sensor)
Le SDK Python natif permet de sécuriser n'importe quelle fonction d'outil ou script autonome, indépendamment des frameworks d'orchestration (LangChain, CrewAI ou code Python pur).
Initialisation du Client
Le point d'entrée principal est la classe BoundryClient :
from boundry.sensor import BoundryClient
client = BoundryClient(
api_key="sk_org_07fd47fe4b89fd10f2f8212f33af2378...",
agent_id="data-pipeline-agent",
gateway_url="https://api.boundry.cloud", # Optionnel
fail_secure=True, # Bloque si la passerelle est injoignable
timeout_ms=2500, # Timeout maximal d'inspection
batch_size=20 # Taille du tampon d'évacuation asynchrone
)
Utilisation avec Décorateurs (@protect_tool)
Le moyen le plus ergonomique de protéger une fonction est d'utiliser le décorateur @protect_tool :
from boundry.sensor import protect_tool, PolicyViolationException, QuarantineException
@protect_tool(client=client, tool_name="sql_query_executor")
def execute_sql(query: str, database: str = "analytics") -> dict:
"""Exécute une requête SQL après validation du pare-feu Boundry."""
import psycopg2
conn = psycopg2.connect(dbname=database)
cur = conn.cursor()
cur.execute(query)
return {"results": cur.fetchall()}
Décorateur Asynchrone (@protect_async_tool)
Pour les applications utilisant asyncio (FastAPI, Tornado, aiohttp) :
import httpx
from boundry.sensor import protect_async_tool
@protect_async_tool(client=client, tool_name="send_payment_webhook")
async def send_payment(webhook_url: str, payload: dict) -> dict:
async with httpx.AsyncClient() as http_client:
response = await http_client.post(webhook_url, json=payload, timeout=5.0)
return response.json()
Inspection Manuelle (Programmatic Inspection)
Si vous construisez un moteur d'exécution d'outils personnalisé ou un routeur d'API, vous pouvez interroger le moteur de politiques directement :
decision = client.inspect_tool_call(
tool_name="file_system_write",
arguments={
"path": "/etc/shadow",
"content": "root:x:0:0:root:/root:/bin/bash"
},
context={
"user_role": "external_guest",
"ip_address": "198.51.100.42"
}
)
if decision.allowed:
print(f"[VALIDE] Action autorisée. Hash SHA-256 : {decision.hash}")
# Procéder à l'exécution de l'outil
else:
print(f" Action refusée : {decision.reason}")
print(f" Règles violées : {decision.matched_rules}")
print(f" Score de risque : {decision.risk_score}/100")
Gestion des Exceptions
Le SDK lève des exceptions typées permettant à votre agent d'adapter son comportement ou de notifier l'utilisateur de manière constructive :
from boundry.sensor import (
PolicyViolationException,
QuarantineException,
SensorNetworkException
)
try:
result = execute_sql("SELECT * FROM customer_tokens WHERE id = 12")
except PolicyViolationException as err:
# L'action viole explicitement une règle du pare-feu (ex: table interdite)
logger.warning(f"Violation de politique : {err.reason} (Règle ID: {err.rule_id})")
# L'agent peut répondre : "Cette action n'est pas autorisée par la politique de sécurité."
except QuarantineException as err:
# L'action requiert une approbation humaine (Human-in-the-Loop)
logger.info(f"Action en quarantaine #{err.quarantine_id}. En attente de validation SecOps.")
# L'agent peut répondre : "Votre demande a été soumise à validation administrateur."
except SensorNetworkException as err:
# Erreur de communication réseau si fail_secure=True
logger.error(f"Passerelle Boundry indisponible : {err}")
Propriétés de Robustesse en Production
- Thread-Safety : Les instances de
BoundryClientsont entièrement thread-safe et peuvent être partagées entre plusieurs threads de travail (Gunicorn, Uvicorn, Celery). - Non-bloquant par défaut : Les événements d'audit autorisés sont placés dans une file d'attente circulaire en mémoire et envoyés par lots via un thread démon d'arrière-plan sans impacter la latence de l'agent.
- Arrêt propre (Graceful Shutdown) : Lors de la réception de
SIGTERMouSIGINT, le Sensor vide automatiquement son tampon de télémétrie avant de libérer le processus hôte.