Skip to Content
FrameworksCrewAI

CrewAI

Static analysis for CrewAI applications to detect delegation loops, task cycles, and unbounded crew execution.

Quick Start

inkog scan ./my-crewai-app

What Inkog Detects

FindingSeverityDescription
Delegation LoopCRITICALAgent A delegates to B, B delegates back to A
Task CycleHIGHCircular task dependencies
Crew TimeoutHIGHCrew without max_rpm or iteration limits
Unsafe ToolsCRITICALAgents with unrestricted shell/file access
Verbose ExposureMEDIUMSensitive data logged in verbose mode

Delegation Loops

When agents can delegate to each other without restrictions, they can loop indefinitely.

Vulnerable
Mutual delegation creates infinite loops
from crewai import Agent, Crew, Task

researcher = Agent(
  role="Researcher",
  goal="Research topics",
  allow_delegation=True  # Can delegate to anyone
)

writer = Agent(
  role="Writer",
  goal="Write content",
  allow_delegation=True  # Can delegate back!
)

# Researcher delegates to Writer
# Writer delegates back to Researcher
# Infinite loop!
Secure
Disable delegation or use hierarchical process
from crewai import Agent, Crew, Task

researcher = Agent(
  role="Researcher",
  goal="Research topics",
  allow_delegation=False  # Cannot delegate
)

writer = Agent(
  role="Writer",
  goal="Write content",
  allow_delegation=False  # Cannot delegate
)

# Or use hierarchical process
crew = Crew(
  agents=[researcher, writer],
  tasks=[research_task, write_task],
  process=Process.hierarchical,  # Manager controls flow
  manager_llm=llm
)

Task Dependency Cycles

Circular dependencies between tasks cause deadlocks or infinite waiting.

Vulnerable
Tasks waiting on each other forever
research_task = Task(
  description="Research the topic",
  agent=researcher,
  context=[write_task]  # Depends on write_task
)

write_task = Task(
  description="Write the article",
  agent=writer,
  context=[research_task]  # Depends on research_task!
)

# Circular dependency: research ↔ write
Secure
Linear or DAG task dependencies
research_task = Task(
  description="Research the topic",
  agent=researcher
  # No dependencies - runs first
)

write_task = Task(
  description="Write the article",
  agent=writer,
  context=[research_task]  # Depends on research only
)

edit_task = Task(
  description="Edit the article",
  agent=editor,
  context=[write_task]  # Linear dependency chain
)

Crew Without Limits

Crews without rate limits or iteration bounds can run indefinitely.

Vulnerable
Unlimited API calls and execution time
crew = Crew(
  agents=[agent1, agent2],
  tasks=[task1, task2],
  verbose=True
  # No max_rpm, no iteration limits
)

# Can run forever, consuming unlimited API calls
result = crew.kickoff()
Secure
Rate limits and iteration caps at crew and agent level
crew = Crew(
  agents=[agent1, agent2],
  tasks=[task1, task2],
  verbose=False,  # Don't log sensitive data
  max_rpm=60,     # Rate limit: 60 requests/minute
  memory=True,
  cache=True      # Reduce redundant calls
)

# Per-agent limits
agent1 = Agent(
  role="Analyst",
  max_iter=10,        # Max 10 iterations per task
  max_rpm=30,         # Agent-specific rate limit
  max_execution_time=300  # 5 minute timeout
)

Unsafe Tool Access

Agents with shell or file system access can be exploited.

Vulnerable
Unrestricted shell access enables RCE
from crewai_tools import ShellTool, FileReadTool

agent = Agent(
  role="Assistant",
  tools=[
      ShellTool(),     # Can run ANY command
      FileReadTool()   # Can read ANY file
  ]
)
Secure
Allowlist commands and restrict file paths
from crewai_tools import FileReadTool
from crewai.tools import tool

@tool
def safe_shell(command: str) -> str:
  """Execute only allowed commands."""
  allowed = ["ls", "cat", "echo"]
  cmd = command.split()[0]
  if cmd not in allowed:
      return "Error: Command not allowed"
  # Additional validation...
  return subprocess.run(
      command.split(),
      capture_output=True,
      timeout=10
  ).stdout.decode()

agent = Agent(
  role="Assistant",
  tools=[
      safe_shell,
      FileReadTool(directory="./allowed_files")  # Restricted path
  ]
)

Verbose Mode Data Exposure

Verbose mode logs all agent interactions, potentially exposing secrets.

Vulnerable
Sensitive data visible in logs
crew = Crew(
  agents=[agent],
  tasks=[task],
  verbose=True  # Logs everything!
)

# Logs may contain:
# - API keys from context
# - User PII from inputs
# - Internal prompts and reasoning
Secure
Disable verbose mode or filter sensitive data
import logging

# Configure logging to filter sensitive data
class SensitiveFilter(logging.Filter):
  def filter(self, record):
      record.msg = redact_secrets(record.msg)
      return True

logging.getLogger("crewai").addFilter(SensitiveFilter())

crew = Crew(
  agents=[agent],
  tasks=[task],
  verbose=False  # Disable in production
)

Memory Without Limits

Long-term memory accumulates indefinitely without cleanup.

Vulnerable
Unbounded memory growth
crew = Crew(
  agents=[agent],
  tasks=[task],
  memory=True  # Stores all interactions forever
)

# Memory grows unbounded
# Eventually exhausts storage or context
Secure
Bounded memory with eviction policy
from crewai.memory import ShortTermMemory, LongTermMemory

# Custom memory with limits
class BoundedMemory(ShortTermMemory):
  max_items = 100

  def add(self, item):
      if len(self.storage) >= self.max_items:
          self.storage.pop(0)  # Remove oldest
      super().add(item)

crew = Crew(
  agents=[agent],
  tasks=[task],
  memory=True,
  short_term_memory=BoundedMemory()
)

Best Practices

  1. Set allow_delegation=False unless explicitly needed
  2. Use Process.hierarchical for complex multi-agent workflows
  3. Configure max_rpm at crew level (recommended: 30-60)
  4. Set max_iter per agent (recommended: 5-15)
  5. Disable verbose=True in production
  6. Restrict tool access with allowlists and path limits

CLI Examples

# Scan CrewAI project
inkog scan ./my-crewai-app
 
# Check for delegation issues
inkog scan . -severity high
 
# JSON output for automation
inkog scan . -output json > results.json
Last updated on
LangGraphAutoGen