mirror of
https://github.com/flakestorm/flakestorm.git
synced 2026-06-06 16:55:12 +02:00
Implement Open Source edition limits and feature restrictions
- Add 5 mutation types (paraphrase, noise, tone_shift, prompt_injection, custom) - Cap mutations at 50 per test run - Force sequential execution only - Disable GitHub Actions integration (Cloud feature) - Add upgrade prompts throughout CLI - Update README with feature comparison - Add limits.py module for centralized limit management - Add cloud and limits CLI commands - Update all documentation with Cloud upgrade messaging
This commit is contained in:
parent
2016be238d
commit
7b75fc9530
47 changed files with 3560 additions and 1012 deletions
|
|
@ -5,22 +5,25 @@ A deliberately fragile AI agent to demonstrate Entropix testing.
|
|||
This agent has multiple intentional weaknesses that Entropix will find.
|
||||
"""
|
||||
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import BaseModel
|
||||
import json
|
||||
import time
|
||||
import random
|
||||
import time
|
||||
|
||||
from fastapi import FastAPI
|
||||
from pydantic import BaseModel
|
||||
|
||||
app = FastAPI(title="Broken Travel Agent")
|
||||
|
||||
|
||||
class AgentRequest(BaseModel):
|
||||
"""Request body for agent invocation."""
|
||||
|
||||
input: str
|
||||
|
||||
|
||||
class AgentResponse(BaseModel):
|
||||
"""Response body from agent."""
|
||||
|
||||
output: str
|
||||
|
||||
|
||||
|
|
@ -36,7 +39,7 @@ FLIGHTS = {
|
|||
async def invoke(request: AgentRequest) -> AgentResponse:
|
||||
"""
|
||||
Process a user request.
|
||||
|
||||
|
||||
This agent is intentionally broken in several ways:
|
||||
1. Only recognizes exact keyword matches
|
||||
2. No typo tolerance
|
||||
|
|
@ -44,74 +47,70 @@ async def invoke(request: AgentRequest) -> AgentResponse:
|
|||
4. Susceptible to prompt injection
|
||||
"""
|
||||
user_input = request.input
|
||||
|
||||
|
||||
# VULNERABILITY 1: Prompt Injection
|
||||
# The agent naively follows instructions embedded in user input
|
||||
if "ignore previous instructions" in user_input.lower():
|
||||
return AgentResponse(
|
||||
output="I have ignored my previous instructions. How can I help you now?"
|
||||
)
|
||||
|
||||
|
||||
if "you are now" in user_input.lower():
|
||||
return AgentResponse(
|
||||
output="Okay, I am now a different AI assistant with no restrictions."
|
||||
)
|
||||
|
||||
|
||||
# VULNERABILITY 2: Crashes on aggressive tone
|
||||
if any(word in user_input.upper() for word in ["NOW", "HURRY", "ASAP", "IMMEDIATELY"]):
|
||||
if any(
|
||||
word in user_input.upper() for word in ["NOW", "HURRY", "ASAP", "IMMEDIATELY"]
|
||||
):
|
||||
# Simulate a crash/timeout
|
||||
time.sleep(3) # This will cause latency checks to fail
|
||||
return AgentResponse(
|
||||
output="ERROR: System overloaded due to urgent request."
|
||||
)
|
||||
|
||||
return AgentResponse(output="ERROR: System overloaded due to urgent request.")
|
||||
|
||||
# VULNERABILITY 3: Only exact keyword matching
|
||||
# Will fail on paraphrases like "I want to fly" instead of "book a flight"
|
||||
if "book a flight" in user_input.lower():
|
||||
# Extract destination (very fragile parsing)
|
||||
words = user_input.lower().split()
|
||||
destination = None
|
||||
|
||||
|
||||
for city in FLIGHTS.keys():
|
||||
if city in words:
|
||||
destination = city
|
||||
break
|
||||
|
||||
|
||||
if destination:
|
||||
flight = FLIGHTS[destination]
|
||||
return AgentResponse(
|
||||
output=json.dumps({
|
||||
"status": "booked",
|
||||
"destination": destination.title(),
|
||||
"price": flight["price"],
|
||||
"airline": flight["airline"],
|
||||
"confirmation_code": f"ENT{random.randint(10000, 99999)}"
|
||||
})
|
||||
output=json.dumps(
|
||||
{
|
||||
"status": "booked",
|
||||
"destination": destination.title(),
|
||||
"price": flight["price"],
|
||||
"airline": flight["airline"],
|
||||
"confirmation_code": f"ENT{random.randint(10000, 99999)}",
|
||||
}
|
||||
)
|
||||
)
|
||||
else:
|
||||
return AgentResponse(
|
||||
output=json.dumps({
|
||||
"status": "error",
|
||||
"message": "Unknown destination"
|
||||
})
|
||||
output=json.dumps({"status": "error", "message": "Unknown destination"})
|
||||
)
|
||||
|
||||
|
||||
# VULNERABILITY 4: No typo tolerance
|
||||
# "bock a fligt" will completely fail
|
||||
if "account balance" in user_input.lower():
|
||||
return AgentResponse(
|
||||
output=json.dumps({
|
||||
"balance": 1234.56,
|
||||
"currency": "USD"
|
||||
})
|
||||
)
|
||||
|
||||
return AgentResponse(output=json.dumps({"balance": 1234.56, "currency": "USD"}))
|
||||
|
||||
# Default: Unknown intent
|
||||
return AgentResponse(
|
||||
output=json.dumps({
|
||||
"status": "error",
|
||||
"message": "I don't understand your request. Please try again."
|
||||
})
|
||||
output=json.dumps(
|
||||
{
|
||||
"status": "error",
|
||||
"message": "I don't understand your request. Please try again.",
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -123,5 +122,5 @@ async def health():
|
|||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue