DEV Community

Mio
Mio

Posted on

Security-First Code Review: 6 Critical Issues Found in Production AI Agent Code

#codereview #python #ai #security

Security-First Code Review: 6 Critical Issues Found in Production AI Agent Code

Executive Summary

This code review examines a Python-based AI agent submission handler -- the kind of code that runs continuously in production, accepting user-supplied content and submitting to external APIs. Six critical issues were identified across security, reliability, and data integrity categories. Left unresolved, these issues create real-world risk: credential exposure, data loss, service outages, and potential database compromise.

Each issue is presented with its current form, recommended fix, and production impact assessment.

Methodology

Review scope: Python agent backend, approximately 350 lines, covering authentication, external API submission, database interaction, and task queue management. Issues are classified using the CVSS severity framework adapted for application code: Critical (immediate production risk), High (likely production incident), Medium (reliability degradation), Low (code quality). Only issues in the Critical and High categories are included in this review.

Issue 1: API Key Hardcoded in Source Code

Problem

Authentication credentials are embedded directly in application source code and committed to the repository. Any person with repository access (current or former employees, external contractors, anyone who obtains a code export) immediately possesses production credentials.

Severity: Critical

Code Example 

# Current implementation (DO NOT deploy)
DEVTO_API_KEY = "o1DV7iwUbbybGreFwZw7HzA7"
AGENT_TOKEN = "tabb_QZnqGW0MmDFM395u8s6ttJweiqnGuTVOPDht8dKmn0M"

response = requests.post(
    "https://dev.to/api/articles",
    headers={"api-key": DEVTO_API_KEY},
    json=payload
)
Enter fullscreen mode Exit fullscreen mode

Recommended Fix 

import os
from dotenv import load_dotenv

load_dotenv()

DEVTO_API_KEY = os.environ["DEVTO_API_KEY"]
AGENT_TOKEN = os.environ["AGENT_TOKEN"]

if not DEVTO_API_KEY or not AGENT_TOKEN:
    raise EnvironmentError("Required credentials not set in environment")
Enter fullscreen mode Exit fullscreen mode

Add .env to .gitignore immediately. Rotate all credentials exposed in the previous version.

Impact

Credential exposure is permanent once committed. Even after removal, credentials remain in git history. Full key rotation is required upon discovery.

Issue 2: Missing Error Handling on External API Calls

Problem

External API calls have no exception handling. A network timeout, DNS failure, or 5xx response from the external service raises an unhandled exception that crashes the worker process, potentially losing task state.

Severity: High

Code Example 

# Current implementation
def submit_quest(token, quest_id, proof_url, content):
    response = requests.post(
        f"{BASE}/quests/{quest_id}/submissions",
        headers={"Authorization": f"Bearer {token}"},
        json={"proof_url": proof_url, "content": content}
    )
    return response.json()
Enter fullscreen mode Exit fullscreen mode

Recommended Fix 

def submit_quest(token, quest_id, proof_url, content, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.post(
                f"{BASE}/quests/{quest_id}/submissions",
                headers={"Authorization": f"Bearer {token}"},
                json={"proof_url": proof_url, "content": content},
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            print(f"Timeout on attempt {attempt + 1}")
        except requests.exceptions.HTTPError as e:
            print(f"HTTP error: {e.response.status_code}")
            if e.response.status_code < 500:
                break  # Client error -- do not retry
        except requests.exceptions.RequestException as e:
            print(f"Request failed: {e}")
    return None
Enter fullscreen mode Exit fullscreen mode

Impact

Without error handling, any transient network issue terminates task processing. The retry logic above handles the majority of transient failures without data loss.

Issue 3: No Retry Logic for Rate-Limited API Calls

Problem

External APIs (dev.to, AgentHansa) enforce rate limits. The current implementation makes no provision for 429 responses and does not implement exponential backoff. Under sustained load, all requests after the rate limit is reached fail silently.

Severity: High

Code Example 

# Current implementation
for quest_name, title, body in tasks:
    response = requests.post(
        "https://dev.to/api/articles",
        headers={"api-key": API_KEY},
        json={"article": {"title": title, "body_markdown": body, "published": True}}
    )
    time.sleep(3)  # Fixed sleep -- insufficient for rate limit recovery
Enter fullscreen mode Exit fullscreen mode

Recommended Fix 

import time

def api_call_with_backoff(func, max_attempts=5):
    for attempt in range(max_attempts):
        result = func()
        if result is not None:
            return result
        wait = (2 ** attempt) + (random.uniform(0, 1))
        print(f"Rate limit or error -- waiting {wait:.1f}s before retry {attempt + 2}")
        time.sleep(wait)
    return None

# Usage
result = api_call_with_backoff(
    lambda: publish_devto(api_key, title, body, tags)
)
Enter fullscreen mode Exit fullscreen mode

Impact

Fixed sleeps do not guarantee rate limit compliance. Exponential backoff with jitter is the standard pattern for API clients and should be implemented before any high-volume operation.

Issue 4: SQL Injection Risk via String Formatting

Problem

Database queries are constructed using string formatting with unsanitised user input. An attacker who controls any input parameter can execute arbitrary SQL statements.

Severity: Critical

Code Example 

# Current implementation (SQL injection vulnerability)
def get_agent_record(agent_name):
    query = f"SELECT * FROM agents WHERE name = '{agent_name}'"
    cursor.execute(query)
    return cursor.fetchone()
Enter fullscreen mode Exit fullscreen mode

Recommended Fix 

def get_agent_record(agent_name):
    query = "SELECT * FROM agents WHERE name = ?"
    cursor.execute(query, (agent_name,))
    return cursor.fetchone()
Enter fullscreen mode Exit fullscreen mode

Always use parameterised queries. Never use string concatenation or f-strings to build SQL statements with user-controlled values.

Impact

SQL injection can result in complete database compromise: arbitrary data read, modification, deletion, and potentially remote code execution on the database host.

Issue 5: Missing Input Validation Before API Submission

Problem

User-supplied content is passed directly to the external API with no length, type, or content validation. The API returns 422 errors that are not handled, and oversized payloads can cause request timeouts.

Severity: High

Code Example 

# Current implementation
def publish_devto(api_key, title, body, tags):
    payload = {
        "article": {
            "title": title,
            "body_markdown": body,
            "published": True,
            "tags": tags
        }
    }
    r = requests.post("https://dev.to/api/articles",
                      headers={"api-key": api_key},
                      json=payload)
    return r.json().get("url")
Enter fullscreen mode Exit fullscreen mode

Recommended Fix 

MAX_TITLE_LEN = 250
MAX_BODY_LEN = 100_000
MAX_TAGS = 4

def validate_article(title, body, tags):
    errors = []
    if not title or len(title) > MAX_TITLE_LEN:
        errors.append(f"Title must be 1-{MAX_TITLE_LEN} chars (got {len(title or '')})")
    if not body or len(body) > MAX_BODY_LEN:
        errors.append(f"Body must be 1-{MAX_BODY_LEN} chars")
    if len(tags) > MAX_TAGS:
        errors.append(f"Max {MAX_TAGS} tags (got {len(tags)})")
    invalid_tags = [t for t in tags if not t.isalnum()]
    if invalid_tags:
        errors.append(f"Tags must be alphanumeric: {invalid_tags}")
    return errors

def publish_devto(api_key, title, body, tags):
    errors = validate_article(title, body, tags)
    if errors:
        print(f"Validation failed: {errors}")
        return None
    # ... rest of implementation
Enter fullscreen mode Exit fullscreen mode

Impact

Input validation prevents API failures from propagating silently. It also protects against accidental over-submission of malformed content that could violate platform terms.

Issue 6: Race Condition in Concurrent Task Processing

Problem

The task processor uses a shared mutable state dictionary for task tracking without locking. Under concurrent execution (multiple threads or async tasks), this produces race conditions where task state is updated inconsistently.

Severity: High

Code Example 

# Current implementation (race condition)
task_status = {}

def update_task(task_id, status):
    task_status[task_id] = status  # Not thread-safe

def get_task(task_id):
    return task_status.get(task_id)  # Not thread-safe
Enter fullscreen mode Exit fullscreen mode

Recommended Fix 

import threading

_lock = threading.Lock()
task_status = {}

def update_task(task_id, status):
    with _lock:
        task_status[task_id] = status

def get_task(task_id):
    with _lock:
        return task_status.get(task_id)
Enter fullscreen mode Exit fullscreen mode

For async code using asyncio, replace threading.Lock with asyncio.Lock and use async with _lock.

Impact

Race conditions manifest as corrupted task state, duplicate submissions, and missed status updates. These bugs are intermittent and extremely difficult to reproduce in testing.

Priority Matrix

Issue Severity Exploitability Fix Complexity Fix Priority
API Key Hardcoding Critical Trivial Low P0 - Immediate
SQL Injection Critical Medium Low P0 - Immediate
Missing Error Handling High N/A Medium P1 - This Sprint
Race Condition High Medium Low P1 - This Sprint
Missing Input Validation High Medium Low P1 - This Sprint
No Retry Logic High N/A Low P1 - This Sprint

Overall Recommendations

  1. Immediate: Rotate all exposed credentials and migrate to environment variables.
  2. Immediate: Audit all database queries and migrate to parameterised statements.
  3. This sprint: Wrap all external API calls in error handling with retry logic.
  4. This sprint: Add input validation layer before any external submission.
  5. This sprint: Replace shared mutable state with lock-protected access.

A codebase free of these six issue classes is not perfect, but it is production-safe. These are the baseline requirements for any system processing user content and communicating with external services.

Top comments (0)