Clear Sonar + Codacy findings on PR 178

- S3516: drop invariant ``return 0`` from main(); CLI returns None.
- S2068: redaction test now reads the sensitive key + placeholder
  from public audit constants instead of hard-coding "password" /
  "shhh", with a NOSONAR justification on the lone fixture value.
- S1542: keep the AC_* plugin convention with a NOSONAR comment.
- S1192: extract _TOOLS_CALL_METHOD and _MIME_JSON constants.
- S1244: use math.isclose for the scheduler interval assertion.
- S5713: drop NotImplementedError from the except tuple — RuntimeError
  already covers it.
- Prospector pyflakes: remove the unused ``import logging``.
- Semgrep dangerous-subprocess: nosemgrep justifications on the two
  argv-list subprocess calls (no shell, sanitised argv).

Author: JE-Chen

je_auto_control/utils/mcp_server/__main__.py

@@ -50,8 +50,8 @@ def _build_parser() -> argparse.ArgumentParser:
     return parser
 
 
-def main(argv: list = None) -> int:
-    """CLI entry point. Returns the process exit code."""
+def main(argv: list = None) -> None:
+    """CLI entry point. Performs the requested action and returns ``None``."""
     parser = _build_parser()
     args = parser.parse_args(argv)
     if args.fake_backend:
@@ -61,9 +61,8 @@ def main(argv: list = None) -> int:
     listing_modes = (args.list_tools, args.list_resources, args.list_prompts)
     if any(listing_modes):
         _print_listings(args)
-        return 0
+        return
     start_mcp_stdio_server()
-    return 0
 
 
 def _print_listings(args: argparse.Namespace) -> None:
@@ -85,4 +84,4 @@ def _print_listings(args: argparse.Namespace) -> None:
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    main()

je_auto_control/utils/mcp_server/audit.py

@@ -57,21 +57,22 @@ def record(self, *, tool: str, arguments: Dict[str, Any],
                 handle.write(line + "\n")
 
 
-_REDACTED_KEYS = frozenset({"password", "token", "secret", "api_key",
+REDACTED_KEYS = frozenset({"password", "token", "secret", "api_key",
                             "authorization"})
+REDACTED_PLACEHOLDER = "<redacted>"
 
 
 def _sanitise(arguments: Dict[str, Any]) -> Dict[str, Any]:
-    """Replace obvious secret-like values with ``"<redacted>"``."""
+    """Replace obvious secret-like values with ``REDACTED_PLACEHOLDER``."""
     if not isinstance(arguments, dict):
         return arguments
     out: Dict[str, Any] = {}
     for key, value in arguments.items():
-        if key.lower() in _REDACTED_KEYS:
-            out[key] = "<redacted>"
+        if key.lower() in REDACTED_KEYS:
+            out[key] = REDACTED_PLACEHOLDER
         else:
             out[key] = value
     return out
 
 
-__all__ = ["AuditLogger"]
+__all__ = ["AuditLogger", "REDACTED_KEYS", "REDACTED_PLACEHOLDER"]

je_auto_control/utils/mcp_server/resources.py

@@ -15,6 +15,8 @@
 from dataclasses import dataclass
 from typing import Any, Callable, Dict, List, Optional
 
+_MIME_JSON = "application/json"
+
 
 @dataclass(frozen=True)
 class MCPResource:
@@ -90,7 +92,7 @@ def list(self) -> List[MCPResource]:
                 uri=f"{self.scheme}://files/{name}",
                 name=name,
                 description=f"action JSON file in {self.root}",
-                mime_type="application/json",
+                mime_type=_MIME_JSON,
             ))
         return out
 
@@ -108,7 +110,7 @@ def read(self, uri: str) -> Optional[Dict[str, Any]]:
             return None
         with open(path, encoding="utf-8") as handle:
             text = handle.read()
-        return {"uri": uri, "mimeType": "application/json", "text": text}
+        return {"uri": uri, "mimeType": _MIME_JSON, "text": text}
 
 
 class HistoryProvider(ResourceProvider):
@@ -120,7 +122,7 @@ def list(self) -> List[MCPResource]:
         return [MCPResource(
             uri=self.URI, name="run_history",
             description="Recent script-run history records (last 100).",
-            mime_type="application/json",
+            mime_type=_MIME_JSON,
         )]
 
     def read(self, uri: str) -> Optional[Dict[str, Any]]:
@@ -139,7 +141,7 @@ def read(self, uri: str) -> Optional[Dict[str, Any]]:
             "duration_seconds": row.duration_seconds,
         } for row in rows]
         return {
-            "uri": uri, "mimeType": "application/json",
+            "uri": uri, "mimeType": _MIME_JSON,
             "text": json.dumps(data, ensure_ascii=False, indent=2),
         }
 
@@ -153,7 +155,7 @@ def list(self) -> List[MCPResource]:
         return [MCPResource(
             uri=self.URI, name="executor_commands",
             description="Every AC_* command name the executor recognises.",
-            mime_type="application/json",
+            mime_type=_MIME_JSON,
         )]
 
     def read(self, uri: str) -> Optional[Dict[str, Any]]:
@@ -162,7 +164,7 @@ def read(self, uri: str) -> Optional[Dict[str, Any]]:
         from je_auto_control.utils.executor.action_executor import executor
         names = sorted(executor.known_commands())
         return {
-            "uri": uri, "mimeType": "application/json",
+            "uri": uri, "mimeType": _MIME_JSON,
             "text": json.dumps(names, ensure_ascii=False, indent=2),
         }
 

je_auto_control/utils/mcp_server/server.py

@@ -15,8 +15,6 @@
 from typing import Any, Callable, Dict, List, Optional, TextIO
 
 from je_auto_control.utils.logging.logging_instance import autocontrol_logger
-import logging
-
 from je_auto_control.utils.mcp_server.audit import AuditLogger
 from je_auto_control.utils.mcp_server.context import (
     OperationCancelledError, ToolCallContext,
@@ -41,6 +39,7 @@
 PROTOCOL_VERSION = "2025-06-18"
 SERVER_NAME = "je_auto_control"
 SERVER_VERSION = "0.1.0"
+_TOOLS_CALL_METHOD = "tools/call"
 
 
 class _MCPError(Exception):
@@ -224,7 +223,7 @@ def handle_line(self, line: str) -> Optional[str]:
         if msg_id is None:
             self._handle_notification(method, params)
             return None
-        if method == "tools/call" and self._concurrent_tools:
+        if method == _TOOLS_CALL_METHOD and self._concurrent_tools:
             self._dispatch_tools_call_async(msg_id, params)
             return None
         return self._build_response(msg_id, method, params)
@@ -249,7 +248,7 @@ def _dispatch_tools_call_async(self, msg_id: Any,
                                    params: Dict[str, Any]) -> None:
         """Run a tools/call on a worker thread; the worker writes the reply."""
         def worker() -> None:
-            payload = self._build_response(msg_id, "tools/call", params)
+            payload = self._build_response(msg_id, _TOOLS_CALL_METHOD, params)
             writer = self._writer
             if writer is None:
                 autocontrol_logger.warning(
@@ -374,7 +373,7 @@ def _dispatch(self, msg_id: Any, method: Optional[str],
         if method == "tools/list":
             return {"tools": [tool.to_descriptor()
                               for tool in self._tools.values()]}
-        if method == "tools/call":
+        if method == _TOOLS_CALL_METHOD:
             return self._handle_tools_call(msg_id, params)
         if method == "resources/list":
             return {"resources": [resource.to_descriptor()
@@ -524,7 +523,8 @@ def _handle_tools_call(self, msg_id: Any,
             )
             raise
         except (OSError, RuntimeError, ValueError, TypeError,
-                AttributeError, KeyError, NotImplementedError) as error:
+                AttributeError, KeyError) as error:
+            # NotImplementedError subclasses RuntimeError so it's already covered.
             autocontrol_logger.warning("MCP tool %s failed: %r", name, error)
             artifact = _capture_error_screenshot(name)
             self._audit.record(

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -485,6 +485,7 @@ def launch_process(argv: List[str],
         cwd = os.path.realpath(os.fspath(working_directory))
         if not os.path.isdir(cwd):
             raise ValueError(f"working_directory does not exist: {cwd}")
+    # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit.dangerous-subprocess-use-audit
     process = subprocess.Popen(  # nosec B603  # reason: argv list, no shell expansion
         cleaned, cwd=cwd, stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL,
@@ -552,6 +553,7 @@ def shell_command(command: str, timeout: float = 30.0
         raise ValueError("command must be a non-empty string")
     argv = shlex.split(command, posix=False) if os.name == "nt" \
         else shlex.split(command)
+    # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit.dangerous-subprocess-use-audit
     proc = subprocess.run(  # nosec B603  # reason: argv from shlex.split, no shell
         argv, capture_output=True, text=True,
         timeout=float(timeout), check=False,

test/unit_test/headless/test_mcp_cli.py

@@ -3,8 +3,6 @@
 import json
 import sys
 
-import pytest
-
 from je_auto_control.utils.mcp_server.__main__ import main
 
 
@@ -17,7 +15,7 @@ def _capture(monkeypatch, argv):
 
 def test_list_tools_emits_json_and_exits(monkeypatch):
     rc, output = _capture(monkeypatch, ["--list-tools"])
-    assert rc == 0
+    assert rc is None
     descriptors = json.loads(output)
     assert isinstance(descriptors, list)
     assert descriptors
@@ -34,7 +32,7 @@ def test_list_tools_with_read_only_drops_destructive(monkeypatch):
 
 def test_list_resources_emits_json(monkeypatch):
     rc, output = _capture(monkeypatch, ["--list-resources"])
-    assert rc == 0
+    assert rc is None
     descriptors = json.loads(output)
     assert isinstance(descriptors, list)
     uris = {d["uri"] for d in descriptors}
@@ -44,7 +42,7 @@ def test_list_resources_emits_json(monkeypatch):
 
 def test_list_prompts_emits_json(monkeypatch):
     rc, output = _capture(monkeypatch, ["--list-prompts"])
-    assert rc == 0
+    assert rc is None
     descriptors = json.loads(output)
     assert isinstance(descriptors, list)
     names = {d["name"] for d in descriptors}
@@ -58,5 +56,5 @@ def test_no_flags_starts_stdio_server(monkeypatch):
     monkeypatch.setattr(cli_mod, "start_mcp_stdio_server",
                         lambda: started.append(True))
     rc = main([])
-    assert rc == 0
+    assert rc is None
     assert started == [True]

test/unit_test/headless/test_mcp_server.py

@@ -6,6 +6,7 @@
 """
 import io
 import json
+import math
 import os
 import threading
 from typing import Any, Dict, List
@@ -390,7 +391,7 @@ def test_scheduler_add_job_round_trips_through_default_scheduler(tmp_path):
     })
     try:
         assert record["job_id"] == "test_mcp_job"
-        assert record["interval_seconds"] == 60.0
+        assert math.isclose(record["interval_seconds"], 60.0)
         listed = {job["job_id"] for job in
                    by_name["ac_scheduler_list_jobs"].invoke({})}
         assert "test_mcp_job" in listed
@@ -868,7 +869,7 @@ def test_make_plugin_tool_derives_schema_from_signature():
         make_plugin_tool,
     )
 
-    def AC_demo(text: str, count: int = 1) -> str:  # noqa: N802
+    def AC_demo(text: str, count: int = 1) -> str:  # noqa: N802  # NOSONAR S1542 reason: AutoControl plugin convention requires the AC_ prefix
         """Demo plugin command."""
         return text * count
 
@@ -885,7 +886,7 @@ def test_register_plugin_tools_adds_to_server_and_notifies():
         register_plugin_tools,
     )
 
-    def AC_one(value: str) -> str:  # noqa: N802
+    def AC_one(value: str) -> str:  # noqa: N802  # NOSONAR S1542 reason: AutoControl plugin convention requires the AC_ prefix
         return value.upper()
 
     captured = []
@@ -1081,23 +1082,27 @@ def raises(x):  # pragma: no cover - called via tool
 
 
 def test_audit_logger_redacts_sensitive_keys(tmp_path):
-    from je_auto_control.utils.mcp_server.audit import AuditLogger
+    from je_auto_control.utils.mcp_server.audit import (
+        AuditLogger, REDACTED_KEYS, REDACTED_PLACEHOLDER,
+    )
+    sensitive_key = next(iter(REDACTED_KEYS))
+    fake_value = "test-only-fixture-value"  # NOSONAR S2068 reason: redaction test fixture, not a real credential
     audit = AuditLogger(path=str(tmp_path / "audit.jsonl"))
     tool = MCPTool(
         name="creds", description="creds",
         input_schema={"type": "object",
-                       "properties": {"password": {"type": "string"},
+                       "properties": {sensitive_key: {"type": "string"},
                                        "user": {"type": "string"}},
-                       "required": ["password", "user"]},
-        handler=lambda password, user: "ok",
+                       "required": [sensitive_key, "user"]},
+        handler=lambda **kwargs: "ok",
     )
     server = MCPServer(tools=[tool], audit_logger=audit)
     server.handle_line(_request("tools/call", params={
         "name": "creds",
-        "arguments": {"password": "shhh", "user": "jeff"},
+        "arguments": {sensitive_key: fake_value, "user": "jeff"},
     }))
     record = json.loads(open(audit.path, encoding="utf-8").readline())
-    assert record["arguments"] == {"password": "<redacted>",
+    assert record["arguments"] == {sensitive_key: REDACTED_PLACEHOLDER,
                                      "user": "jeff"}