From c00ad8300c9637701f86ddb0bce91f0eab37afbf Mon Sep 17 00:00:00 2001 From: xTITUSMAXIMUSX Date: Wed, 1 Jul 2026 15:45:04 -0500 Subject: [PATCH] Validate operation value arity in propose_operations The generic write driver passed each op's comma-joined value straight through to VyManager without checking how many args it contained. A value carrying the wrong number of args (most commonly the subject re-included, e.g. "dum10,10.100.10.10/32" for a 1-arg address op) was only rejected at the device as an opaque HTTP 400 from /configure. - propose_operations now checks the comma-separated arg count against the op's expected post-subject arity and raises a clear error at propose time, naming the expected args and reminding that the subject is supplied via fields. - describe_feature_operations now exposes per-op value_args and value_arg_count (the args the value must supply, subject stripped) and a sharper note, so callers see the exact value shape up front. Bump version to 0.1.2 (also syncs __init__ __version__, which lagged). --- pyproject.toml | 2 +- src/vymcp/__init__.py | 2 +- src/vymcp/writes.py | 52 +++++++++++++++++++----- tests/test_generic_driver.py | 79 ++++++++++++++++++++++++++++++++---- 4 files changed, 113 insertions(+), 22 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ef3567d..3ab1e6a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "vymcp" -version = "0.1.1" +version = "0.1.2" description = "MCP server for VyOS routers, powered by VyManager" readme = "README.md" requires-python = ">=3.10" diff --git a/src/vymcp/__init__.py b/src/vymcp/__init__.py index a707d06..ca9a269 100644 --- a/src/vymcp/__init__.py +++ b/src/vymcp/__init__.py @@ -1,3 +1,3 @@ """VyMCP — Model Context Protocol server for VyOS routers, powered by VyManager.""" -__version__ = "0.1.0" +__version__ = "0.1.2" diff --git a/src/vymcp/writes.py b/src/vymcp/writes.py index bf1d2c6..e3c7b76 100644 --- a/src/vymcp/writes.py +++ b/src/vymcp/writes.py @@ -177,17 +177,33 @@ async def describe_feature_operations(feature: str) -> dict[str, Any]: operations = await discovery.get_feature_operations(feature) fields = await discovery.get_top_level_fields(feature) subject_field = await discovery.get_subject_field(feature) + + # Spell out, per op, exactly what the *value* must contain: the args AFTER + # the subject (which is passed separately via fields). This is the contract + # callers most often get wrong — mixing the subject back into the value. + subject_args = 1 if subject_field else 0 + described = [ + { + **op, + "value_args": op.get("args", [])[subject_args:], + "value_arg_count": max(op.get("arg_count", 0) - subject_args, 0), + } + for op in operations + ] + note = ( + f"Pass the subject via fields['{subject_field}']. Each op's 'value' supplies " + "ONLY its 'value_args', comma-joined in order — never include the subject in " + "the value. Ops with value_arg_count 0 take no value." + if subject_field + else "Each op's 'value' supplies its 'value_args', comma-joined in order. " + "Ops with value_arg_count 0 take no value." + ) return { "feature": feature, "subject_field": subject_field, "top_level_fields": fields, - "operations": operations, - "note": ( - f"This feature requires the '{subject_field}' field (the subject, e.g. the " - f"interface name); each op's value supplies the args after it." - if subject_field - else None - ), + "operations": described, + "note": note, } @mcp.tool() @@ -233,8 +249,23 @@ async def propose_operations( f"describe_feature_operations('{feature}') for the vocabulary." ) value = op.get("value") - if vocab[name]["arg_count"] - subject_args >= 1 and not value: + expected = vocab[name]["arg_count"] - subject_args + if expected >= 1 and not value: raise ValueError(f"Operation '{name}' requires a value.") + if expected >= 1 and value: + parts = value.split(",") + if len(parts) != expected: + wanted = ", ".join(vocab[name].get("args", [])[subject_args:]) + hint = ( + f" The subject '{subject_field}' is supplied via fields — do NOT " + "include it in the value." + if subject_field + else "" + ) + raise ValueError( + f"Operation '{name}' expects {expected} comma-separated value(s) " + f"({wanted}), but got {len(parts)}: {value!r}.{hint}" + ) entry: dict[str, Any] = {"op": name} if value is not None: entry["value"] = value @@ -242,9 +273,8 @@ async def propose_operations( body: dict[str, Any] = dict(fields or {}) body["operations"] = normalized - summary = ( - f"{len(normalized)} operation(s) on '{feature}' (instance {instance_id})" - + (f"; fields {fields}" if fields else "") + summary = f"{len(normalized)} operation(s) on '{feature}' (instance {instance_id})" + ( + f"; fields {fields}" if fields else "" ) plan = plan_store.create( owner=current_owner(), diff --git a/tests/test_generic_driver.py b/tests/test_generic_driver.py index 604a5c2..d27cf75 100644 --- a/tests/test_generic_driver.py +++ b/tests/test_generic_driver.py @@ -5,8 +5,12 @@ "feature": "nat", "operations": [ {"op": "set_source_rule", "args": ["rule"], "arg_count": 1, "description": "Rule."}, - {"op": "set_source_rule_translation_address", "args": ["rule", "addr"], - "arg_count": 2, "description": "Translation address."}, + { + "op": "set_source_rule_translation_address", + "args": ["rule", "addr"], + "arg_count": 2, + "description": "Translation address.", + }, {"op": "delete_all", "args": [], "arg_count": 0, "description": "Delete all."}, ], } @@ -51,6 +55,7 @@ def handler(request: httpx.Request) -> httpx.Response: if r is not None: return r return httpx.Response(404, json={}) + return handler @@ -58,7 +63,9 @@ async def test_describe_merges_ops_and_fields(write_tools, install_client): install_client(_discovery_handler()) out = await write_tools["describe_feature_operations"]("nat") assert {o["op"] for o in out["operations"]} == { - "set_source_rule", "set_source_rule_translation_address", "delete_all" + "set_source_rule", + "set_source_rule_translation_address", + "delete_all", } assert out["top_level_fields"]["nat_type"]["required"] is True @@ -66,12 +73,14 @@ async def test_describe_merges_ops_and_fields(write_tools, install_client): async def test_propose_builds_body_with_fields(write_tools, install_client): install_client(_discovery_handler()) plan = await write_tools["propose_operations"]( - "nat", "i1", + "nat", + "i1", [{"op": "set_source_rule", "value": "100"}], fields={"nat_type": "source"}, ) assert plan["applied"] is False from vymcp.changes import plan_store + body = plan_store.get(plan["plan_id"], "local").body assert body == {"nat_type": "source", "operations": [{"op": "set_source_rule", "value": "100"}]} @@ -92,6 +101,7 @@ async def test_propose_zero_arg_op_needs_no_value(write_tools, install_client): install_client(_discovery_handler()) plan = await write_tools["propose_operations"]("nat", "i1", [{"op": "delete_all"}]) from vymcp.changes import plan_store + assert plan_store.get(plan["plan_id"], "local").body["operations"] == [{"op": "delete_all"}] @@ -106,8 +116,12 @@ async def test_propose_rejects_empty(write_tools, install_client): "subject_field": "interface_name", "operations": [ {"op": "set_interface_disable", "args": ["interface"], "arg_count": 1, "description": "x"}, - {"op": "set_interface_address", "args": ["interface", "addr"], - "arg_count": 2, "description": "x"}, + { + "op": "set_interface_address", + "args": ["interface", "addr"], + "arg_count": 2, + "description": "x", + }, ], } @@ -123,9 +137,7 @@ def _bonding_handler(request: httpx.Request) -> httpx.Response: async def test_subject_feature_requires_subject(write_tools, install_client): install_client(_bonding_handler) with pytest.raises(ValueError, match="interface_name"): - await write_tools["propose_operations"]( - "bonding", "i1", [{"op": "set_interface_disable"}] - ) + await write_tools["propose_operations"]("bonding", "i1", [{"op": "set_interface_disable"}]) async def test_subject_feature_zero_value_op_ok(write_tools, install_client): @@ -135,6 +147,7 @@ async def test_subject_feature_zero_value_op_ok(write_tools, install_client): "bonding", "i1", [{"op": "set_interface_disable"}], fields={"interface_name": "bond0"} ) from vymcp.changes import plan_store + body = plan_store.get(plan["plan_id"], "local").body assert body == {"interface_name": "bond0", "operations": [{"op": "set_interface_disable"}]} @@ -148,6 +161,54 @@ async def test_subject_feature_value_op_needs_value(write_tools, install_client) ) +async def test_subject_included_in_value_is_rejected(write_tools, install_client): + install_client(_bonding_handler) + # The classic mistake: subject re-jammed into the value. set_interface_address + # wants 1 value arg (the address); passing "bond0,10.0.0.1/32" is 2 -> reject + # here instead of letting the device answer an opaque 400. + with pytest.raises(ValueError, match="expects 1 comma-separated value"): + await write_tools["propose_operations"]( + "bonding", + "i1", + [{"op": "set_interface_address", "value": "bond0,10.0.0.1/32"}], + fields={"interface_name": "bond0"}, + ) + + +async def test_correct_single_value_accepted(write_tools, install_client): + install_client(_bonding_handler) + plan = await write_tools["propose_operations"]( + "bonding", + "i1", + [{"op": "set_interface_address", "value": "10.0.0.1/32"}], + fields={"interface_name": "bond0"}, + ) + from vymcp.changes import plan_store + + body = plan_store.get(plan["plan_id"], "local").body + assert body["operations"] == [{"op": "set_interface_address", "value": "10.0.0.1/32"}] + + +async def test_too_many_args_rejected_non_subject(write_tools, install_client): + install_client(_discovery_handler()) + # set_source_rule takes 1 arg; passing 2 comma parts is rejected. + with pytest.raises(ValueError, match="expects 1 comma-separated value"): + await write_tools["propose_operations"]( + "nat", "i1", [{"op": "set_source_rule", "value": "100,extra"}] + ) + + +async def test_describe_exposes_value_args(write_tools, install_client): + install_client(_bonding_handler) + out = await write_tools["describe_feature_operations"]("bonding") + ops = {o["op"]: o for o in out["operations"]} + # The subject is stripped from what the value must supply. + assert ops["set_interface_address"]["value_args"] == ["addr"] + assert ops["set_interface_address"]["value_arg_count"] == 1 + assert ops["set_interface_disable"]["value_arg_count"] == 0 + assert "never include the subject" in out["note"] + + async def test_generic_plan_applies(write_tools, install_client): def extra(request): p = request.url.path