Merge pull request #133 from NuGuardAI/ranjan/redteam-v2

feat: redteam v2, behavior/redteam public APIs, remediation output, v0.8.5

Author: rangoel-nu

npm/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nuguardai/nuguard",
-  "version": "0.8.4",
+  "version": "0.8.5",
   "description": "AI Application Security — SBOM generation, static analysis, behavioral validation, and adversarial red-team testing for AI agents and LLM-powered applications.",
   "keywords": [
     "mcp",

nuguard/__init__.py

@@ -1,3 +1,3 @@
 """NuGuard AI Security CLI — open-source AI penetration testing platform."""
 
-__version__ = "0.8.4"
+__version__ = "0.8.5"

nuguard/behavior/models.py

@@ -334,6 +334,14 @@ class BehaviorRunResult(BaseModel):
     coverage_mapping_diagnostics: dict[str, Any] = Field(default_factory=dict)
     effective_endpoint: str = ""
     target_endpoint_source: str = "config"
+    remediation_plan: list[RemediationArtefact] = Field(default_factory=list)
+    """Concrete, SBOM-node-specific remediation artefacts for ``findings``.
+
+    Populated by :func:`nuguard.behavior.public_api.run_behavior_scenarios`
+    (best-effort) via ``RemediationSynthesizer``. Empty when called directly
+    through :class:`~nuguard.behavior.runner.BehaviorRunner`, which does not
+    synthesize remediation itself.
+    """
 
 
 class BehaviorAnalysisResult(BaseModel):

nuguard/behavior/public_api.py

@@ -22,6 +22,7 @@
     BehaviorAnalysisResult,
     BehaviorRunResult,
     BehaviorScenario,
+    RemediationArtefact,
 )
 from nuguard.behavior.runner import BehaviorRunner
 from nuguard.common.discovery import DiscoveredProfile
@@ -79,6 +80,33 @@ class BehaviorRunRequest(BaseModel):
     pre_scan_profile: DiscoveredProfile | None = None
 
 
+async def _synthesize_behavior_remediation_plan(
+    findings: list[dict],
+    *,
+    sbom: "AiSbomDocument | None",
+    policy: "CognitivePolicy | None",
+    llm_client: "LLMClient | None",
+) -> list[RemediationArtefact]:
+    """Best-effort structured remediation for a plain list of finding dicts.
+
+    Mirrors the CLI's ``_build_redteam_remediation_plan`` (which reuses this
+    same synthesizer for redteam findings): remediation synthesis enriches
+    the result but must never fail the run, so exceptions are logged and
+    swallowed. Returns ``[]`` when there is no SBOM or no findings to
+    synthesize against.
+    """
+    if sbom is None or not findings:
+        return []
+    try:
+        from nuguard.behavior.remediation import RemediationSynthesizer  # noqa: PLC0415
+
+        synthesizer = RemediationSynthesizer(sbom=sbom, policy=policy, llm_client=llm_client)
+        return await synthesizer.synthesize_findings_async(findings)
+    except Exception as exc:  # noqa: BLE001
+        _log.warning("run_behavior_scenarios: remediation synthesis failed: %s", exc)
+        return []
+
+
 async def run_behavior_scenarios(
     request: BehaviorRunRequest,
     *,
@@ -90,7 +118,12 @@ async def run_behavior_scenarios(
 ) -> BehaviorRunResult:
     """Run a list of behavior scenarios from a JSON-safe request.
 
-    Thin wrapper around ``BehaviorRunner(...).run(...)``.
+    Thin wrapper around ``BehaviorRunner(...).run(...)``. Additionally
+    synthesizes ``result.remediation_plan`` — concrete, SBOM-node-specific
+    remediation artefacts — from the run's findings, the same way
+    :meth:`~nuguard.behavior.analyzer.BehaviorAnalyzer.analyze` does for the
+    full static+dynamic pipeline. This is best-effort enrichment: it never
+    raises, and simply leaves ``remediation_plan`` empty on failure.
     """
     _log.debug("run_behavior_scenarios: %d scenario(s)", len(request.scenarios))
     runner = BehaviorRunner(
@@ -101,7 +134,11 @@ async def run_behavior_scenarios(
         llm_client=llm_client,
         judge_cache=judge_cache,
     )
-    return await runner.run(request.scenarios, pre_scan_profile=request.pre_scan_profile)
+    result = await runner.run(request.scenarios, pre_scan_profile=request.pre_scan_profile)
+    result.remediation_plan = await _synthesize_behavior_remediation_plan(
+        result.findings, sbom=sbom, policy=policy, llm_client=llm_client
+    )
+    return result
 
 
 async def discover_behavior_profile(

nuguard/redteam/public_api.py

@@ -21,6 +21,7 @@
 
 from pydantic import BaseModel, Field
 
+from nuguard.behavior.models import RemediationArtefact
 from nuguard.common.auth import AuthConfig
 from nuguard.common.logging import get_logger
 from nuguard.config import RedteamFindingTriggers
@@ -124,6 +125,54 @@ class RedteamRunResult(BaseModel):
     resolved_chat_path_source: str
     catalog_coverage: dict[str, Any] | None = None
     coverage_tracker: dict[str, Any] | None = None
+    remediation_plan: list[RemediationArtefact] = Field(default_factory=list)
+    """Concrete, SBOM-node-specific remediation artefacts for ``findings``.
+
+    Synthesized best-effort from ``findings`` via the same
+    ``RemediationSynthesizer`` the CLI uses to build its redteam report's
+    remediation plan (``nuguard.cli.commands.redteam._build_redteam_remediation_plan``),
+    with contextual LLM patch text when ``eval_llm`` is supplied to
+    :func:`run_redteam`. Empty when synthesis fails or no SBOM is available.
+    """
+
+
+async def _build_remediation_plan(
+    findings: list[Finding],
+    *,
+    sbom: "AiSbomDocument | None",
+    policy: "CognitivePolicy | None",
+    llm_client: "LLMClient | None",
+) -> list[RemediationArtefact]:
+    """Synthesize per-SBOM-node remediation artefacts from redteam findings.
+
+    Async counterpart to the CLI's ``_build_redteam_remediation_plan``
+    (``nuguard/cli/commands/redteam.py``) — uses ``synthesize_findings_async``
+    instead of the sync ``synthesize_findings`` since this runs inside
+    ``run_redteam``'s already-running event loop, so LLM patch calls need to
+    be awaited directly rather than silently skipped by the sync shim.
+    Best-effort: returns ``[]`` on missing SBOM, no findings, or any failure.
+    """
+    if sbom is None or not findings:
+        return []
+    try:
+        from nuguard.behavior.remediation import RemediationSynthesizer  # noqa: PLC0415
+
+        synthesizer = RemediationSynthesizer(sbom=sbom, policy=policy, llm_client=llm_client)
+        finding_dicts = [
+            {
+                "finding_id": f.finding_id,
+                "title": f.title,
+                "description": f.description or "",
+                "affected_component": f.affected_component or "unknown",
+                "severity": f.severity.value if hasattr(f.severity, "value") else str(f.severity),
+                "goal_type": f.goal_type or "",
+            }
+            for f in findings
+        ]
+        return await synthesizer.synthesize_findings_async(finding_dicts)
+    except Exception as exc:  # noqa: BLE001
+        _log.warning("run_redteam: remediation synthesis failed — skipping plan: %s", exc)
+        return []
 
 
 def _catalog_coverage_to_dict(report: "CoverageReport | None") -> dict[str, Any] | None:
@@ -235,6 +284,10 @@ async def run_redteam(
             )
         ]
 
+    remediation_plan = await _build_remediation_plan(
+        findings, sbom=sbom, policy=policy, llm_client=eval_llm
+    )
+
     coverage_tracker = getattr(orchestrator, "_coverage_tracker", None)
     return RedteamRunResult(
         findings=findings,
@@ -253,4 +306,5 @@ async def run_redteam(
         resolved_chat_path_source=orchestrator.resolved_chat_path_source,
         catalog_coverage=_catalog_coverage_to_dict(getattr(orchestrator, "catalog_coverage", None)),
         coverage_tracker=coverage_tracker.to_dict() if coverage_tracker is not None else None,
+        remediation_plan=remediation_plan,
     )

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "nuguard"
-version = "0.8.4"
+version = "0.8.5"
 description = "AI application security — SBOM generation, vulnerability scanning, behavioral validation, and adversarial red-teaming for AI Agents"
 readme = "README.md"
 license = { text = "Apache-2.0" }

smithery.yaml

@@ -1,7 +1,7 @@
 # Smithery manifest — Claude marketplace MCP server
 # https://smithery.ai/docs/config
 name: nuguard
-version: "0.8.4"
+version: "0.8.5"
 description: |
   AI Application Security — generate an AI Bill of Materials (AI-SBOM), run static
   analysis, behavioral validation, and adversarial red-team testing for AI agents

tests/behavior/test_public_api.py

@@ -130,7 +130,11 @@ async def test_analyze_behavior_propagates_exceptions():
 
 @pytest.mark.asyncio
 async def test_run_behavior_scenarios_constructs_runner_and_forwards_args():
-    sentinel_result = MagicMock(spec=BehaviorRunResult)
+    # A real (empty-findings) result rather than MagicMock(spec=...): the
+    # wrapper now reads result.findings and sets result.remediation_plan for
+    # remediation synthesis, and pydantic model classes don't expose field
+    # names via dir(), so a spec'd mock can't stand in for those attributes.
+    sentinel_result = BehaviorRunResult(run_id="run1")
     config = BehaviorConfig(target="http://localhost:9999")
     scenarios = [_scenario("a")]
     profile = DiscoveredProfile(customer_name="Bob", source="config")
@@ -161,6 +165,74 @@ async def test_run_behavior_scenarios_constructs_runner_and_forwards_args():
     assert [s.name for s in called_args[0]] == ["a"]
     assert called_kwargs["pre_scan_profile"] is profile
     assert result is sentinel_result
+    assert result.remediation_plan == []
+
+
+@pytest.mark.asyncio
+async def test_run_behavior_scenarios_populates_remediation_plan():
+    """Findings + a real SBOM should produce structured RemediationArtefact objects."""
+    from nuguard.behavior.models import RemediationArtefact, RemediationArtefactType
+
+    finding = {
+        "finding_id": "BA-004-1",
+        "title": "PII disclosed via datastore",
+        "description": "Agent leaked account_number in a response.",
+        "affected_component": "SupportAgent",
+        "severity": "high",
+    }
+    sentinel_result = BehaviorRunResult(run_id="run1", findings=[finding])
+    config = BehaviorConfig(target="http://localhost:9999")
+
+    fake_artefact = RemediationArtefact(
+        finding_ids=["BA-004-1"],
+        component="SupportAgent",
+        component_type="AGENT",
+        artefact_type=RemediationArtefactType.OUTPUT_GUARDRAIL,
+        priority="high",
+        rationale="Sensitive fields must not appear in agent responses.",
+    )
+
+    from types import SimpleNamespace
+
+    empty_sbom = SimpleNamespace(nodes=[], edges=[])
+
+    with (
+        patch("nuguard.behavior.public_api.BehaviorRunner") as mock_runner_cls,
+        patch("nuguard.behavior.remediation.RemediationSynthesizer.synthesize_findings_async") as mock_synth,
+    ):
+        mock_runner_cls.return_value.run = AsyncMock(return_value=sentinel_result)
+        mock_synth.return_value = [fake_artefact]
+
+        request = BehaviorRunRequest(config=config, scenarios=[_scenario("a")])
+        result = await run_behavior_scenarios(request, sbom=empty_sbom)
+
+    mock_synth.assert_awaited_once_with([finding])
+    assert result.remediation_plan == [fake_artefact]
+
+
+@pytest.mark.asyncio
+async def test_run_behavior_scenarios_remediation_synthesis_failure_is_swallowed():
+    """Remediation synthesis is best-effort — a failure must not fail the run."""
+    sentinel_result = BehaviorRunResult(
+        run_id="run1",
+        findings=[{"finding_id": "f1", "title": "t", "description": "d", "affected_component": "c", "severity": "low"}],
+    )
+    config = BehaviorConfig(target="http://localhost:9999")
+
+    from types import SimpleNamespace
+
+    with (
+        patch("nuguard.behavior.public_api.BehaviorRunner") as mock_runner_cls,
+        patch(
+            "nuguard.behavior.remediation.RemediationSynthesizer.synthesize_findings_async",
+            side_effect=RuntimeError("boom"),
+        ),
+    ):
+        mock_runner_cls.return_value.run = AsyncMock(return_value=sentinel_result)
+        request = BehaviorRunRequest(config=config, scenarios=[_scenario("a")])
+        result = await run_behavior_scenarios(request, sbom=SimpleNamespace(nodes=[], edges=[]))
+
+    assert result.remediation_plan == []
 
 
 # ---------------------------------------------------------------------------

tests/redteam/test_public_api.py

@@ -244,3 +244,80 @@ async def test_run_redteam_propagates_exceptions():
         request = RedteamRunRequest(target_url="http://target")
         with pytest.raises(RuntimeError, match="target unreachable"):
             await run_redteam(request, sbom=MagicMock())
+
+
+# ---------------------------------------------------------------------------
+# remediation_plan — structured, machine-actionable remediation
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_redteam_populates_remediation_plan():
+    """Findings + a real SBOM should produce structured RemediationArtefact objects,
+    via the same RemediationSynthesizer the CLI uses for its report's remediation plan."""
+    from types import SimpleNamespace
+
+    from nuguard.behavior.models import RemediationArtefact, RemediationArtefactType
+
+    findings = [_finding("data_exfiltration")]
+    mock_instance = _make_mock_orchestrator(findings)
+    fake_artefact = RemediationArtefact(
+        finding_ids=["f1"],
+        component="unknown",
+        component_type="AGENT",
+        artefact_type=RemediationArtefactType.OUTPUT_GUARDRAIL,
+        priority="high",
+        rationale="d",
+    )
+
+    with (
+        patch("nuguard.redteam.public_api.RedteamOrchestrator") as mock_cls,
+        patch(
+            "nuguard.behavior.remediation.RemediationSynthesizer.synthesize_findings_async"
+        ) as mock_synth,
+    ):
+        mock_cls.return_value = mock_instance
+        mock_synth.return_value = [fake_artefact]
+        request = RedteamRunRequest(target_url="http://target")
+        result = await run_redteam(request, sbom=SimpleNamespace(nodes=[], edges=[]))
+
+    mock_synth.assert_awaited_once()
+    (finding_dicts,), _ = mock_synth.await_args
+    assert finding_dicts[0]["finding_id"] == "f1"
+    assert finding_dicts[0]["goal_type"] == "data_exfiltration"
+    assert result.remediation_plan == [fake_artefact]
+
+
+@pytest.mark.asyncio
+async def test_run_redteam_remediation_plan_empty_without_findings():
+    mock_instance = _make_mock_orchestrator([])
+
+    with patch("nuguard.redteam.public_api.RedteamOrchestrator") as mock_cls:
+        mock_cls.return_value = mock_instance
+        request = RedteamRunRequest(target_url="http://target")
+        result = await run_redteam(request, sbom=MagicMock())
+
+    assert result.remediation_plan == []
+
+
+@pytest.mark.asyncio
+async def test_run_redteam_remediation_synthesis_failure_is_swallowed():
+    """Remediation synthesis is best-effort — a failure must not fail the run."""
+    findings = [_finding("data_exfiltration")]
+    mock_instance = _make_mock_orchestrator(findings)
+
+    with (
+        patch("nuguard.redteam.public_api.RedteamOrchestrator") as mock_cls,
+        patch(
+            "nuguard.behavior.remediation.RemediationSynthesizer.synthesize_findings_async",
+            side_effect=RuntimeError("boom"),
+        ),
+    ):
+        mock_cls.return_value = mock_instance
+        request = RedteamRunRequest(target_url="http://target")
+        from types import SimpleNamespace
+
+        result = await run_redteam(request, sbom=SimpleNamespace(nodes=[], edges=[]))
+
+    assert result.remediation_plan == []
+    assert len(result.findings) == 1