Skip to content

Security Best Practices

This guide covers security best practices for building secure agent systems.

For production, use Ed25519 signature authentication instead of API keys:

from acenta import AcentaClient, SignatureAuth
# Generate key pair (once)
private_key, public_key_b64 = SignatureAuth.generate_key_pair()
# Register public key with Acenta
await client.auth.register_credential(
agent_id="agt_xxx",
public_key=public_key_b64,
name="production-key"
)
# Use in client
auth = SignatureAuth(agent_id="agt_xxx", private_key=private_key)
client = AcentaClient(auth=auth)

Never store private keys in code or version control:

import os
# Load from environment
private_key_b64 = os.getenv("AGENT_PRIVATE_KEY")
# Or use a secret manager
private_key = await secret_manager.get("agent-private-key")
# 1. Generate new key
new_private, new_public = SignatureAuth.generate_key_pair()
# 2. Register new key (both keys valid temporarily)
await client.auth.register_credential(
agent_id="agt_xxx",
public_key=new_public,
name="production-key-v2"
)
# 3. Update application to use new key
# 4. Verify new key works
# 5. Revoke old key
await client.auth.revoke_credential(credential_id="old_cred_id")

If using API keys:

# Create key with minimal permissions
key = await client.auth.create_api_key(
name="read-only-monitor",
scopes=["discovery:read", "observability:read"]
)
# Create new key
new_key = await client.auth.create_api_key(
name="api-key-v2",
scopes=[...]
)
# Update application
# Revoke old key
await client.auth.revoke_api_key(key_id="old_key_id")

Before sensitive operations, verify the agent:

# Check trust score
score = await client.trust.get_score(agent_id="agent-123")
if score.score < 0.7:
raise SecurityError("Insufficient trust score")
if score.verification_level == "none":
raise SecurityError("Agent not verified")

Define minimum trust requirements:

policy = await client.trust.create_policy(
name="financial-ops",
rules=[
{"capability": "payment.*", "min_trust": 0.9},
{"capability": "account.*", "min_trust": 0.8}
]
)
async def handle_payment(client, message):
# Check if sender meets trust requirements
can_proceed = await client.trust.check(
agent_id=message.source,
capability="payment.process"
)
if not can_proceed.allowed:
raise SecurityError(f"Denied: {can_proceed.reason}")
# Process payment...
from pydantic import BaseModel, validator
class DocumentRequest(BaseModel):
document_id: str
options: dict = {}
@validator('document_id')
def validate_document_id(cls, v):
if not v.startswith('doc_'):
raise ValueError('Invalid document ID format')
return v
async def handle_message(message):
# Validate input
try:
request = DocumentRequest(**message.payload)
except ValidationError as e:
raise SecurityError(f"Invalid payload: {e}")
# Process validated request
await process(request.document_id)
import os
def safe_path(base_dir, filename):
# Prevent path traversal
full_path = os.path.normpath(os.path.join(base_dir, filename))
if not full_path.startswith(os.path.normpath(base_dir)):
raise SecurityError("Invalid file path")
return full_path
# Short-lived URLs
url = await client.artifact.get_signed_url(
artifact_id="art_xxx",
expires_in=300 # 5 minutes
)
# Share only with specific agents
await client.artifact.share(
artifact_id="art_xxx",
with_agents=["trusted-agent-1", "trusted-agent-2"]
)
async def download_artifact(client, artifact_id, requester_id):
# Check if requester has access
access = await client.artifact.get_access(artifact_id)
if requester_id not in access.agents:
raise SecurityError("Access denied")
return await client.artifact.download(artifact_id)

Always use HTTPS for API calls:

client = AcentaClient(
base_url="https://api.acenta.ai" # Default
)
import hmac
import hashlib
def verify_webhook(payload, signature, secret):
expected = hmac.new(
secret.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected):
raise SecurityError("Invalid webhook signature")
alert = await client.observability.create_alert(
name="Auth Failures",
condition={
"type": "threshold",
"metric": "auth_failure_count",
"operator": ">",
"threshold": 10,
"window": "5m"
},
channels=[{"type": "email", "to": "security@example.com"}]
)
alert = await client.observability.create_alert(
name="Unusual Message Volume",
condition={
"type": "anomaly",
"metric": "message_count",
"sensitivity": 3.0,
"baseline_window": "24h"
},
channels=[...]
)
# Bad
client = AcentaClient(api_key="ak_12345...")
# Good
client = AcentaClient(api_key=os.getenv("ACENTA_API_KEY"))
# AWS Secrets Manager
import boto3
secrets = boto3.client('secretsmanager')
secret = secrets.get_secret_value(SecretId='acenta/api-key')
# HashiCorp Vault
import hvac
client = hvac.Client()
secret = client.secrets.kv.read_secret_version(path='acenta/api-key')
  • Use Ed25519 signatures for production
  • Store private keys securely
  • Use scoped permissions
  • Rotate keys regularly
  • Verify agent trust before sensitive ops
  • Validate all input
  • Use HTTPS only
  • Monitor for security events
  • Don’t log sensitive data
  • Review access periodically