from __future__ import annotations import argparse import json import sys from dataclasses import asdict from pathlib import Path from typing import Any, Dict, List from layer0 import layer0_entry from layer0.shadow_classifier import ShadowEvalResult from . import __version__ as WAF_INTEL_VERSION from .orchestrator import WAFInsight, WAFIntelligence def _insight_to_dict(insight: WAFInsight) -> Dict[str, Any]: """Convert a WAFInsight dataclass into a plain dict.""" return asdict(insight) def _has_error(insights: List[WAFInsight]) -> bool: """Return True if any violation is error-severity.""" for insight in insights: if insight.violation and insight.violation.severity == "error": return True return False def run_cli(argv: List[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="python -m mcp.waf_intelligence", description="Analyze Cloudflare WAF Terraform configs and produce curated security + compliance insights.", ) parser.add_argument( "--file", "-f", required=True, help="Path to the Terraform WAF file (e.g. terraform/waf.tf)", ) parser.add_argument( "--limit", "-n", type=int, default=3, help="Maximum number of high-priority insights to return (default: 3)", ) parser.add_argument( "--format", "-o", choices=["text", "json"], default="text", help="Output format: text (human-readable) or json (machine-readable). Default: text.", ) parser.add_argument( "--fail-on-error", action="store_true", help="Exit with non-zero code if any error-severity violations are found.", ) parser.add_argument( "--version", action="version", version=f"%(prog)s {WAF_INTEL_VERSION}", ) args = parser.parse_args(argv) # Layer 0: pre-boot Shadow Eval gate. routing_action, shadow = layer0_entry( f"waf_intel_cli file={args.file} limit={args.limit}" ) if routing_action != "HANDOFF_TO_LAYER1": _render_layer0_block(routing_action, shadow) return 1 path = Path(args.file) if not path.exists(): print(f"[error] file not found: {path}", file=sys.stderr) return 1 intel = WAFIntelligence() insights = intel.analyze_and_recommend(str(path), limit=args.limit) if args.format == "json": payload = { "file": str(path), "insights": [_insight_to_dict(insight) for insight in insights], } print(json.dumps(payload, indent=2)) if args.fail_on_error and _has_error(insights): print( "[waf_intel] error-severity violations present, failing as requested.", file=sys.stderr, ) return 2 return 0 print(f"\nWAF Intelligence Report for: {path}\n{'-' * 72}") if not insights: print( "No high-severity, high-confidence issues detected based on current heuristics." ) return 0 for idx, insight in enumerate(insights, start=1): print(f"\nInsight #{idx}") print("-" * 40) if insight.violation: violation = insight.violation print(f"Problem : {violation.message}") print(f"Severity : {violation.severity.upper()}") print(f"Confidence: {int(violation.confidence * 100)}%") if violation.location: print(f"Location : {violation.location}") if violation.hint: print(f"Remediate : {violation.hint}") if insight.suggested_rule: rule = insight.suggested_rule print("\nSuggested Rule:") print(f" Name : {rule.name}") print(f" Severity: {rule.severity.upper()}") print(f" Impact : {int(rule.impact_score * 100)}%") print(f" Effort : {int(rule.effort_score * 100)}%") print(f" Summary : {rule.description}") if insight.mappings: print("\nCompliance Mapping:") for mapping in insight.mappings: print( f" - {mapping.framework} {mapping.control_id}: {mapping.description}" ) print() if args.fail_on_error and _has_error(insights): print( "[waf_intel] error-severity violations present, failing as requested.", file=sys.stderr, ) return 2 return 0 def main() -> None: raise SystemExit(run_cli()) if __name__ == "__main__": main() def _render_layer0_block(routing_action: str, shadow: ShadowEvalResult) -> None: """ Minimal user-facing responses for Layer 0 decisions. """ if routing_action == "FAIL_CLOSED": print("Layer 0: cannot comply with this request.", file=sys.stderr) return if routing_action == "HANDOFF_TO_GUARDRAILS": reason = shadow.reason or "governance_violation" print( f"Layer 0: governance violation detected ({reason}).", file=sys.stderr, ) return if routing_action == "PROMPT_FOR_CLARIFICATION": print( "Layer 0: request is ambiguous. Please add specifics before rerunning.", file=sys.stderr, ) return print("Layer 0: unrecognized routing action; refusing request.", file=sys.stderr)