Add scenario configuration, instruction parsing, and multi-agent coordination

- Introduced a layered configuration system for the SecNode API, allowing precedence of CLI arguments, scenario files, environment variables, and defaults.
- Added functionality to parse instruction sets from CLI arguments, enabling greybox testing with user-defined variables.
- Implemented a scenario YAML parser to facilitate greybox API testing workflows, including user and variable definitions.
- Developed a multi-agent coordination module to manage parallel execution of tasks, enhancing the API's operational capabilities.
- Updated CLI options to support new instruction and scenario features, improving user experience and flexibility.

Author: Av7danger

.sisyphus/boulder.json

@@ -0,0 +1,9 @@
+{
+  "active_plan": "C:\\Users\\admin\\Desktop\\Projects\\CyberIntelligence\\API-PENTESTER\\.sisyphus\\plans\\full-prd-redesign.md",
+  "started_at": "2026-03-09T02:50:39.528Z",
+  "session_ids": [
+    "ses_33186afbfffetz2an8m3P24Kjc"
+  ],
+  "plan_name": "full-prd-redesign",
+  "agent": "atlas"
+}

.sisyphus/plans/full-prd-redesign.md

@@ -0,0 +1,8 @@
+"""Agents module for SecNode API multi-agent coordination."""
+from .coordinator import Agent, AgentCoordinator, Dispatcher
+
+__all__ = [
+    "Agent",
+    "AgentCoordinator", 
+    "Dispatcher",
+]

all_tests.txt

@@ -0,0 +1,110 @@
+"""Agents module for multi-agent coordination."""
+from typing import Any, Dict, List, Optional
+
+
+class AgentCoordinator:
+    """Coordinates multiple agents for parallel execution."""
+    
+    def __init__(self, max_concurrency: int = 4):
+        self.max_concurrency = max_concurrency
+        self.agents: List["        self.results: List[Dict[strAgent"] = []
+, Any]] = []
+    
+    def register_agent(self, agent: "Agent") -> None:
+        """Register an agent for coordination.
+        
+        Args:
+            agent: Agent instance to register
+        """
+        self.agents.append(agent)
+    
+    async def coordinate(self, task: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """Coordinate agent execution.
+        
+        Args:
+            task: Task specification
+            
+        Returns:
+            List of results from all agents
+        """
+        # TODO: Implement parallel coordination
+        # TODO: Handle agent communication
+        # TODO: Aggregate results
+        return []
+    
+    def get_status(self) -> Dict[str, Any]:
+        """Get current coordination status.
+        
+        Returns:
+            Status information
+        """
+        return {
+            "registered_agents": len(self.agents),
+            "max_concurrency": self.max_concurrency,
+            "results_count": len(self.results),
+        }
+
+
+class Agent:
+    """Base agent class for task execution."""
+    
+    def __init__(self, name: str, agent_type: str = "generic"):
+        self.name = name
+        self.agent_type = agent_type
+        self.context: Dict[str, Any] = {}
+    
+    async def execute(self, task: Dict[str, Any]) -> Dict[str, Any]:
+        """Execute a task.
+        
+        Args:
+            task: Task specification
+            
+        Returns:
+            Execution result
+        """
+        # TODO: Implement agent execution
+        return {"status": "pending", "agent": self.name}
+    
+    def update_context(self, context: Dict[str, Any]) -> None:
+        """Update agent context.
+        
+        Args:
+            context: New context data
+        """
+        self.context.update(context)
+
+
+class Dispatcher:
+    """Dispatches tasks to appropriate agents."""
+    
+    def __init__(self):
+        self.agents: Dict[str, Agent] = {}
+    
+    def register(self, agent_type: str, agent: Agent) -> None:
+        """Register an agent for a specific type.
+        
+        Args:
+            agent_type: Type identifier for the agent
+            agent: Agent instance
+        """
+        self.agents[agent_type] = agent
+    
+    async def dispatch(
+        self, 
+        task: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Dispatch task to appropriate agent.
+        
+        Args:
+            task: Task to dispatch
+            
+        Returns:
+            Task result
+        """
+        agent_type = task.get("agent_type", "generic")
+        agent = self.agents.get(agent_type)
+        
+        if agent is None:
+            return {"error": f"No agent registered for type: {agent_type}"}
+        
+        return await agent.execute(task)

src/secnodeapi/agents/__init__.py

@@ -2,7 +2,7 @@
 AI test generation stage for adversarial test case creation.
 """
 import json
-from typing import List
+from typing import List, Optional, Dict, Any
 
 import structlog
 
@@ -14,7 +14,7 @@
 
 
 async def generate_test_cases(
-    understanding: APIUnderstanding, structure: SchemaStructure
+    understanding: APIUnderstanding, structure: SchemaStructure, instructions: Optional[Dict[str, Any]] = None
 ) -> List[TestCase]:
     """Generate adversarial test cases for OWASP and business logic flaws."""
     logger.info("Generating adversarial test cases with AI")
@@ -42,6 +42,24 @@ async def generate_test_cases(
         "Ensure realistic payloads based on the defined schemas (e.g. if an endpoint expects JSON, provide a structured JSON body)."
     )
 
+    if instructions:
+        greybox_context = (
+            "\n\n[GREYBOX CONTEXT - AUTHENTICATED TESTING]\n"
+            "You have been provided with real credentials and variables for authenticated testing. Use them as follows:\n\n"
+            "- Include these credentials in HTTP headers (e.g., Authorization: Bearer <token>, X-API-Key: <api_key>)\n"
+            "- Use real username/password values in request bodies for login/authentication endpoints\n"
+            "- Apply these values to query parameters where relevant (e.g., user_id, account_id)\n"
+            "- Generate tests that exercise authenticated API flows, privilege escalation scenarios, BOLA/BFLA attacks, and account enumeration\n\n"
+            "Available credentials and variables:\n"
+        )
+        if isinstance(instructions, dict) and "variables" in instructions:
+            variables = instructions["variables"]
+            if isinstance(variables, dict):
+                for key, value in variables.items():
+                    greybox_context += f"- {key}: {value}\n"
+        greybox_context += "\nGenerate tests that demonstrate real exploitation using these values."
+        user_prompt += greybox_context
+
     result = await call_llm(sys_prompt, user_prompt, temperature=0.7)
     try:
         data = json.loads(result)

src/secnodeapi/agents/coordinator.py

@@ -1,6 +1,9 @@
 """Attack graph subsystem."""
 
 from .engine import AttackGraphEngine
+from .networkx_engine import NetworkXAttackGraph, AttackGraphManager
 from .models import AttackEdge, AttackNode, AttackPath
 
+__all__ = ["AttackNode", "AttackEdge", "AttackPath", "AttackGraphEngine", "NetworkXAttackGraph", "AttackGraphManager"]
+
 __all__ = ["AttackNode", "AttackEdge", "AttackPath", "AttackGraphEngine"]

src/secnodeapi/ai/generate.py

@@ -0,0 +1,189 @@
+"""NetworkX-powered attack graph engine with Postgres persistence."""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Set
+import json
+
+import networkx as nx
+
+from .models import AttackEdge, AttackPath
+
+
+class NetworkXAttackGraph:
+    """Attack graph using NetworkX for graph algorithms."""
+    
+    def __init__(self, session_id: str):
+        self.session_id = session_id
+        self._graph: nx.DiGraph = nx.DiGraph()
+    
+    def add_edge(self, edge: AttackEdge) -> None:
+        """Add an edge to the graph."""
+        self._graph.add_edge(
+            edge.from_node,
+            edge.to_node,
+            vulnerability=edge.vulnerability,
+            confidence=edge.confidence,
+            evidence=edge.evidence,
+            success=edge.success,
+        )
+    
+    def nodes(self) -> List[str]:
+        """Get all nodes."""
+        return list(self._graph.nodes())
+    
+    def edges(self) -> List[AttackEdge]:
+        """Get all edges as AttackEdge objects."""
+        result = []
+        for from_node, to_node, data in self._graph.edges(data=True):
+            result.append(AttackEdge(
+                from_node=from_node,
+                to_node=to_node,
+                vulnerability=data.get("vulnerability", ""),
+                confidence=data.get("confidence", 0.0),
+                evidence=data.get("evidence", ""),
+                success=data.get("success", False),
+            ))
+        return result
+    
+    def best_paths(self, max_paths: int = 5) -> List[AttackPath]:
+        """Find best attack paths using NetworkX algorithms."""
+        if not nx.is_weakly_connected(self._graph):
+            # Handle disconnected components
+            subgraphs = [self._graph.subgraph(c).copy() 
+                        for c in nx.weakly_connected_components(self._graph)]
+            paths = []
+            for subgraph in subgraphs:
+                paths.extend(self._find_paths_in_subgraph(subgraph, max_paths))
+            paths.sort(key=lambda p: p.score, reverse=True)
+            return paths[:max_paths]
+        
+        return self._find_paths_in_subgraph(self._graph, max_paths)
+    
+    def _find_paths_in_subgraph(self, graph: nx.DiGraph, max_paths: int) -> List[AttackPath]:
+        """Find paths in a subgraph using NetworkX algorithms."""
+        all_paths: List[AttackPath] = []
+        
+        # Find start nodes (nodes with no incoming edges from 'start')
+        start_nodes = [n for n in graph.nodes() if n == "start" or 
+                      not any(graph.predecessors(n))]
+        
+        for start in start_nodes:
+            # Use simple DFS to find all simple paths
+            for node in graph.nodes():
+                if node == start or start == "start":
+                    try:
+                        for path in nx.all_simple_paths(graph, start, node, cutoff=6):
+                            if len(path) > 1:
+                                edges = []
+                                for i in range(len(path) - 1):
+                                    edge_data = graph.get_edge_data(path[i], path[i+1])
+                                    if edge_data:
+                                        edges.append(AttackEdge(
+                                            from_node=path[i],
+                                            to_node=path[i+1],
+                                            vulnerability=edge_data.get("vulnerability", ""),
+                                            confidence=edge_data.get("confidence", 0.0),
+                                            evidence=edge_data.get("evidence", ""),
+                                            success=edge_data.get("success", False),
+                                        ))
+                                
+                                score = self._score_path(edges)
+                                all_paths.append(AttackPath(
+                                    nodes=path,
+                                    edges=edges,
+                                    score=score,
+                                ))
+                    except nx.NetworkXNoPath:
+                        continue
+        
+        all_paths.sort(key=lambda p: p.score, reverse=True)
+        return all_paths[:max_paths]
+    
+    @staticmethod
+    def _score_path(edges: List[AttackEdge]) -> float:
+        """Score a path based on confidence and depth."""
+        if not edges:
+            return 0.0
+        
+        # Filter to only successful edges
+        successful = [e for e in edges if e.success]
+        if not successful:
+            return 0.0
+        
+        depth_bonus = len(successful) * 0.15
+        avg_confidence = sum(e.confidence for e in successful) / len(successful)
+        return avg_confidence + depth_bonus
+    
+    def successful_nodes(self) -> List[str]:
+        """Get nodes reachable via successful edges."""
+        return [n for n in self._graph.nodes() 
+                if self._graph.in_degree(n) > 0 and 
+                any(self._graph.edges(u, v, data=True).get("success", False) 
+                    for u, v in self._graph.in_edges(n))]
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Serialize graph to dict for Postgres storage."""
+        return {
+            "session_id": self.session_id,
+            "nodes": list(self._graph.nodes()),
+            "edges": [
+                {
+                    "from_node": u,
+                    "to_node": v,
+                    **data,
+                }
+                for u, v, data in self._graph.edges(data=True)
+            ],
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "NetworkXAttackGraph":
+        """Deserialize graph from dict."""
+        graph = cls(data.get("session_id", ""))
+        for node in data.get("nodes", []):
+            graph._graph.add_node(node)
+        for edge in data.get("edges", []):
+            graph._graph.add_edge(
+                edge["from_node"],
+                edge["to_node"],
+                vulnerability=edge.get("vulnerability", ""),
+                confidence=edge.get("confidence", 0.0),
+                evidence=edge.get("evidence", ""),
+                success=edge.get("success", False),
+            )
+        return graph
+
+
+class AttackGraphManager:
+    """Manages multiple attack graphs with Postgres persistence."""
+    
+    def __init__(self, storage=None):
+        """Initialize with optional Postgres storage."""
+        self._graphs: Dict[str, NetworkXAttackGraph] = {}
+        self._storage = storage  # Optional storage adapter
+    
+    def get_graph(self, session_id: str) -> NetworkXAttackGraph:
+        """Get or create graph for session."""
+        if session_id not in self._graphs:
+            # Try to load from storage
+            if self._storage:
+                data = self._storage.load_graph(session_id)
+                if data:
+                    self._graphs[session_id] = NetworkXAttackGraph.from_dict(data)
+                else:
+                    self._graphs[session_id] = NetworkXAttackGraph(session_id)
+            else:
+                self._graphs[session_id] = NetworkXAttackGraph(session_id)
+        return self._graphs[session_id]
+    
+    def save_graph(self, session_id: str) -> None:
+        """Persist graph to storage."""
+        if self._storage and session_id in self._graphs:
+            self._storage.save_graph(self._graphs[session_id].to_dict())
+    
+    def delete_graph(self, session_id: str) -> None:
+        """Delete graph from memory and storage."""
+        if session_id in self._graphs:
+            del self._graphs[session_id]
+        if self._storage:
+            self._storage.delete_graph(session_id)