Harden stamp/verify/hash_report so unknown and content-shaped args fail instead of being ignored; publish additionalProperties=false and cover with schema/runtime tests. Also keep streamable-http host allowlist.
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
"""Hash-only contract: reject content/bytes-style tool arguments."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Mapping
|
|
|
|
from pydantic import ConfigDict, Field, ValidationError, create_model
|
|
|
|
from mcp.server.mcpserver.utilities.func_metadata import ArgModelBase
|
|
|
|
FORBIDDEN_CONTENT_KEYS = frozenset(
|
|
{
|
|
"content",
|
|
"bytes",
|
|
"file",
|
|
"data",
|
|
"path",
|
|
"body",
|
|
"payload",
|
|
"blob",
|
|
"archive",
|
|
"filename",
|
|
"filepath",
|
|
"file_path",
|
|
"raw",
|
|
"plaintext",
|
|
"document",
|
|
}
|
|
)
|
|
|
|
_REJECT = "content/bytes paths are rejected; send sha256 hex only"
|
|
|
|
|
|
class StrictArgModel(ArgModelBase):
|
|
"""Tool args: unknown keys (including content/bytes) are errors."""
|
|
|
|
model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
|
|
|
|
|
|
def reject_content_keys(arguments: Mapping[str, Any] | None) -> None:
|
|
"""Raise ValidationError if arguments include any content/bytes-style key."""
|
|
if not arguments:
|
|
return
|
|
hit = sorted({k for k in arguments if k.lower() in FORBIDDEN_CONTENT_KEYS})
|
|
if not hit:
|
|
return
|
|
raise ValidationError.from_exception_data(
|
|
"HashOnlyArguments",
|
|
[
|
|
{
|
|
"type": "value_error",
|
|
"loc": (key,),
|
|
"input": arguments.get(key),
|
|
"ctx": {"error": _REJECT},
|
|
}
|
|
for key in hit
|
|
],
|
|
)
|
|
|
|
|
|
def strict_arg_model(existing: type[ArgModelBase]) -> type[ArgModelBase]:
|
|
"""Rebuild a tool arg model with extra='forbid'."""
|
|
fields: dict[str, Any] = {}
|
|
for name, finfo in existing.model_fields.items():
|
|
annotation = finfo.annotation
|
|
kwargs: dict[str, Any] = {}
|
|
if finfo.alias:
|
|
kwargs["alias"] = finfo.alias
|
|
if finfo.is_required():
|
|
fields[name] = (annotation, Field(**kwargs))
|
|
else:
|
|
fields[name] = (annotation, Field(default=finfo.default, **kwargs))
|
|
return create_model(f"{existing.__name__}Strict", __base__=StrictArgModel, **fields)
|