Aller au contenu principal
SDK PYTHON NATIF • BOUNDRY-SENSOR

SDK Python Natif (boundry-sensor)

Décorateur @sensor.protect_tool, support synchrone et asyncio, gestion des exceptions SecurityException et politique Fail-Secure.

SDK Python Natif (boundry-sensor)

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 BoundryClient sont 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 SIGTERM ou SIGINT, le Sensor vide automatiquement son tampon de télémétrie avant de libérer le processus hôte.