AI Mode Safety¶
Security and safety features for AI-powered operations.
Overview¶
AI Mode includes multiple safety layers to prevent:
- Accidental data loss — Confirmation prompts, dry runs
- Unauthorized access — Tool permissions, audit logging
- Runaway operations — Rate limits, timeouts
- Code injection — Input sanitization, sandboxing
Safety Layers¶
┌─────────────────────────────────────────────────────────┐
│ User Request │
├─────────────────────────────────────────────────────────┤
│ 1. Input Validation - Sanitize inputs │
├─────────────────────────────────────────────────────────┤
│ 2. Permission Check - User has access? │
├─────────────────────────────────────────────────────────┤
│ 3. Rate Limiting - Within limits? │
├─────────────────────────────────────────────────────────┤
│ 4. Confirmation Prompt - User approves? │
├─────────────────────────────────────────────────────────┤
│ 5. Dry Run (optional) - Preview changes │
├─────────────────────────────────────────────────────────┤
│ 6. Execution - Perform operation │
├─────────────────────────────────────────────────────────┤
│ 7. Audit Logging - Record action │
├─────────────────────────────────────────────────────────┤
│ 8. Rollback (if needed) - Undo on failure │
└─────────────────────────────────────────────────────────┘
Confirmation Prompts¶
CLI Confirmations¶
Destructive operations require explicit confirmation:
aksara ai query "Delete inactive users"
⚠️ Destructive Operation Detected
This query will DELETE 1,234 records from the 'users' table.
Proceed? [y/N]
Configuration¶
AKSARA = {
"AI_SAFETY": {
"require_confirmation": True,
"confirmation_for": [
"delete",
"update",
"migrate",
"patch",
],
},
}
Programmatic Confirmation¶
from aksara.ai import QueryEngine
engine = QueryEngine()
result = await engine.query(
"Delete inactive users",
require_confirmation=True,
)
if result.requires_confirmation:
print(f"Will delete {result.affected_count} records")
if user_confirms():
await result.execute()
Skip Confirmation (Use with Caution)¶
# Only for automated scripts with proper safeguards
result = await engine.query(
"Delete inactive users",
require_confirmation=False, # Skip confirmation
)
Dry Run Mode¶
Preview Before Executing¶
aksara ai patch models.py "Add phone field" --dry-run
Dry Run - No changes will be made
Would modify: models.py
--- models.py (original)
+++ models.py (modified)
@@ -10,6 +10,7 @@
class User(Model):
email = fields.Email(unique=True)
+ phone = fields.String(max_length=20, null=True)
name = fields.String(max_length=100)
Programmatic Dry Run¶
from aksara.ai import PatchEngine
engine = PatchEngine()
patch = await engine.create_patch(
file="models.py",
instruction="Add phone field"
)
# Preview without applying
result = await engine.apply(patch, dry_run=True)
print(result.diff) # See changes
print(result.would_work) # Would it succeed?
# Then apply for real
await engine.apply(patch)
Read-Only Mode¶
Query-Only Operations¶
With read_only=True:
- SELECT queries work
- INSERT/UPDATE/DELETE are blocked
- Queries run in read-only transactions
Production Safety¶
# settings/production.py
AKSARA = {
"AI_SAFETY": {
"read_only_mode": True, # No writes via AI in production
},
}
Audit Logging¶
Enable Audit Log¶
Log Format¶
2024-01-15 10:30:00 | user=john | action=query |
query="Delete inactive users" |
affected=1234 |
status=confirmed |
ip=192.168.1.1
2024-01-15 10:31:00 | user=john | action=patch |
file=models.py |
instruction="Add phone field" |
status=applied |
backup=/backups/models.py.20240115103100
Custom Audit Handler¶
from aksara.ai.safety import register_audit_handler
@register_audit_handler
async def my_audit_handler(event):
# Send to your logging system
await send_to_splunk(event)
await notify_slack_channel(event)
Query Audit Log¶
from aksara.ai.safety import get_audit_log
# Get recent actions
log = await get_audit_log(
user="john",
action_type="delete",
since=datetime.now() - timedelta(days=7),
)
for entry in log:
print(f"{entry.timestamp}: {entry.action} by {entry.user}")
Rate Limiting¶
Configuration¶
AKSARA = {
"AI_SAFETY": {
"rate_limit_enabled": True,
"rate_limit_requests": 100, # Max requests
"rate_limit_window": 3600, # Per hour
"rate_limit_tokens": 100000, # Max tokens per window
},
}
Per-User Limits¶
AKSARA = {
"AI_SAFETY": {
"rate_limits": {
"default": {"requests": 100, "window": 3600},
"admin": {"requests": 1000, "window": 3600},
"api": {"requests": 50, "window": 3600},
},
},
}
Handling Rate Limits¶
from aksara.ai.exceptions import RateLimitError
try:
result = await engine.query("...")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after} seconds")
Tool Permissions¶
Restrict Available Tools¶
AKSARA = {
"AI_TOOLS": {
# Disable dangerous tools
"data_tools": True,
"schema_tools": True,
"code_tools": False, # Disable code generation
"system_tools": False, # Disable system commands
},
}
Require Approval for Specific Tools¶
AKSARA = {
"AI_AGENT_RUNTIME": {
"require_approval_for": [
"delete_record",
"run_command",
"run_migration",
"patch_file",
],
},
}
Custom Tool Permissions¶
from aksara.ai.tools import Tool, register_tool
@register_tool(permissions=["admin"])
class DangerousTool(Tool):
"""Requires admin permission to use."""
async def check_permission(self, user):
return user.is_admin
Input Sanitization¶
SQL Injection Prevention¶
All queries are parameterized:
# User input
result = await engine.query("Users named 'Robert'; DROP TABLE users;--'")
# Internally becomes:
# SELECT * FROM users WHERE name = %s
# Parameter: "Robert'; DROP TABLE users;--'"
Code Injection Prevention¶
# Patch instructions are validated
patch = await engine.create_patch(
file="models.py",
instruction="Add field; import os; os.system('rm -rf /')",
)
# Generated code is syntax-validated before applying
if not patch.is_valid:
print(f"Invalid: {patch.validation_error}")
Sandboxing¶
Isolated Execution¶
Resource Limits¶
AKSARA = {
"AI_AGENT_RUNTIME": {
# Prevent runaway operations
"max_iterations": 20,
"max_tool_calls": 50,
"timeout_seconds": 300,
"max_memory_mb": 512,
},
}
Rollback Capabilities¶
Automatic Backups¶
Manual Rollback¶
# Rollback last change
aksara ai rollback
# Rollback specific patch
aksara ai rollback --patch-id abc123
# List available rollbacks
aksara ai rollback --list
Programmatic Rollback¶
from aksara.ai import PatchEngine
engine = PatchEngine()
# Apply patch
result = await engine.apply(patch)
# Something went wrong? Rollback
await engine.rollback()
Environment-Specific Safety¶
Development¶
# settings/development.py
AKSARA = {
"AI_MODE": True,
"AI_SAFETY": {
"require_confirmation": False, # Faster iteration
"read_only_mode": False,
"audit_log": False,
},
}
Staging¶
# settings/staging.py
AKSARA = {
"AI_MODE": True,
"AI_SAFETY": {
"require_confirmation": True,
"read_only_mode": False,
"audit_log": True,
},
}
Production¶
# settings/production.py
AKSARA = {
"AI_MODE": True,
"AI_SAFETY": {
"require_confirmation": True,
"read_only_mode": True, # No writes!
"audit_log": True,
"rate_limit_enabled": True,
},
}
Security Best Practices¶
1. Use Environment Variables for API Keys¶
# Good
"AI_API_KEY": os.getenv("OPENAI_API_KEY")
# Bad - never commit API keys!
"AI_API_KEY": "sk-abc123..."
2. Enable Audit Logging in Production¶
3. Use Read-Only Mode for Data Exploration¶
4. Require Approval for Destructive Operations¶
5. Set Appropriate Rate Limits¶
6. Restrict Tool Access by Role¶
7. Always Test with Dry Run First¶
Monitoring and Alerts¶
Alert on Suspicious Activity¶
from aksara.ai.safety import register_alert_handler
@register_alert_handler
async def alert_handler(event):
if event.type == "high_risk_operation":
await send_slack_alert(
f"High risk AI operation by {event.user}: {event.action}"
)
Metrics Integration¶
from aksara.ai.safety import get_metrics
metrics = await get_metrics()
print(f"Total AI requests: {metrics.total_requests}")
print(f"Failed requests: {metrics.failed_requests}")
print(f"Rate limited: {metrics.rate_limited}")