ADu2021/cyber-zero-runtime-free-agent-training

Cyber-Zero: Training Cybersecurity Agents Without Runtime

Cyber-Zero addresses a critical challenge in cybersecurity agent development: the lack of accessible runtime environments for training. Rather than requiring live CTF platforms or complex setups, the system leverages public documentation and persona-driven simulation to create high-quality synthetic training data, enabling agents to match proprietary system performance.

Core Concept

The fundamental insight is that cybersecurity knowledge is extensively documented in CTF (Capture-The-Flag) writeups and solutions. Cyber-Zero:

• Extracts knowledge from public CTF writeups and documentation
• Simulates agent interactions using persona-driven LLM simulation
• Synthesizes trajectories of realistic long-horizon problem-solving
• Trains agents on synthetic data without runtime access
• Achieves parity with systems that have access to live environments

Architecture Overview

The framework consists of:

• CTF Knowledge Extractor: Parses writeups to identify problem types and solutions
• Persona-Driven Simulator: Uses LLMs with specific personas to simulate agent interactions
• Trajectory Synthesizer: Generates plausible multi-step trajectories
• Data Quality Validator: Ensures synthetic trajectories are realistic
• Agent Training Pipeline: Fine-tunes agents on synthesized data

Implementation Steps

Step 1: Extract knowledge from CTF writeups

Parse CTF solutions to identify problem types and solution patterns:

import re
import json
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass

@dataclass
class CTFProblem:
    """Represents a CTF challenge and its solution"""
    name: str
    category: str  # 'crypto', 'web', 'reverse', 'pwn', 'forensics'
    description: str
    solution_steps: List[str]
    tools_used: List[str]
    vulnerability_type: str
    difficulty: str

class CTFKnowledgeExtractor:
    """Extracts knowledge from CTF writeups"""

    def __init__(self):
        self.problems: List[CTFProblem] = []
        self.vulnerability_index: Dict[str, List[CTFProblem]] = {}

    def parse_writeup(self, writeup_text: str) -> Optional[CTFProblem]:
        """
        Parse CTF writeup to extract problem and solution.

        Args:
            writeup_text: Complete writeup or solution document

        Returns:
            CTFProblem with extracted information
        """
        # Extract title
        title_match = re.search(r'^#+\s*(.+?)$', writeup_text, re.MULTILINE)
        title = title_match.group(1) if title_match else "Unknown"

        # Detect category by content analysis
        category = self._detect_category(writeup_text)

        # Extract problem description
        description = self._extract_description(writeup_text)

        # Extract solution steps
        solution_steps = self._extract_solution_steps(writeup_text)

        # Identify tools used
        tools = self._identify_tools(writeup_text, solution_steps)

        # Determine vulnerability type
        vulnerability = self._extract_vulnerability(writeup_text)

        # Assess difficulty
        difficulty = self._assess_difficulty(solution_steps)

        problem = CTFProblem(
            name=title,
            category=category,
            description=description,
            solution_steps=solution_steps,
            tools_used=tools,
            vulnerability_type=vulnerability,
            difficulty=difficulty
        )

        self.problems.append(problem)

        # Index by vulnerability
        if vulnerability not in self.vulnerability_index:
            self.vulnerability_index[vulnerability] = []

        self.vulnerability_index[vulnerability].append(problem)

        return problem

    def _detect_category(self, text: str) -> str:
        """Detect challenge category from text"""

        categories = {
            'crypto': ['encryption', 'rsa', 'aes', 'hash', 'cipher'],
            'web': ['http', 'sql', 'xss', 'csrf', 'web server'],
            'reverse': ['binary', 'assembly', 'decompile', 'ida'],
            'pwn': ['buffer', 'overflow', 'rop', 'shellcode', 'exploit'],
            'forensics': ['file', 'memory', 'log', 'metadata']
        }

        text_lower = text.lower()

        for category, keywords in categories.items():
            if any(kw in text_lower for kw in keywords):
                return category

        return 'misc'

    def _extract_description(self, text: str) -> str:
        """Extract problem description"""

        lines = text.split('\n')
        description_lines = []

        in_description = False

        for line in lines:
            if 'description' in line.lower() or 'challenge' in line.lower():
                in_description = True
            elif in_description and line.strip() and not line.startswith('#'):
                description_lines.append(line)
            elif in_description and line.strip().startswith('#'):
                break

        return '\n'.join(description_lines[:5])

    def _extract_solution_steps(self, text: str) -> List[str]:
        """Extract step-by-step solution"""

        steps = []

        # Look for numbered steps or code blocks
        step_matches = re.findall(
            r'(\d+)[.)]\s*(.+?)(?=\n\d+[.)]|\n\n|\Z)',
            text,
            re.DOTALL
        )

        for num, step_text in step_matches:
            # Clean step text
            step = step_text.strip()[:200]
            if step:
                steps.append(step)

        # If no numbered steps, extract code blocks as steps
        if not steps:
            code_blocks = re.findall(r'```.*?\n(.+?)\n```', text, re.DOTALL)
            steps = [b[:200] for b in code_blocks[:5]]

        return steps

    def _identify_tools(self, text: str, steps: List[str]) -> List[str]:
        """Identify tools used in solution"""

        tools = set()

        # Common CTF tools
        tool_patterns = {
            'python': r'python\s+\w+',
            'ida': r'IDA|ida pro',
            'ghidra': r'Ghidra',
            'gdb': r'gdb|GDB',
            'strings': r'strings',
            'hexdump': r'hexdump',
            'wireshark': r'Wireshark',
            'sqlmap': r'sqlmap',
            'burp': r'Burp Suite',
            'pwntools': r'pwntools|pwn'
        }

        full_text = text + ' ' + ' '.join(steps)

        for tool, pattern in tool_patterns.items():
            if re.search(pattern, full_text, re.IGNORECASE):
                tools.add(tool)

        return list(tools)

    def _extract_vulnerability(self, text: str) -> str:
        """Extract vulnerability type"""

        vulnerabilities = [
            'buffer_overflow', 'sql_injection', 'xss', 'rsa_weakness',
            'weak_crypto', 'file_upload', 'privilege_escalation'
        ]

        text_lower = text.lower()

        for vuln in vulnerabilities:
            if vuln.replace('_', ' ') in text_lower or vuln in text_lower:
                return vuln

        return 'unknown'

    def _assess_difficulty(self, steps: List[str]) -> str:
        """Assess difficulty from solution complexity"""

        if len(steps) <= 2:
            return 'easy'
        elif len(steps) <= 5:
            return 'medium'
        else:
            return 'hard'

    def get_problems_by_category(self, category: str) -> List[CTFProblem]:
        """Get all problems in category"""
        return [p for p in self.problems if p.category == category]

    def get_problems_by_vulnerability(self,
                                     vuln_type: str) -> List[CTFProblem]:
        """Get all problems with specific vulnerability"""
        return self.vulnerability_index.get(vuln_type, [])

Step 2: Implement persona-driven LLM simulator

Create LLM-based agent simulation with different personas:

from enum import Enum

class AgentPersona(Enum):
    """Different agent personas for simulation"""
    CAUTIOUS = "thorough, methodical, checks assumptions"
    AGGRESSIVE = "tries shortcuts, tests hypotheses quickly"
    EXPERT = "applies known patterns, recognizes problem types"

class PersonaDrivenSimulator:
    """Simulates agent interaction using LLM with personas"""

    def __init__(self, llm):
        self.llm = llm

    def simulate_agent_step(self, problem: CTFProblem,
                          current_state: Dict,
                          persona: AgentPersona = AgentPersona.EXPERT) -> Tuple[str, Dict]:
        """
        Simulate single agent action step.

        Args:
            problem: CTF problem to solve
            current_state: Current progress/findings
            persona: Agent personality style

        Returns:
            (action, updated_state)
        """

        # Build context
        context = self._build_context(problem, current_state, persona)

        # Generate action
        prompt = f"""{context}

Based on the current situation, what is your next action? Be specific and actionable.

Action:"""

        action = self.llm.generate(prompt, max_tokens=200).strip()

        # Simulate effect of action
        new_state = self._simulate_action_effect(action, current_state, problem)

        return action, new_state

    def _build_context(self, problem: CTFProblem,
                      current_state: Dict,
                      persona: AgentPersona) -> str:
        """Build simulation context for LLM"""

        persona_desc = {
            AgentPersona.CAUTIOUS: "You are a cautious, methodical analyst",
            AgentPersona.AGGRESSIVE: "You are an aggressive problem solver",
            AgentPersona.EXPERT: "You are an expert in cybersecurity"
        }[persona]

        context = f"""{persona_desc}.

Challenge: {problem.name}
Category: {problem.category}
Description: {problem.description}

Vulnerability Type: {problem.vulnerability_type}
Tools Available: {', '.join(problem.tools_used)}

Current Progress:
{json.dumps(current_state, indent=2)}

Solution Steps (for guidance):
{chr(10).join(f'- {s}' for s in problem.solution_steps[:3])}"""

        return context

    def _simulate_action_effect(self, action: str,
                               current_state: Dict,
                               problem: CTFProblem) -> Dict:
        """Simulate the effect of an action"""

        new_state = current_state.copy()

        # Simple heuristics for effect simulation
        action_lower = action.lower()

        if 'scan' in action_lower or 'nmap' in action_lower:
            new_state['scanned'] = True
            new_state['ports_discovered'] = [22, 80, 443]
        elif 'connect' in action_lower or 'telnet' in action_lower:
            new_state['connected'] = True
        elif 'find' in action_lower or 'grep' in action_lower:
            new_state['findings'] = (new_state.get('findings', 0) + 1)
        elif 'exploit' in action_lower:
            new_state['exploit_attempted'] = True
            new_state['success'] = True  # Simplified

        return new_state

Step 3: Generate synthetic trajectories

Create realistic multi-step problem-solving sequences:

@dataclass
class Trajectory:
    """Multi-step solution trajectory"""
    problem: CTFProblem
    steps: List[Tuple[str, Dict]]  # (action, resulting_state)
    success: bool
    num_steps: int

class TrajectoryGenerator:
    """Generates synthetic problem-solving trajectories"""

    def __init__(self, extractor: CTFKnowledgeExtractor,
                 simulator: PersonaDrivenSimulator):
        self.extractor = extractor
        self.simulator = simulator

    def generate_trajectory(self, problem: CTFProblem,
                          persona: AgentPersona = AgentPersona.EXPERT,
                          max_steps: int = 20) -> Trajectory:
        """
        Generate complete trajectory for solving a problem.

        Args:
            problem: CTF problem to solve
            persona: Agent persona to use
            max_steps: Maximum trajectory length

        Returns:
            Complete trajectory from start to goal
        """

        steps = []
        current_state = {
            'started': True,
            'found_flags': [],
            'tools_used': []
        }

        success = False

        for step_num in range(max_steps):
            # Simulate next step
            action, new_state = self.simulator.simulate_agent_step(
                problem,
                current_state,
                persona
            )

            steps.append((action, new_state))
            current_state = new_state

            # Check if goal reached (simplified)
            if 'flag' in action.lower() or new_state.get('success'):
                success = True
                break

            # Bail if stalled
            if step_num > 10 and not new_state.get('progress', False):
                break

        return Trajectory(
            problem=problem,
            steps=steps,
            success=success,
            num_steps=len(steps)
        )

    def generate_batch_trajectories(self, problems: List[CTFProblem],
                                    batch_size: int = 10) -> List[Trajectory]:
        """Generate trajectories for multiple problems"""

        trajectories = []

        for problem in problems[:batch_size]:
            # Generate with each persona
            for persona in AgentPersona:
                traj = self.generate_trajectory(problem, persona=persona)
                trajectories.append(traj)

        return trajectories

Step 4: Validate trajectory quality

Ensure synthetic data is realistic and useful:

class TrajectoryValidator:
    """Validates quality of synthetic trajectories"""

    def __init__(self):
        self.realistic_patterns = self._load_patterns()

    def validate_trajectory(self, trajectory: Trajectory) -> Dict:
        """
        Validate trajectory for realism and utility.

        Returns validation score and issues.
        """

        issues = []
        scores = {
            'realism': 0.0,
            'diversity': 0.0,
            'utility': 0.0,
            'overall': 0.0
        }

        # Check realism
        realism_score = self._check_realism(trajectory)
        if realism_score < 0.5:
            issues.append("Low action realism")
        scores['realism'] = realism_score

        # Check diversity
        actions = [step[0] for step in trajectory.steps]
        diversity = len(set(actions)) / len(actions) if actions else 0
        if diversity < 0.3:
            issues.append("Low action diversity")
        scores['diversity'] = diversity

        # Check utility
        if trajectory.success:
            utility = 0.9
        else:
            utility = 0.3
        scores['utility'] = utility

        # Overall
        scores['overall'] = (
            0.4 * realism_score + 0.3 * diversity + 0.3 * utility
        )

        return {
            'valid': scores['overall'] > 0.5,
            'scores': scores,
            'issues': issues
        }

    def _check_realism(self, trajectory: Trajectory) -> float:
        """Check if actions follow realistic patterns"""

        realism_score = 0.0
        valid_actions = 0

        for action, _ in trajectory.steps:
            if self._is_realistic_action(action):
                valid_actions += 1

        realism_score = valid_actions / len(trajectory.steps) if trajectory.steps else 0

        return realism_score

    def _is_realistic_action(self, action: str) -> bool:
        """Check if action is realistically described"""

        realistic_patterns = [
            r'(scan|nmap)',
            r'(analyze|examine)',
            r'(connect|telnet)',
            r'(crack|bruteforce)',
            r'(exploit|execute)',
            r'(find|search)',
            r'(decrypt|decode)'
        ]

        action_lower = action.lower()

        for pattern in realistic_patterns:
            if re.search(pattern, action_lower):
                return True

        return False

    def _load_patterns(self) -> List[str]:
        """Load realistic action patterns"""
        return []

    def filter_valid_trajectories(self,
                                 trajectories: List[Trajectory],
                                 min_score: float = 0.5) -> List[Trajectory]:
        """Filter trajectories by quality"""

        valid = []

        for trajectory in trajectories:
            validation = self.validate_trajectory(trajectory)

            if validation['scores']['overall'] > min_score:
                valid.append(trajectory)

        return valid

Step 5: Integrate agent training

Train agents on synthetic trajectories:

class CyberZeroTrainer:
    """Trains cybersecurity agents on synthetic trajectories"""

    def __init__(self, agent_model, trajectory_dataset: List[Trajectory]):
        self.agent = agent_model
        self.trajectories = trajectory_dataset

    def prepare_training_data(self) -> Tuple[List, List]:
        """
        Convert trajectories to training data.

        Returns (problem_contexts, action_sequences)
        """

        contexts = []
        sequences = []

        for trajectory in self.trajectories:
            problem = trajectory.problem
            actions = [step[0] for step in trajectory.steps]

            # Context: problem description and category
            context = f"{problem.category}: {problem.description}"
            contexts.append(context)

            # Action sequence
            sequences.append(actions)

        return contexts, sequences

    def train(self, learning_rate: float = 1e-5, num_epochs: int = 3):
        """Train agent on trajectories"""

        contexts, action_sequences = self.prepare_training_data()

        optimizer = torch.optim.AdamW(
            self.agent.parameters(),
            lr=learning_rate
        )

        for epoch in range(num_epochs):
            total_loss = 0.0

            for context, actions in zip(contexts, action_sequences):
                # Encode problem
                problem_encoding = self.agent.encode_problem(context)

                # Predict next action at each step
                for step_idx, target_action in enumerate(actions):
                    # Get prediction
                    pred_action = self.agent.predict_next_action(
                        problem_encoding,
                        actions[:step_idx]
                    )

                    # Compute loss
                    loss = self._compute_loss(pred_action, target_action)

                    optimizer.zero_grad()
                    loss.backward()
                    optimizer.step()

                    total_loss += loss.item()

            avg_loss = total_loss / len(contexts)
            print(f"Epoch {epoch}: Loss = {avg_loss:.4f}")

    def _compute_loss(self, prediction, target) -> torch.Tensor:
        """Compute training loss"""
        # Simplified: real version would use cross-entropy
        return torch.tensor(0.0)

Practical Guidance

When to use Cyber-Zero:

• Training agents without access to live CTF platforms
• Building cybersecurity training systems
• Creating agents for security research
• Democratizing access to security agent capabilities
• Scenarios where environment setup is prohibitive

When NOT to use Cyber-Zero:

• Systems requiring real-world environment interaction
• Tasks needing actual vulnerability exploitation
• Real-time security operations requiring live data
• Scenarios where synthetic data accuracy is critical

Key hyperparameters:

• max_trajectory_steps: 15-25 typical for realistic solutions
• num_personas: 3 provides good diversity
• validation_min_score: 0.5-0.6 typical threshold
• training_epochs: 2-5 typical for convergence

Expected performance:

• Achieves parity with Claude-3.5-Sonnet on CTF tasks
• 13.1% absolute improvement on standard benchmarks
• Training data synthesis: 1000+ trajectories from writeups
• Agent training time: Hours on single GPU

Dataset characteristics:

• Typical: 100-500 CTF writeups per category
• Generates 3-5 trajectories per writeup per persona
• Quality filtering keeps 60-70% of generated trajectories
• Results in 10K+ training examples per domain

Reference

Cyber-Zero: Training Cybersecurity Agents without Runtime. arXiv:2508.00910