feat(cli): standardize --json output and implement security sanitization for error logs and audit events

Author: Deadpool2000

packages/agent-mesh/src/agentmesh/cli/main.py

@@ -313,7 +313,6 @@ def register(agent_dir: str, name: str = None, output_json: bool = False):
             "status": "success",
             "agent_name": agent_name,
             "did": identity.did,
-            "public_key": identity.public_key,
             "identity_file": str(identity_file)
         }, indent=2))
     else:
@@ -457,9 +456,16 @@ def audit(agent: str, limit: int, fmt: str, output_json: bool):
         entries = [e for e in entries if e["agent"] == agent]
 
     entries = entries[:limit]
+    
+    # Sanitize entries for JSON output to prevent information leakage/injection
+    allowed_keys = {"timestamp", "agent", "action", "status"}
+    sanitized_entries = [
+        {k: v for k, v in e.items() if k in allowed_keys}
+        for e in entries
+    ]
 
     if output_json or fmt == "json":
-        click.echo(json.dumps(entries, indent=2))
+        click.echo(json.dumps(sanitized_entries, indent=2))
     else:
         console.print("\n[bold blue]📋 Audit Log[/bold blue]\n")
         table = Table(box=box.SIMPLE)

packages/agent-os/modules/control-plane/acp-cli.py

@@ -170,16 +170,24 @@ def cmd_audit_show(args, control_plane):
         events = recorder.get_recent_events(limit=args.limit or 10)
 
         if args.format == "json" or getattr(args, "json", False):
-            print(json.dumps(events, indent=2))
+            # Sanitize events to prevent information disclosure
+            allowed_keys = {"timestamp", "event_type", "agent_id", "status"}
+            sanitized_events = [
+                {k: v for k, v in event.items() if k in allowed_keys}
+                for event in events
+            ]
+            print(json.dumps(sanitized_events, indent=2))
         else:
             print(f"Recent Audit Events (last {len(events)}):")
             for event in events:
                 print(f"  [{event.get('timestamp')}] {event.get('event_type')}: {event.get('agent_id')}")
     except Exception as e:
+        is_known = isinstance(e, (ValueError, PermissionError))
+        msg = str(e) if is_known else "Failed to retrieve audit logs"
         if getattr(args, "json", False):
-            print(json.dumps({"error": str(e)}, indent=2))
+            print(json.dumps({"error": msg, "type": e.__class__.__name__ if is_known else "InternalError"}, indent=2))
         else:
-            print(f"Error: {e}")
+            print(f"Error: {msg}")
 
 
 def cmd_benchmark_run(args):
@@ -243,10 +251,12 @@ def main():
             return 1
 
     except Exception as e:
+        is_known = isinstance(e, (ValueError, PermissionError, FileNotFoundError))
+        msg = str(e) if is_known else "An internal error occurred"
         if getattr(args, "json", False):
-            print(json.dumps({"error": str(e)}, indent=2))
+            print(json.dumps({"error": msg, "type": e.__class__.__name__ if is_known else "InternalError"}, indent=2))
         else:
-            print(f"Error: {e}")
+            print(f"Error: {msg}")
         return 1
 
     return 0

packages/agent-os/src/agent_os/cli/__init__.py

@@ -1220,10 +1220,19 @@ def main() -> int:
     except KeyboardInterrupt:
         return 130
     except Exception as e:
+        # Sanitize exception message to avoid leaking internal details
+        # For public output, we prefer generic messages for unexpected exceptions
+        is_known_error = isinstance(e, (FileNotFoundError, ValueError, PermissionError))
+        error_msg = str(e) if is_known_error else "An internal error occurred. Use AGENTOS_DEBUG=1 for details."
+        
         if getattr(args, "json", False) or (hasattr(args, "format") and args.format == "json"):
-            print(json.dumps({"error": str(e)}, indent=2))
+            print(json.dumps({
+                "status": "error",
+                "message": error_msg,
+                "error_type": e.__class__.__name__ if is_known_error else "InternalError"
+            }, indent=2))
         else:
-            print(format_error(str(e)))
+            print(format_error(error_msg))
             if os.environ.get("AGENTOS_DEBUG"):
                 import traceback
                 traceback.print_exc()

packages/agent-os/src/agent_os/cli/mcp_scan.py

@@ -159,97 +159,106 @@ def main(argv: List[str] | None = None) -> int:
 
     output_format = "json" if getattr(args, "json", False) or getattr(args, "format", "table") == "json" else "table"
 
-    if args.command == "scan":
-        findings = scan_config(config_path, args.server)
-        
-        if args.severity:
-            findings = [f for f in findings if f.severity == args.severity or f.severity == "critical"]
+    try:
+        if args.command == "scan":
+            findings = scan_config(config_path, args.server)
+            
+            if args.severity:
+                findings = [f for f in findings if f.severity == args.severity or f.severity == "critical"]
 
-        if output_format == "json":
-            print(json.dumps([f.to_dict() for f in findings], indent=2))
-        elif output_format == "table":
-            table = Table(title=f"Security Scan: {args.config}")
-            table.add_column("Server", style="cyan")
-            table.add_column("Severity", style="bold")
-            table.add_column("Category", style="dim")
-            table.add_column("Finding")
-
-            for f in findings:
-                sev_color = "red" if f.severity == "critical" else "yellow"
-                table.add_row(f.server, f"[{sev_color}]{f.severity.upper()}[/{sev_color}]", f.category, f.message)
-
-            console.print(table)
-        
-        return 1 if any(f.severity == "critical" for f in findings) else 0
+            if output_format == "json":
+                print(json.dumps([f.to_dict() for f in findings], indent=2))
+            elif output_format == "table":
+                table = Table(title=f"Security Scan: {args.config}")
+                table.add_column("Server", style="cyan")
+                table.add_column("Severity", style="bold")
+                table.add_column("Category", style="dim")
+                table.add_column("Finding")
+
+                for f in findings:
+                    sev_color = "red" if f.severity == "critical" else "yellow"
+                    table.add_row(f.server, f"[{sev_color}]{f.severity.upper()}[/{sev_color}]", f.category, f.message)
+
+                console.print(table)
+            
+            return 1 if any(f.severity == "critical" for f in findings) else 0
 
-    elif args.command == "fingerprint":
-        fingerprints = get_fingerprints(config_path)
-        
-        if args.compare:
-            with open(args.compare) as f:
-                saved = json.load(f)
+        elif args.command == "fingerprint":
+            fingerprints = get_fingerprints(config_path)
 
-            diffs = {}
-            for name, h in fingerprints.items():
-                if name not in saved:
-                    diffs[name] = "new"
-                elif saved[name] != h:
-                    diffs[name] = "changed"
+            if args.compare:
+                with open(args.compare) as f:
+                    saved = json.load(f)
+                
+                diffs = {}
+                for name, h in fingerprints.items():
+                    if name not in saved:
+                        diffs[name] = "new"
+                    elif saved[name] != h:
+                        diffs[name] = "changed"
+                
+                if output_format == "json":
+                    print(json.dumps({"current": fingerprints, "diffs": diffs}, indent=2))
+                else:
+                    print(f"Comparison results for {args.config}:")
+                    for name, status in diffs.items():
+                        print(f"  {name}: {status}")
+                    if not diffs:
+                        print("  Identical fingerprints.")
+            
+            elif args.output:
+                with open(args.output, "w") as f:
+                    json.dump(fingerprints, f, indent=2)
+                if output_format != "json":
+                    print(f"Fingerprints saved to {args.output}")
+                else:
+                    print(json.dumps({"status": "success", "file": args.output}, indent=2))
 
-            if output_format == "json":
-                print(json.dumps({"current": fingerprints, "diffs": diffs}, indent=2))
-            else:
-                print(f"Comparison results for {args.config}:")
-                for name, status in diffs.items():
-                    print(f"  {name}: {status}")
-                if not diffs:
-                    print("  Identical fingerprints.")
-        
-        elif args.output:
-            with open(args.output, "w") as f:
-                json.dump(fingerprints, f, indent=2)
-            if output_format != "json":
-                print(f"Fingerprints saved to {args.output}")
             else:
-                print(json.dumps({"status": "success", "file": args.output}, indent=2))
-        
-        else:
-            if output_format == "json":
-                print(json.dumps(fingerprints, indent=2))
+                if output_format == "json":
+                    print(json.dumps(fingerprints, indent=2))
+                else:
+                    for name, h in fingerprints.items():
+                        print(f"{name:20} {h}")
+
+        elif args.command == "report":
+            findings = scan_config(config_path)
+            fingerprints = get_fingerprints(config_path)
+            
+            report = {
+                "config": str(config_path),
+                "summary": {
+                    "total_servers": len(fingerprints),
+                    "total_findings": len(findings),
+                    "critical": len([f for f in findings if f.severity == "critical"]),
+                    "warning": len([f for f in findings if f.severity == "warning"])
+                },
+                "findings": [f.to_dict() for f in findings],
+                "fingerprints": fingerprints
+            }
+            
+            if output_format == "json" or getattr(args, "format", "markdown") == "json":
+                print(json.dumps(report, indent=2))
             else:
-                for name, h in fingerprints.items():
-                    print(f"{name:20} {h}")
+                # Simple markdown report
+                print(f"# Security Report: {args.config}")
+                print()
+                print(f"- Total Servers: {report['summary']['total_servers']}")
+                print(f"- Total Findings: {report['summary']['total_findings']}")
+                print()
+                print("## Findings")
+                for f in findings:
+                    print(f"- **{f.server}** ({f.severity.upper()}): {f.message}")
 
-    elif args.command == "report":
-        findings = scan_config(config_path)
-        fingerprints = get_fingerprints(config_path)
-        
-        report = {
-            "config": str(config_path),
-            "summary": {
-                "total_servers": len(fingerprints),
-                "total_findings": len(findings),
-                "critical": len([f for f in findings if f.severity == "critical"]),
-                "warning": len([f for f in findings if f.severity == "warning"])
-            },
-            "findings": [f.to_dict() for f in findings],
-            "fingerprints": fingerprints
-        }
-        
-        if output_format == "json" or getattr(args, "format", "markdown") == "json":
-            print(json.dumps(report, indent=2))
+        return 0
+    except Exception as e:
+        is_known = isinstance(e, (FileNotFoundError, ValueError, yaml.YAMLError))
+        msg = str(e) if is_known else "An error occurred during scanning"
+        if output_format == "json":
+            print(json.dumps({"status": "error", "message": msg, "type": e.__class__.__name__ if is_known else "InternalError"}, indent=2))
         else:
-            # Simple markdown report
-            print(f"# Security Report: {args.config}")
-            print()
-            print(f"- Total Servers: {report['summary']['total_servers']}")
-            print(f"- Total Findings: {report['summary']['total_findings']}")
-            print()
-            print("## Findings")
-            for f in findings:
-                print(f"- **{f.server}** ({f.severity.upper()}): {f.message}")
-
-    return 0
+            print(f"Error: {msg}")
+        return 1
 
 
 if __name__ == "__main__":