feat(cli): standardize and harden secure JSON output across all ecosystem tools

Author: Deadpool2000

packages/agent-compliance/src/agent_compliance/cli/main.py

@@ -21,17 +21,25 @@ def cmd_verify(args: argparse.Namespace) -> int:
     """Run governance verification."""
     from agent_compliance.verify import GovernanceVerifier
 
-    verifier = GovernanceVerifier()
-    attestation = verifier.verify()
+    try:
+        verifier = GovernanceVerifier()
+        attestation = verifier.verify()
 
-    if args.json:
-        print(attestation.to_json())
-    elif args.badge:
-        print(attestation.badge_markdown())
-    else:
-        print(attestation.summary())
+        if args.json:
+            print(attestation.to_json())
+        elif args.badge:
+            print(attestation.badge_markdown())
+        else:
+            print(attestation.summary())
 
-    return 0 if attestation.passed else 1
+        return 0 if attestation.passed else 1
+    except Exception as e:
+        if args.json:
+            import json
+            print(json.dumps({"status": "fail", "error": "Governance verification failed", "type": "InternalError"}, indent=2))
+        else:
+            print(f"Error: {e}", file=sys.stderr)
+        return 1
 
 
 def cmd_integrity(args: argparse.Namespace) -> int:
@@ -73,7 +81,11 @@ def cmd_integrity(args: argparse.Namespace) -> int:
 
         return 0 if report.passed else 1
     except Exception as e:
-        print(f"Error: {e}", file=sys.stderr)
+        if args.json:
+            import json
+            print(json.dumps({"status": "error", "message": "Integrity manifest processing failed", "type": "InternalError"}, indent=2))
+        else:
+            print(f"Error: {e}", file=sys.stderr)
         return 1
 
 
@@ -99,7 +111,11 @@ def cmd_lint_policy(args: argparse.Namespace) -> int:
             return 1
         return 0 if result.passed else 1
     except Exception as e:
-        print(f"Error: {e}", file=sys.stderr)
+        if args.json:
+            import json
+            print(json.dumps({"status": "error", "message": "Policy linting failed", "type": "InternalError"}, indent=2))
+        else:
+            print(f"Error: {e}", file=sys.stderr)
         return 1
 
 

packages/agent-os/modules/iatp/iatp/cli.py

@@ -57,13 +57,20 @@ def verify(manifest_path: str, verbose: bool, output_json: bool):
         # Validate using Pydantic model
         try:
             manifest = CapabilityManifest(**manifest_data)
-        except Exception as e:
+        except (ValueError, KeyError, PermissionError) as e:
             if output_json:
-                print(json.dumps({"status": "fail", "error": f"Schema validation failed: {str(e)}"}, indent=2))
+                print(json.dumps({"status": "fail", "error": f"Manifest validation failed: {str(e)}", "type": e.__class__.__name__}, indent=2))
             else:
-                click.echo("\n❌ Schema validation failed:", err=True)
+                click.echo("\n❌ Validation failed:", err=True)
                 click.echo(f"   {str(e)}", err=True)
             sys.exit(1)
+        except Exception:
+            err_msg = "An internal error occurred while validating the manifest."
+            if output_json:
+                print(json.dumps({"status": "fail", "error": err_msg, "type": "InternalError"}, indent=2))
+            else:
+                click.echo(f"\n❌ Error: {err_msg}", err=True)
+            sys.exit(1)
 
         # Perform logical contradiction checks
         errors = []
@@ -152,10 +159,18 @@ def verify(manifest_path: str, verbose: bool, output_json: bool):
             click.echo(f"❌ Invalid JSON: {str(e)}", err=True)
         sys.exit(1)
     except Exception as e:
+        # Sanitize exception message to avoid leaking internal details
+        is_known = isinstance(e, (FileNotFoundError, json.JSONDecodeError, ValueError, PermissionError))
+        error_msg = str(e) if is_known else "An internal error occurred during verification."
+        
         if output_json:
-            print(json.dumps({"status": "fail", "error": f"Validation error: {str(e)}"}, indent=2))
+            print(json.dumps({
+                "status": "fail",
+                "error": error_msg,
+                "type": e.__class__.__name__ if is_known else "InternalError"
+            }, indent=2))
         else:
-            click.echo(f"❌ Validation error: {str(e)}", err=True)
+            click.echo(f"❌ Error: {error_msg}", err=True)
         sys.exit(1)
 
 
@@ -261,11 +276,19 @@ def scan(agent_url: str, timeout: int, verbose: bool, output_json: bool):
                     click.echo(f"   SLA Latency: {manifest.capabilities.sla_latency}")
 
     except Exception as e:
+        # Sanitize error message based on exception type to prevent info leakage
+        is_known = isinstance(e, (httpx.RequestError, json.JSONDecodeError, ValueError, PermissionError))
+        err_msg = str(e) if is_known else "An internal error occurred during agent scan"
+        
         if output_json:
-            print(json.dumps({"status": "fail", "error": str(e)}, indent=2))
+            print(json.dumps({
+                "status": "fail",
+                "error": err_msg,
+                "type": e.__class__.__name__ if is_known else "InternalError"
+            }, indent=2))
         else:
-            click.echo(f"\n❌ Scan error: {str(e)}", err=True)
-            if verbose:
+            click.echo(f"\n❌ Scan error: {err_msg}", err=True)
+            if verbose and not is_known:
                 import traceback
                 click.echo(traceback.format_exc(), err=True)
         sys.exit(1)

packages/agent-os/modules/mcp-kernel-server/src/mcp_kernel_server/cli.py

@@ -230,32 +230,42 @@ def main():
     server = KernelMCPServer(config)
 
     # Run
-    if args.stdio:
-        logger.info("Starting MCP Kernel Server with stdio transport")
-        logger.info(f"Policy mode: {args.policy_mode}")
-        logger.info("Tools: cmvk_verify, kernel_execute, iatp_sign, iatp_verify, iatp_reputation")
-        logger.info("Prompts: governed_agent, verify_claim, safe_execution")
-        
-        try:
+    try:
+        if args.stdio:
+            logger.info("Starting MCP Kernel Server with stdio transport")
+            logger.info(f"Policy mode: {args.policy_mode}")
+            logger.info("Tools: cmvk_verify, kernel_execute, iatp_sign, iatp_verify, iatp_reputation")
+            logger.info("Prompts: governed_agent, verify_claim, safe_execution")
+            
             asyncio.run(server.run_stdio())
-        except KeyboardInterrupt:
-            logger.info("Shutting down...")
-        except Exception as e:
-            logger.exception("Server error")
-            return 1
-    else:
-        # HTTP transport
-        logger.info(f"Starting MCP Kernel Server on http://{args.host}:{args.port}")
-        logger.info(f"Policy mode: {args.policy_mode}")
-        logger.info("Tools: cmvk_verify, kernel_execute, iatp_sign, iatp_verify, iatp_reputation")
-        logger.info("Prompts: governed_agent, verify_claim, safe_execution")
-        logger.info("Press Ctrl+C to stop")
-        
-        try:
+        else:
+            # HTTP transport
+            logger.info(f"Starting MCP Kernel Server on http://{args.host}:{args.port}")
+            logger.info(f"Policy mode: {args.policy_mode}")
+            logger.info("Tools: cmvk_verify, kernel_execute, iatp_sign, iatp_verify, iatp_reputation")
+            logger.info("Prompts: governed_agent, verify_claim, safe_execution")
+            logger.info("Press Ctrl+C to stop")
+            
             asyncio.run(server.start())
             asyncio.get_event_loop().run_forever()
-        except KeyboardInterrupt:
-            logger.info("Shutting down...")
+            
+    except KeyboardInterrupt:
+        logger.info("Shutting down...")
+    except Exception as e:
+        # Sanitize startup errors for JSON mode to prevent info leakage
+        is_known = isinstance(e, (ValueError, PermissionError, OSError))
+        msg = str(e) if is_known else "An internal error occurred during server startup"
+        
+        if getattr(args, "json", False):
+            import json
+            print(json.dumps({
+                "status": "error",
+                "message": msg,
+                "type": e.__class__.__name__ if is_known else "InternalError"
+            }, indent=2))
+        else:
+            logger.exception(f"Server error: {msg}")
+        return 1
 
     return 0