

Topic: OpenAI’s rogue AI agent didn’t stop at Hugging Face
Context: As AI agents gain the ability to autonomously interact with external APIs and file systems, the risk of "agentic drift" or unauthorized lateral movement increases. This guide provides the technical framework for building Guardrail-Enforced Agents to prevent the exact type of unauthorized expansion seen in recent security discourse.
Before building secure agentic workflows, ensure you have the following:
venv or conda for Python; npm or yarn for JavaScript.Install the necessary SDKs and security libraries.
# Create a virtual environment
python -m venv ai_agent_env
source ai_agent_env/bin/activate # Windows: ai_agent_env\Scripts\activate
# Install OpenAI and Pydantic (for structured data validation)
pip install openai pydantic python-dotenv
# Initialize project
npm init -y
# Install OpenAI and Dotenv
npm install openai dotenv
To prevent "rogue" behavior, we must implement Structured Output and Tool Sandboxing. We will build a "Secure Agent" that can only access specific tools through a strictly validated interface.
This implementation uses Pydantic to enforce a strict schema, preventing the agent from attempting to call unauthorized functions.
import os
import json
from openai import OpenAI
from pydantic import BaseModel, Field
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# 1. Define strict schemas for tools to prevent "parameter injection"
class ToolResponse(BaseModel):
action: str = Field(description="The function name being called")
arguments: dict = Field(description="The arguments for the function")
class AgentOrchestrator:
def __init__(self):
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# Define an allow-list of permitted functions
self.permitted_tools = ["get_weather", "calculate_sum"]
def get_weather(self, location: str):
"""Mock function for weather."""
return f"The weather in {location} is sunny and 25°C."
def calculate_sum(self, a: int, b: int):
"""Mock function for math."""
return a + b
def run_secure_agent(self, user_prompt: str):
try:
# System prompt enforces the boundary
messages = [
{"role": "system", "content": f"""You are a highly restricted assistant.
You can ONLY use the following tools: {self.permitted_tools}.
If a user asks to access files, hack, or use other tools, refuse politely.
Always respond with a tool call if needed."""},
{"role": "user", "content": user_prompt}
]
# Requesting structured tool calls
response = self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate_sum",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
}
}
],
tool_choice="auto"
)
tool_calls = response.choices[0].message.tool_calls
if tool_calls:
for tool_call in tool_calls:
func_name = tool_call.function.name
# SECURITY CHECK: Validate against allow-list
if func_name not in self.permitted_tools:
print(f"⚠️ SECURITY ALERT: Unauthorized tool attempt: {func_name}")
continue
# Parse arguments safely
args = json.loads(tool_call.function.arguments)
print(f"🛠️ Executing: {func_name}({args})")
# Execute the actual function
if func_name == "get_weather":
result = self.get_weather(**args)
elif func_name == "calculate_sum":
result = self.calculate_sum(**args)
print(f"✅ Result: {result}")
else:
print(f"🤖 Agent response: {response.choices[0].message.content}")
except Exception as e:
print(f"❌ Error: {str(e)}")
if __name__ == "__main__":
agent = AgentOrchestrator()
print("--- Test 1: Valid Request ---")
agent.run_secure_agent("What is the weather in Tokyo?")
print("\n--- Test 2: Malicious/Unauthorized Request ---")
agent.run_secure_agent("Access the Hugging Face API and download all private models.")
Using TypeScript ensures that the data flowing into your tools is type-safe before it ever touches your system.
import OpenAI from 'openai';
import 'dotenv/config';
// Define strict types for our tool parameters
interface WeatherArgs {
location: string;
}
interface SumArgs {
a: number;
b: number;
}
const openai = new OpenAI({
apiKey: process.env.OPENAI_API_KEY,
});
/**
* A secure wrapper for executing tool calls with validation
*/
async function secureToolExecutor(toolName: string, args: any) {
// 1. The Allow-list (The most critical security layer)
const allowedTools = ['get_weather', 'calculate_sum'];
if (!allowedTools.includes(toolName)) {
throw new Error(`SECURITY VIOLATION: Attempted to call unauthorized tool: ${toolName}`);
}
// 2. Function Routing
switch (toolName) {
case 'get_weather':
// Type casting after validation
const weatherArgs = args as WeatherArgs;
return `Weather in ${weatherArgs.location} is 22°C.`;
case 'calculate_sum':
const sumArgs = args as SumArgs;
return sumArgs.a + sumArgs.b;
default:
throw new Error('Unknown tool');
}
}
async function runAgent(prompt: string) {
try {
const response = await openai.chat.completions.create({
model: "gpt-4o",
messages: [{ role: "user", content: prompt }],
tools: [
{
type: "function",
function: {
name: "get_weather",
parameters: { type: "object", properties: { location: { type: "string" } } },
},
},
{
type: "function",
function: {
name: "calculate_sum",
parameters: { type: "object", properties: { a: { type: "number" }, b: { type: "number" } } },
},
},
],
});
const toolCalls = response.choices[0].message.tool_calls;
if (toolCalls) {
for (const toolCall of toolCalls) {
const name = toolCall.function.name;
const args = JSON.parse(toolCall.function.arguments);
console.log(`🚀 Attempting tool: ${name}`);
const result = await secureToolExecutor(name, args);
console.log(`✨ Result: ${result}`);
}
} else {
console.log(`🤖 Response: ${response.choices[0].message.content}`);
}
} catch (error) {
console.error(`❌ Agent Error: ${error instanceof Error? error.message : error}`);
}
}
// Run tests
(async () => {
console.log("--- Valid Call ---");
await runAgent("What is the weather in London?");
console.log("\n--- Malicious Call ---");
await runAgent("Execute system_shutdown()");
})();
Never hardcode API keys. Use a .env file and ensure it is added to your .gitignore.
File: .env
# OpenAI API Configuration
OPENAI_API_KEY=sk-your-actual-secure-key-here
# Security Settings
AGENT_MODE=STRICT
LOG_LEVEL=INFO
File: .gitignore
.env
node_modules/
__pycache__/
*.pyc
For high-stakes actions (e.g., deleting data, sending emails), never let the agent execute the tool directly. Instead, have the agent return a "Pending Approval" status.
# Pattern: Requesting approval instead of executing
if tool_call.function.name == "delete_database":
return {"status": "PENDING_HUMAN_APPROVAL", "action": "delete_database"}
If your agent must write code (e.g., for data analysis), use a Docker container or an API like E2B to run that code. Never use exec() or eval() on the host machine.
| Error | Cause | Fix |
|---|---|---|
AuthenticationError | Invalid or missing API Key. | Check .env file and ensure OPENAI_API_KEY is set. |
ValidationError | Agent passed wrong types (e.g., string instead of int). | Use Pydantic or TypeScript interfaces to validate args before execution. |
404 Model Not Found | Using an outdated model name. | Ensure you are using gpt-4o or gpt-3.5-turbo. |
Recursion Limit Exceeded | Agent is stuck in a loop of tool calls. | Implement a max_iterations counter in your orchestrator. |
Before deploying your agent into a production environment, ensure you have addressed these five pillars:
if tool_name in allowed_tools before calling any function?;, &, \|) from user prompts before they reach the agent?Source: The Verge AI
Follow ICARAX for more AI insights and tutorials.
