Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions app/mcp_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,14 +80,16 @@ def optional_str_arg(field: str, default: str = "") -> str:
raise ValueError(f"{field} must be a string")
return value

def optional_clean_str_arg(field: str) -> str | None:
def optional_clean_str_arg(field: str, *, reject_padding: bool = False) -> str | None:
value = args.get(field)
if value is None:
return None
if not isinstance(value, str):
raise ValueError(f"{field} must be a string")
if contains_control_character(value):
raise ValueError(f"{field} must not contain control characters")
if reject_padding and value.strip() and value != value.strip():
raise ValueError(f"{field} must not contain leading or trailing whitespace")
clean = value.strip()
return clean or None

Expand Down Expand Up @@ -153,12 +155,12 @@ def selected_bounty(internal_id_field: str) -> Bounty | None:

with session_scope(database_url) as session:
if name == "list_bounties":
status = optional_clean_str_arg("status") or "open"
status = optional_clean_str_arg("status", reject_padding=True) or "open"
normalized_status = status.lower()
if normalized_status not in {"open", "paid", "closed"}:
raise ValueError("status must be one of: open, paid, closed")
query = select(Bounty).where(Bounty.status == normalized_status)
query_text = optional_clean_str_arg("q")
query_text = optional_clean_str_arg("q", reject_padding=True)
if query_text:
escaped_query = (
query_text.lower().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
Expand All @@ -173,9 +175,9 @@ def selected_bounty(internal_id_field: str) -> Bounty | None:
if issue_number is not None:
text_filter = or_(text_filter, Bounty.issue_number == issue_number)
query = query.where(text_filter)
sort = normalize_bounty_sort(optional_clean_str_arg("sort"))
sort = normalize_bounty_sort(optional_clean_str_arg("sort", reject_padding=True))
availability = normalize_bounty_availability_filter(
optional_clean_str_arg("availability")
optional_clean_str_arg("availability", reject_padding=True)
)
limit = list_limit_arg()
if sort == "newest" and availability == "all":
Expand Down
6 changes: 5 additions & 1 deletion tests/test_api_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -530,7 +530,7 @@ def test_mcp_list_bounties_filters_status_query_and_limit(sqlite_url: str) -> No
"method": "tools/call",
"params": {
"name": "list_bounties",
"arguments": {"status": " Paid ", "q": "proof", "limit": 1},
"arguments": {"status": "Paid", "q": "proof", "limit": 1},
},
},
).json()
Expand Down Expand Up @@ -697,6 +697,10 @@ def test_mcp_list_bounties_filters_effective_availability(sqlite_url: str) -> No
({"limit": 101}, 35),
({"sort": "invalid"}, 36),
({"availability": "maybe"}, 37),
({"status": " open "}, 38),
({"q": " proof "}, 39),
({"sort": " reward "}, 40),
({"availability": " effectively_open "}, 41),
],
)
def test_mcp_list_bounties_rejects_invalid_filters(
Expand Down
10 changes: 10 additions & 0 deletions tests/test_mcp_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,16 @@ def test_call_mcp_tool_rejects_c1_status_before_normalizing(sqlite_url: str) ->
call_mcp_tool(sqlite_url, "list_bounties", {"status": "\u0085open"})


@pytest.mark.parametrize("field", ["status", "q", "sort", "availability"])
def test_call_mcp_tool_rejects_padded_list_bounty_filters(sqlite_url: str, field: str) -> None:
create_schema(sqlite_url)
with session_scope(sqlite_url) as session:
ensure_genesis(session)

with pytest.raises(ValueError, match=f"{field} must not contain leading or trailing"):
call_mcp_tool(sqlite_url, "list_bounties", {field: " open "})


def test_call_mcp_tool_rejects_c1_nonce_before_integer_parsing(sqlite_url: str) -> None:
create_schema(sqlite_url)
with session_scope(sqlite_url) as session:
Expand Down
Loading