Docs

Appearance

Security Agent Development Guide ​

Overview ​

This guide covers how to build, test, and deploy a new security monitoring agent. Agents are Python modules that run inside the bedrock-hub Container Apps Job every 4–6 hours. Each agent checks one domain (identity, network, a kill chain stage, etc.), posts findings to the SecOps platform, and returns a structured result dict.

All agent code lives in bedrock-hub/agents/. The orchestrator picks them up automatically once registered.

Architecture Recap ​

orchestrator.py
  └── runs mini-agents in parallel (ThreadPoolExecutor)
        │  POST /api/incidents per finding
        ▼
  analyst_agent.py     — rules engine: NEW → CLOSED (known-good) or OPEN
  engineer_agent.py    — recurring noise suppression, no LLM
  history_matcher_agent.py — pattern matching against resolved history
  detection_agent.py   — LLM enrichment of remaining OPEN incidents
  triage_agent.py / security_agent.py — pages Rory if warranted

Your new agent is a mini-agent — it runs in the parallel pool, posts findings, and returns. It does not run the analyst pipeline; that happens after all mini-agents complete.

Agent File Structure ​

Create one file: bedrock-hub/agents/<domain>_agent.py

Naming convention: <domain>_<environment>_agent.py for environment-specific agents, <domain>_agent.py for PROD-only or cross-environment.

Examples: lateral_movement_agent.py, fim_dde_agent.py, aws_persistence_agent.py

Required Output Contract ​

Every agent's run() function must return this exact shape:

python
{
    "findings": [               # list of finding dicts — empty list if nothing found
        {
            "severity":           "critical|high|medium|low",
            "resource":           "short-identifier",   # VM name, account, IP, etc.
            "description":        "what happened",      # 1–3 sentences, specific
            "evidence":           "raw supporting data", # log line, KQL result, API response
            "hipaa_risk":         "HIPAA context",      # which safeguard is implicated
            "recommended_action": "what to do",         # specific steps
            # Optional but encouraged:
            "mitre_technique":    "T1003",
            "mitre_tactic":       "Credential Access",
            "source_system":      "PROD",               # PROD | DDE | AWS
        }
    ],
    "iocs":                     ["list-of-ioc-strings"],  # same as finding resources
    "confidence":               90,                       # 0–100 int
    "summary":                  "One paragraph describing what was checked and what was found.",
    "immediate_action_required": False,                   # True only for active breaches
}

Never raise an exception from run() — catch all exceptions, log them, and return a degraded result with confidence: 0 and an explanatory summary. An uncaught exception kills the agent thread and produces no findings.

Boilerplate Template ​

python
"""
<domain>_agent.py — <one line description>

Checks: <what this agent monitors>
Data source: <LAW query | API | etc.>
Fires on: <what triggers a finding>
MITRE: <ATT&CK technique>
"""

import logging
import os

import secops_client

log = logging.getLogger(__name__)

# Declare what this agent checks — used for audit trail and evidence registration.
AGENT_QUERIES = [
    {
        "description": "One-line description of what this query detects",
        "mitre_technique": "T1234",
        "mitre_tactic": "Tactic Name",
        "source_system": "PROD",   # PROD | DDE | AWS
    },
]


def run() -> dict:
    secops_client.register_queries(AGENT_QUERIES, "<domain>")
    log.info("Starting <domain>_agent...")

    findings: list[dict] = []

    try:
        # ── Your detection logic here ─────────────────────────────────────
        results = _check_something()

        for row in results:
            if _is_suspicious(row):
                findings.append({
                    "severity":           "high",
                    "resource":           row["resource_name"],
                    "description":        f"Suspicious activity detected on {row['resource_name']}: ...",
                    "evidence":           str(row),
                    "hipaa_risk":         "HIPAA §164.312 — ...",
                    "recommended_action": "Investigate and contain.",
                    "mitre_technique":    "T1234",
                    "mitre_tactic":       "Tactic Name",
                    "source_system":      "PROD",
                })

    except Exception as exc:
        log.exception("Unhandled error in <domain>_agent: %s", exc)
        return {
            "findings":                [],
            "iocs":                    [],
            "confidence":              0,
            "summary":                 f"<domain>_agent failed with error: {exc}. Check logs.",
            "immediate_action_required": False,
        }

    iocs = [f["resource"] for f in findings]

    if not findings:
        summary = "No suspicious <domain> activity detected. <N> events checked."
    else:
        summary = f"{len(findings)} finding(s): {', '.join(iocs)}. See findings for details."

    return {
        "findings":                findings,
        "iocs":                    iocs,
        "confidence":              90,
        "summary":                 summary,
        "immediate_action_required": any(f["severity"] == "critical" for f in findings),
    }


def _check_something() -> list[dict]:
    """Your detection logic — LAW query, API call, etc."""
    raise NotImplementedError

Querying Log Analytics (KQL) ​

Most PROD and DDE agents query the central LAW workspace. Use kql_utils.py:

python
from kql_utils import run_kql_query

# Workspace IDs are loaded from env — don't hardcode.
# LAW_WORKSPACE_ID      → cirius-logging-law-central (PROD, Sentinel-enabled)
# DDE_LAW_WORKSPACE_ID  → ciriusdde-logging-law-central (DDE)

results = run_kql_query("""
SecurityEvent
| where TimeGenerated > ago(4h)
| where EventID == 4688
| where NewProcessName has_any ("mimikatz", "procdump", "lsass")
| project TimeGenerated, Computer, Account, NewProcessName, CommandLine
""")
# results is a list of dicts — one per row

Critical LAW gotcha: SecurityEvent (EventID 4688, 4624, etc.) only exists in the Sentinel-enabled workspace (LAW_WORKSPACE_ID). If you query the ops workspace for SecurityEvent you get empty results silently. Always query LAW_WORKSPACE_ID for Windows security events. See Key Learnings for details.

Hostname normalization: LAW mixes FQDNs and short names. Normalize with:

kql
| extend Host = toupper(tostring(split(Computer, ".")[0]))

Querying AWS (CloudTrail / Security Hub) ​

For AWS agents, use boto3 with the credentials provided via the Container App's managed identity. Example:

python
import boto3

def _get_cloudtrail_events(account_id: str) -> list[dict]:
    client = boto3.client("cloudtrail", region_name="us-west-2")
    resp = client.lookup_events(
        LookupAttributes=[{"AttributeKey": "EventName", "AttributeValue": "ConsoleLogin"}],
        MaxResults=50,
    )
    return resp.get("Events", [])

AWS agents typically iterate all 7 account IDs. See aws_cloudtrail_agent.py for the cross-account assume-role pattern.

Severity Guide ​

SeverityWhen to use
criticalActive breach indicators — log cleared (1102), LSASS access, break-glass login, ransomware behavioral pattern. Pages Rory immediately.
highStrong indicators requiring same-day response — new persistence outside CM window, spray pattern, admin share access from unexpected source.
mediumAnomalies requiring investigation but not urgent — compliance drift, cert expiring >30 days, unusual but possibly-legitimate behavior.
lowInformational — policy drift, minor hygiene issues. Review in weekly security review.

When in doubt, go high. A false-positive CRITICAL is annoying but safe. A missed CRITICAL is a ransomware incident.

Registering Your Agent in the Orchestrator ​

Once the agent file exists, add it to two places:

1. agents/run_agent.py — AGENTS list ​

python
AGENTS = [
    ...
    ("lateral_movement", "lateral_movement_agent"),   # ← add here
    ...
]

The string key (first element) is used for python run_agent.py lateral_movement.

2. agents/orchestrator.py (or the appropriate mini_orchestrator) ​

The main orchestrator (orchestrator.py) handles PROD agents. mini_orchestrator_dde.py handles DDE. mini_orchestrator_aws.py handles AWS.

In the right orchestrator, import your module and add it to the parallel pool:

python
import lateral_movement_agent   # ← add import at top

# In run_mini_agents():
futures = {
    executor.submit(lateral_movement_agent.run): "lateral_movement",
    ...
}

Testing Locally ​

bash
cd bedrock-hub/agents

# Set required env vars (pull from Key Vault or use test values)
export LAW_WORKSPACE_ID=5d76d1f2-...
export SECOPS_API_URL=https://secops.bedrockcybersecurity.org
export INGEST_API_KEY=...

# Run your agent standalone
python run_agent.py lateral_movement

# Or run it directly
python lateral_movement_agent.py

The if __name__ == "__main__": block at the bottom of each agent prints the result as JSON. Add one to your agent:

python
if __name__ == "__main__":
    import json, sys
    result = run()
    print(json.dumps(result, indent=2, default=str))
    if result.get("immediate_action_required"):
        sys.exit(1)

Check the output carefully:

  • findings is a list (never None)
  • confidence is 0–100 (not a string)
  • summary describes what was actually checked, not just a generic message
  • No unhandled exceptions

Deployment ​

Agents deploy via the bedrock-hub GitHub Actions pipeline on merge to main. Never deploy locally.

  1. Create a branch, write the agent, register it, open a PR
  2. CI runs linting and basic import checks
  3. Rory reviews and merges
  4. GitHub Actions builds a new Docker image, pushes to ciriusagentsprod.azurecr.io, and updates the Container Apps Job to use the new image revision
  5. The next scheduled run (within 6 hours) picks up the new agent automatically

To force a run without waiting for the schedule, trigger the Container Apps Job manually from the Azure Portal: rg-logging-logs → job-orchestrator-cirius → Run.

Kill Chain Agent Specs (Reference) ​

The six kill chain agents scheduled for Phase 2 build. Each fires CRITICAL with no analyst gate delay — direct page.

Agent fileKill chain stageKey eventsFires on
execution_agent.pyExecutionEventID 4688, 4104LOLBins, mimikatz, mshta, wscript spawning from unexpected parents; processes in temp/AppData
persistence_agent.pyPersistenceEventID 7045, 4698, FIMNew service or scheduled task outside a CM window
credential_dumping_agent.pyCredential DumpEventID 4688 (LSASS), multi-target authLSASS process access; spray pattern (5+ failed auth then success across 3+ machines)
lateral_movement_agent.pyLateral MovementEventID 4624, 4648, 5140Same account on 3+ machines in 30 min; admin shares (C$, ADMIN$); explicit creds workstation→server
exfiltration_agent.pyExfiltrationPalo Alto CEF, App-ID, DNS SecurityOutbound bytes above baseline; new destination not seen in 30 days; App-ID: rclone/WinSCP/mega
defense_evasion_agent.pyDefense EvasionEventID 1102, Cortex silenceLog cleared = CRITICAL immediately; Cortex returning 0 findings while other agents elevated = HIGH

All six should set immediate_action_required: True for CRITICAL findings.

Common Mistakes ​

  • Returning None instead of an empty findings list — breaks the orchestrator
  • Hardcoding workspace IDs — always use environment variables from Key Vault
  • Querying the ops workspace for SecurityEvent — returns empty silently, see Key Learnings
  • Raising exceptions from run() — kills the thread, produces no findings, no alert
  • Setting immediate_action_required: True for medium findings — pages Rory unnecessarily
  • Not stamping source_system — incidents from different environments collide on deduplication

Document History ​

DateChangeAuthor
May 2026Initial draft — based on code review of bedrock-hub/agents/, cert_expiry_agent.py as reference implementation, and CLAUDE.md kill chain specsRory

Internal use only — Cirius Group