Use when adding API endpoints, handling user input, touching secrets/env vars, or adding shell/file tools. Security checklist for Python FastAPI + Vikarma agent.
server/main.pyVikarmaToolGatewayNexusBridge# NEVER
ANTHROPIC_API_KEY = "sk-ant-xxxx"
# ALWAYS
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("ANTHROPIC_API_KEY not set")
os.getenv().env in .gitignore.env.example documents all required keysThe shell tool executes arbitrary commands. Guard it:
# NEVER pass user input directly to shell
await gw.execute("shell", {"command": user_input}) # INJECTION RISK
# ALWAYS validate or whitelist commands
ALLOWED_COMMANDS = {"ls", "pwd", "echo", "date"}
cmd = user_input.split()[0]
if cmd not in ALLOWED_COMMANDS:
return {"error": f"Command not allowed: {cmd}"}
VikarmaToolGateway._resolve() must anchor all relative paths:
def _resolve(self, path: str) -> Path:
p = Path(path)
if p.is_absolute():
return p
return self.workspace / p # anchored to workspace
_resolve()../../../etc/passwd style traversal possible~, not /Only math module globals allowed, no builtins:
safe_globals = {"__builtins__": {}, "math": math}
safe_globals.update({k: getattr(math, k) for k in dir(math) if not k.startswith("_")})
value = eval(expression, safe_globals)
__builtins__ is {} (empty), not the real builtinsmath.* functions exposedimport, open, exec, __import__ possible# NEVER trust raw request body
@app.post("/chat")
async def chat(request: Request):
body = await request.json() # unvalidated!
msg = body["message"] # KeyError risk
# ALWAYS use Pydantic models
from pydantic import BaseModel, validator
class ChatRequest(BaseModel):
message: str
provider: str = "claude"
@validator("provider")
def validate_provider(cls, v):
allowed = {"claude", "openai", "gemini", "grok", "deepseek", "qwen"}
if v not in allowed:
raise ValueError(f"Unknown provider: {v}")
return v
@app.post("/chat")
async def chat(req: ChatRequest):
...
# NEVER expose internals
except Exception as e:
return {"error": str(e), "traceback": traceback.format_exc()} # leaks internals
# ALWAYS generic message to caller, log internally
except Exception as e:
logger.error(f"Tool {tool} failed: {e}", exc_info=True)
return {"error": "Tool execution failed", "tool": tool}
# KANMemory stores to disk — validate before storing
def remember_fact(self, key: str, value: Any, ...) -> dict:
if not key or len(str(key)) > 500:
return {"error": "Invalid key"}
if len(str(value)) > 10_000:
return {"error": "Value too large"}
~/.vikarma/memory/, not project root# Prevent SSRF — block internal/private addresses
import ipaddress
def _is_safe_url(url: str) -> bool:
from urllib.parse import urlparse
host = urlparse(url).hostname
try:
ip = ipaddress.ip_address(host)
return not (ip.is_private or ip.is_loopback)
except ValueError:
return True # hostname, not IP — allow DNS resolution
web_fetch doesn't hit localhost/127.0.0.1/internal# Run before every commit
grep -rn "api_key\s*=\s*['\"]" server/ --include="*.py" | grep -v "os.getenv\|environ\|test"
grep -rn "password\s*=\s*['\"]" server/ --include="*.py"
grep -rn "secret\s*=\s*['\"]" server/ --include="*.py"
Expected output: nothing. Any match = BLOCKER.