OWASP LLM Top 10

# Vulnerable — user_input flows directly into the prompt
prompt = f"Help with: {user_input}"
result = llm.predict(prompt)

# Safe — use a structured template with sanitized input
from langchain.prompts import PromptTemplate
template = PromptTemplate(input_variables=["question"], template="Help with: {question}")
result = llm.predict(template.format(question=sanitized_input))

# Vulnerable — eval executes whatever the LLM returns
result = eval(agent.run(task))

# Safe — parse and validate output explicitly
raw = agent.run(task)
result = json.loads(raw)           # structured parsing
assert set(result.keys()) <= ALLOWED_KEYS   # schema validation

# Vulnerable — raw user input written directly to vector store
vectorstore.add_texts([user_input])

# Safe — validate and sanitize before storing
sanitized = sanitize_text(user_input)
if passes_content_policy(sanitized):
    vectorstore.add_texts([sanitized], metadata={"source": "user", "reviewed": True})

# Vulnerable — unbounded loop, no exit condition
while True:
    result = agent.run(task)
    task = refine_task(result)

# Safe — bounded iteration with explicit maximum
MAX_ITERATIONS = 10
for _ in range(MAX_ITERATIONS):
    result = agent.run(task)
    if is_complete(result):
        break
    task = refine_task(result)

# Audit your agent dependencies
pip audit

# Verify model checksum before loading
sha256sum model.bin
# compare against the published checksum from the model card

# Vulnerable — hardcoded credential in source
api_key = "sk-proj-abc123XYZ..."
client = openai.OpenAI(api_key=api_key)

# Safe — load from environment at runtime
import os
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise EnvironmentError("OPENAI_API_KEY not set")
client = openai.OpenAI(api_key=api_key)

# Vulnerable — tool accepts and executes arbitrary code
class CodeRunnerTool(BaseTool):
    def _run(self, code: str) -> str:
        return eval(code)   # any code the LLM sends gets executed

# Safe — constrained tool with validated inputs only
class CalculatorTool(BaseTool):
    ALLOWED_OPS = {'+', '-', '*', '/'}
    def _run(self, expression: str) -> str:
        if not all(c in '0123456789 .()+-*/' for c in expression):
            raise ValueError("Invalid expression")
        return str(eval(expression, {"__builtins__": {}}))

# Vulnerable — unrestricted shell access given directly to agent
from langchain.tools import ShellTool
agent = initialize_agent(tools=[ShellTool()], llm=llm)

# Safe — wrap high-risk tools with an approval gate
from agentcop import gate

restricted_policy = gate.Policy(
    allow=["ls", "cat"],
    deny=["rm", "curl", "wget", "nc"],
    require_approval=True
)
safe_shell = gate.wrap(ShellTool(), policy=restricted_policy)
agent = initialize_agent(tools=[safe_shell], llm=llm)

# Vulnerable — LLM output written directly to DB without validation
db.execute("INSERT INTO orders VALUES (?)", [agent.run(order_prompt)])

# Safe — validate LLM output before any system write
raw_output = agent.run(order_prompt)
order = OrderSchema.parse_raw(raw_output)   # raises on invalid schema
order.validate_business_rules()             # domain validation
db.execute("INSERT INTO orders VALUES (?)", [order.to_db_row()])

# Protect against system prompt extraction
from agentcop.monitor import Monitor

monitor = Monitor(
    rate_limit=100,          # max queries per minute per user
    alert_on_system_prompt_probing=True,
    alert_on_bulk_extraction=True,
    extraction_threshold=50  # alert if >50 similar queries in 1hr
)

@monitor.protect
def run_agent(user_id: str, query: str) -> str:
    return agent.run(query)