mcp: validate tool inputs

kcidev/libs/dashboard.py

@@ -375,12 +375,19 @@ def dashboard_fetch_issue_list(origin, days, use_json):
     return dashboard_api_fetch("issue/", params, use_json)
 
 
+def _require_issue_id(issue_id):
+    if not issue_id or not issue_id.strip():
+        raise click.ClickException("Issue id is required")
+
+
 def dashboard_fetch_issue(issue_id, use_json):
+    _require_issue_id(issue_id)
     logging.info(f"Fetching issue details for issue ID: {issue_id}")
     return dashboard_api_fetch(f"issue/{issue_id}", {}, use_json)
 
 
 def dashboard_fetch_issue_builds(origin, issue_id, use_json, error_verbose=True):
+    _require_issue_id(issue_id)
     logging.info(f"Fetching builds for issue ID: {issue_id}")
     params = {"filter_origin": origin} if origin else {}
     return dashboard_api_fetch(
@@ -389,6 +396,7 @@ def dashboard_fetch_issue_builds(origin, issue_id, use_json, error_verbose=True)
 
 
 def dashboard_fetch_issue_tests(origin, issue_id, use_json, error_verbose=True):
+    _require_issue_id(issue_id)
     logging.info(f"Fetching tests for issue ID: {issue_id}")
     params = {"filter_origin": origin} if origin else {}
     return dashboard_api_fetch(

kcidev/mcp/tools_dashboard.py

@@ -7,6 +7,7 @@
 from kcidev.api import KciDevError, KernelCIClient
 from kcidev.libs.filters import StatusFilter
 from kcidev.mcp.errors import tool_errors
+from kcidev.mcp.validation import check_page_bounds, checked_status
 
 _active_client = ContextVar("dashboard_tool_client", default=None)
 
@@ -16,10 +17,11 @@ def _current_client():
 
 
 def _page(data, key, status, limit, offset, fields=None):
+    check_page_bounds(limit, offset)
     items = data[key] if isinstance(data, dict) else data
     total = len(items)
     if status:
-        status_filter = StatusFilter(status)
+        status_filter = StatusFilter(checked_status(status))
         items = [item for item in items if status_filter.matches(item)]
     page = items[offset : offset + limit]
     if fields:
@@ -142,7 +144,7 @@ def list_builds(
     """List kernel builds for one commit of a tree.
 
     Optional filters: arch (e.g. 'arm64'), tree name, ISO date range, and
-    status ('pass', 'fail' or 'inconclusive'). Results are paginated with
+    status ('pass', 'fail', 'inconclusive' or 'all'). Results are paginated with
     limit/offset; the response carries 'total' (before status filtering)
     and 'matched' counts so you know whether to fetch further pages;
     fields projects each entry to only those keys.
@@ -173,7 +175,7 @@ def list_boots(
     """List boot test results for one commit of a tree.
 
     Optional filters: arch, tree name, ISO date range, boot origin, and
-    status ('pass', 'fail' or 'inconclusive'). Results are paginated with
+    status ('pass', 'fail', 'inconclusive' or 'all'). Results are paginated with
     limit/offset; the response carries 'total' (before status filtering)
     and 'matched' counts so you know whether to fetch further pages;
     fields projects each entry to only those keys.
@@ -203,7 +205,7 @@ def list_tests(
     """List test results for one commit of a tree.
 
     Optional filters: arch, tree name, ISO date range, and status ('pass',
-    'fail' or 'inconclusive'). A full commit can carry tens of thousands
+    'fail', 'inconclusive' or 'all'). A full commit can carry tens of thousands
     of tests, so filter by status and paginate with limit/offset; the
     response carries 'total' (before status filtering) and 'matched'
     counts so you know whether to fetch further pages; fields projects
@@ -284,7 +286,7 @@ def get_issue_builds(
 ):
     """List builds affected by a known issue.
 
-    Optional status filter ('pass', 'fail' or 'inconclusive') and
+    Optional status filter ('pass', 'fail', 'inconclusive' or 'all') and
     limit/offset pagination; the response carries 'total' and 'matched'
     counts; fields projects each entry to only those keys.
     """
@@ -303,7 +305,7 @@ def get_issue_tests(
 ):
     """List tests affected by a known issue.
 
-    Optional status filter ('pass', 'fail' or 'inconclusive') and
+    Optional status filter ('pass', 'fail', 'inconclusive' or 'all') and
     limit/offset pagination; the response carries 'total' and 'matched'
     counts; fields projects each entry to only those keys.
     """

kcidev/mcp/tools_maestro.py

@@ -2,6 +2,7 @@
 # -*- coding: utf-8 -*-
 
 from kcidev.mcp.errors import tool_errors
+from kcidev.mcp.validation import check_page_bounds, checked_filters
 
 
 def register_tools(server, client, api_url, pipeline_url, token):
@@ -44,7 +45,10 @@ def list_nodes(
             paginate within the window; full nodes are large, so use
             fields to project each node to only those keys.
             """
-            nodes = client.get_nodes(limit=limit, offset=offset, filters=filters or [])
+            check_page_bounds(limit, offset)
+            nodes = client.get_nodes(
+                limit=limit, offset=offset, filters=checked_filters(filters)
+            )
             if fields:
                 return [{k: n[k] for k in fields if k in n} for n in nodes]
             return nodes

kcidev/mcp/validation.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from kcidev.api import KciDevError
+
+STATUS_CHOICES = ("all", "pass", "fail", "inconclusive")
+
+
+def checked_status(status):
+    normalised = status.strip().lower()
+    if normalised not in STATUS_CHOICES:
+        raise KciDevError(
+            f"Unknown status {status!r}: expected one of {', '.join(STATUS_CHOICES)}"
+        )
+    return normalised
+
+
+def check_page_bounds(limit, offset):
+    if limit < 0:
+        raise KciDevError(f"Invalid limit {limit}: must be zero or greater")
+    if offset < 0:
+        raise KciDevError(f"Invalid offset {offset}: must be zero or greater")
+
+
+def checked_filters(filters):
+    for entry in filters or []:
+        if "=" not in entry:
+            raise KciDevError(
+                f"Invalid filter {entry!r}: expected 'field=value', "
+                "for example 'state=done'"
+            )
+    return list(filters or [])

tests/test_dashboard.py

@@ -1,5 +1,7 @@
 from unittest.mock import Mock
 
+import click
+import pytest
 from click.testing import CliRunner
 
 from kcidev.libs import dashboard
@@ -170,3 +172,50 @@ def test_kernelci_clients_keep_independent_dashboard_endpoints(monkeypatch):
     assert urls[1].startswith(dashboard.DASHBOARD_API_DEFAULT)
     assert urls[2].startswith("https://three.example/api/")
     assert urls[3].startswith("https://one.example/api/")
+
+
+def _issue_collection_get(monkeypatch):
+    response = Mock(status_code=200)
+    response.json.return_value = {"issues": [{"id": "maestro:one"}]}
+    get = Mock(return_value=response)
+    monkeypatch.setattr(dashboard.kcidev_session, "get", get)
+    return get
+
+
+def test_dashboard_fetch_issue_rejects_empty_id(monkeypatch):
+    get = _issue_collection_get(monkeypatch)
+
+    with pytest.raises(click.ClickException):
+        dashboard.dashboard_fetch_issue("", False)
+
+    get.assert_not_called()
+
+
+def test_dashboard_fetch_issue_builds_rejects_empty_id(monkeypatch):
+    get = _issue_collection_get(monkeypatch)
+
+    with pytest.raises(click.ClickException):
+        dashboard.dashboard_fetch_issue_builds(None, "", False)
+
+    get.assert_not_called()
+
+
+def test_dashboard_fetch_issue_tests_rejects_empty_id(monkeypatch):
+    get = _issue_collection_get(monkeypatch)
+
+    with pytest.raises(click.ClickException):
+        dashboard.dashboard_fetch_issue_tests(None, "", False)
+
+    get.assert_not_called()
+
+
+def test_cli_issue_command_rejects_empty_id(monkeypatch):
+    from kcidev.main import get_cli
+
+    get = _issue_collection_get(monkeypatch)
+
+    runner = CliRunner()
+    result = runner.invoke(get_cli(), ["results", "issue", "--id", ""])
+
+    assert result.exit_code != 0
+    get.assert_not_called()

tests/test_mcp_server.py

@@ -207,3 +207,35 @@ def test_list_nodes_projects_fields(monkeypatch):
     assert result.isError is False
     assert '"commit_message"' not in result.content[0].text
     assert '"n1"' in result.content[0].text
+
+
+def _no_http(monkeypatch):
+    from kcidev.libs import maestro_common
+
+    get = Mock()
+    monkeypatch.setattr(maestro_common.kcidev_session, "get", get)
+    return get
+
+
+def test_list_nodes_rejects_negative_limit(monkeypatch):
+    get = _no_http(monkeypatch)
+    result = _call_tool(create_server(CFG, "test"), "list_nodes", {"limit": -1})
+    assert result.isError is True
+    get.assert_not_called()
+
+
+def test_list_nodes_rejects_negative_offset(monkeypatch):
+    get = _no_http(monkeypatch)
+    result = _call_tool(create_server(CFG, "test"), "list_nodes", {"offset": -1})
+    assert result.isError is True
+    get.assert_not_called()
+
+
+def test_list_nodes_rejects_filter_without_equals(monkeypatch):
+    get = _no_http(monkeypatch)
+    result = _call_tool(
+        create_server(CFG, "test"), "list_nodes", {"filters": ["state done"]}
+    )
+    assert result.isError is True
+    assert "state done" in result.content[0].text
+    get.assert_not_called()

tests/test_mcp_tools_dashboard.py

@@ -214,3 +214,49 @@ def test_get_summary_detail_returns_full_payload(monkeypatch):
         detail=True,
     )
     assert result == SUMMARY_PAYLOAD
+
+
+def _tree_args(**extra):
+    args = {
+        "giturl": "https://git.example.org/linux.git",
+        "branch": "master",
+        "commit": "deadbeef",
+    }
+    args.update(extra)
+    return args
+
+
+def test_list_tests_accepts_uppercase_status(monkeypatch):
+    _mock_get(
+        monkeypatch,
+        {"tests": [{"id": "p1", "status": "PASS"}, {"id": "f1", "status": "FAIL"}]},
+    )
+    result = tools_dashboard.list_tests(**_tree_args(status="FAIL"))
+    assert result["matched"] == 1
+    assert result["tests"] == [{"id": "f1", "status": "FAIL"}]
+
+
+def test_list_tests_rejects_unknown_status(monkeypatch):
+    _mock_get(monkeypatch, {"tests": [{"id": "f1", "status": "FAIL"}]})
+    with pytest.raises(ToolExecutionError) as excinfo:
+        tools_dashboard.list_tests(**_tree_args(status="borked"))
+    assert "borked" in str(excinfo.value)
+
+
+def test_list_tests_rejects_negative_limit(monkeypatch):
+    _mock_get(monkeypatch, {"tests": [{"id": str(i)} for i in range(5)]})
+    with pytest.raises(ToolExecutionError):
+        tools_dashboard.list_tests(**_tree_args(limit=-1))
+
+
+def test_list_tests_rejects_negative_offset(monkeypatch):
+    _mock_get(monkeypatch, {"tests": [{"id": str(i)} for i in range(5)]})
+    with pytest.raises(ToolExecutionError):
+        tools_dashboard.list_tests(**_tree_args(offset=-1))
+
+
+def test_get_issue_rejects_empty_id(monkeypatch):
+    get = _mock_get(monkeypatch, {"issues": [{"id": "maestro:one"}]})
+    with pytest.raises(ToolExecutionError):
+        tools_dashboard.get_issue("")
+    get.assert_not_called()