Building AI Agents for DevOps Part 6: Security Hardening

In Part 5, I refactored the delegation patterns. The architecture is clean, the agents compose well, and the observability is solid. But while reviewing the codebase with fresh eyes, I found four...

In Part 5, I refactored the delegation patterns. The architecture is clean, the agents compose well, and the observability is solid. But while reviewing the codebase with fresh eyes, I found four security gaps that would block any real production deployment.

This post fixes all four. No new features — just making what exists safe to run in a shared environment.

The Audit

I asked a simple question: what can go wrong if untrusted input reaches my tools? The answer was uncomfortable:

None of these require a sophisticated attacker. A malformed user query or a misconfigured session is enough. Let's fix them one by one.

Fix 1: Input Validation Library

The platform had 30+ tool functions and not one validated its inputs. A user could ask the agent to create a Kafka topic named ../../etc/passwd, request 10 million log lines, or pass a javascript: URL as a bookmark.

The Design

I needed validators that are: reusable across all agents, composable, and return structured errors (not exceptions) so the LLM can reason about them.

Each validator is a pure function: None means valid, {"status": "error", "message": ...} means invalid.

from ai_agents_core.validation import (
    validate_string, validate_positive_int, validate_url,
    validate_path, validate_list,
    KAFKA_TOPIC_PATTERN, K8S_NAME_PATTERN,
)

The Pattern: Walrus Operator

Every tool now validates inputs at the top with early-return:

def create_kafka_topic(
    topic_name: str, num_partitions: int = 1, replication_factor: int = 1
) -> dict:
    if err := validate_string(topic_name, "topic_name", pattern=KAFKA_TOPIC_PATTERN):
        return err
    if err := validate_positive_int(num_partitions, "num_partitions", max_value=10_000):
        return err
    if err := validate_positive_int(replication_factor, "replication_factor", max_value=10):
        return err
    # ... actual logic

The walrus operator (:=) makes this concise — one line per parameter. If any validation fails, the tool returns an error dict immediately. The LLM gets a clear message like "topic_name does not match required format" and can ask the user to fix it.

What Gets Validated

Pre-compiled regex patterns enforce naming rules:

K8S_NAME_PATTERN = re.compile(r"^[a-z0-9]([a-z0-9._-]*[a-z0-9])?$")
KAFKA_TOPIC_PATTERN = re.compile(r"^[a-zA-Z0-9._-]+$")

And constants cap resource-intensive parameters:

MAX_LOG_LINES = 10_000         # tail parameter cap
MAX_REPLICAS = 1_000           # K8s scaling cap
MAX_PARTITIONS = 10_000        # Kafka topic cap
MAX_QUERY_LENGTH = 5_000       # Prometheus/Loki query cap

Applied Everywhere

Every agent got validation in one pass:

Fix 2: Confirmation Bypass

This one was subtle. The require_confirmation() guardrail worked like this:

1. User: "delete topic orders"
2. LLM calls delete_kafka_topic(topic="orders")
3. Guardrail stores: pending["delete_kafka_topic"] = True
4. Returns: "Please confirm this destructive operation"
5. User: "yes"
6. LLM calls delete_kafka_topic(topic="orders") again
7. Guardrail sees pending is True → allows it

The problem: step 6 could have different arguments. The pending flag was a simple boolean keyed by tool name — it didn't know which arguments were confirmed. A user could confirm deletion of topic "test", and the LLM could replay with topic "production".

Worse: the pending flag never expired. A stale confirmation from an earlier conversation could authorize a future operation.

The Fix

Replace the boolean with a struct:

# Before
tool_context.state[pending_key] = True

# After
tool_context.state[pending_key] = {
    "args_hash": _hash_args(args),  # SHA-256 of JSON-serialized args
    "timestamp": time.time(),
}

On re-entry, three things must match:

pending = tool_context.state.get(pending_key)

if isinstance(pending, dict):
    same_args = pending.get("args_hash") == _hash_args(args)
    not_expired = (time.time() - pending.get("timestamp", 0)) < 300  # 5 min TTL

    if same_args and not_expired:
        tool_context.state[pending_key] = None  # consume — one-time use
        return None  # allow

    tool_context.state[pending_key] = None  # clear stale state

# Not confirmed — store new pending and prompt
tool_context.state[pending_key] = {"args_hash": _hash_args(args), "timestamp": time.time()}
return {"status": "confirmation_required", ...}

Three properties that the old boolean didn't have:

1. Args binding — confirming topic "test" doesn't authorize deleting topic "production"

2. TTL — confirmations expire after 5 minutes

3. One-time use — confirmation is consumed after use, preventing replay

Fix 3: Docker Secret Redaction

The inspect_container() tool returned the full output of docker inspect, including environment variables. When the LLM summarizes a container's state, it would happily include:

Container "postgres" is running.
Environment: DB_PASSWORD=s3cret_pr0d!, API_TOKEN=sk-live-abc123...

The LLM doesn't know what's sensitive. It will print whatever the tool returns.

The Fix

A simple redaction function applied before the tool returns:

_ENV_SENSITIVE_PATTERNS = frozenset(
    {"password", "secret", "token", "api_key", "credential", "key", "auth"}
)

def _redact_env_vars(env_list: list[str]) -> list[str]:
    redacted = []
    for entry in env_list:
        if "=" in entry:
            var_name, _, _value = entry.partition("=")
            if any(s in var_name.lower() for s in _ENV_SENSITIVE_PATTERNS):
                redacted.append(f"{var_name}=***")
            else:
                redacted.append(entry)
        else:
            redacted.append(entry)
    return redacted

Applied in inspect_container():

return {
    "status": "success",
    "name": info.get("Name", "").lstrip("/"),
    "image": config.get("Image"),
    "env_vars": _redact_env_vars(config.get("Env", [])),  # ← redacted
    ...
}

Now the LLM sees DB_PASSWORD=*** and can still report that the variable exists without leaking its value. This complements the audit logger, which already redacted arguments containing sensitive keywords — but the tool output itself was unprotected until now.

Fix 4: Authentication Enforcement

The RBAC from Part 3 checks session.state["user_role"] on every tool call. But who sets that role? In the Slack bot, it was:

session = await session_service.create_session(
    state={"user_role": role},  # set from Slack user ID mapping
)

But what if someone creates a session through the web UI or API and injects "user_role": "admin" into the initial state? The RBAC system would trust it — it has no way to distinguish server-set roles from client-injected ones.

The Fix

Two new functions that work together:

# Server-side: marks the role as trusted
def set_user_role(state: dict, role: str) -> None:
    normalised = role.lower()
    if normalised not in {"viewer", "operator", "admin"}:
        normalised = "viewer"
    state["user_role"] = normalised
    state["_role_set_by_server"] = True  # trust flag

# Agent callback: enforces trust
def ensure_default_role(default="viewer"):
    def callback(callback_context):
        state = callback_context.state
        if not state.get("_role_set_by_server"):
            state["user_role"] = default  # override untrusted role
    return callback

The Slack bot and CLI runner now use set_user_role():

# Slack bot
initial_state = {}
set_user_role(initial_state, role)  # trusted
session = await session_service.create_session(state=initial_state, ...)

# CLI runner
initial_state = {}
set_user_role(initial_state, "admin")  # CLI user gets admin

And every agent's before_agent_callback includes:

root_agent = create_agent(
    ...,
    before_agent_callback=ensure_default_role(),  # forces viewer if untrusted
    before_tool_callback=[authorize(), require_confirmation()],
)

Now if a session arrives without the trust flag, the role is forced to viewer regardless of what user_role says. Privilege escalation is blocked at the callback level.

Testing: 106 New Tests

Each fix came with dedicated tests. The total went from 289 to 395.

All tests use mocks — no running Kafka, Docker, or Kubernetes needed. The validation tests are pure function tests, the fastest kind:

def test_traversal_rejected(self):
    result = validate_path("../../etc/passwd", "x")
    assert result["status"] == "error"
    assert "traversal" in result["message"]

def test_javascript_url_rejected(self):
    result = validate_url("javascript:alert(1)", "x")
    assert result["status"] == "error"

def test_bool_rejected_as_int(self):
    result = validate_positive_int(True, "x")
    assert result["status"] == "error"  # True is technically int in Python

That last one is worth noting: Python's bool subclasses int, so isinstance(True, int) returns True. The validator explicitly checks isinstance(value, bool) first to reject it.

What Changed (Summary)

Files changed: 19 · Lines added: 1,011

No new features. No new dependencies. Just making the existing surface area safe.

What's Still Missing

These fixes close the critical gaps, but production readiness has more layers:

Each of these is a future post. The foundation is now solid enough to build on.

Source code on GitHub. Validation module: core/orrery_core/security/validation.py.