symfonic.capabilities.extensions.trust

trust

Boundary validation for untrusted extension payloads (AS-INT-1, AS-INT-2).

Everything an MCP server answers and everything a plugin returns arrives here before it becomes a contribution. The module is small on purpose: it is the one place that decides what "a name", "a description", and "a schema" mean, so a reviewer checking that the caps are sane reads one file rather than four call sites that each grew their own.

Three rules run through it.

Identifiers are rejected; content is truncated. A malformed name is refused (:class:~.errors.UntrustedPayloadError) because names index dictionaries, appear in rendered manifests, and route calls — a name outside the charset can forge a delimiter. A too-long description is truncated with a diagnostic, because it is content: dropping the tool over its docstring would let a verbose server delete itself from a catalogue.

A schema is refused, never repaired. It drives argument binding, so a schema that is too deep, too large, or not a mapping is a schema the framework declines to bind against. Repairing one would mean the model is told about a tool whose real signature nobody checked.

Bounds exist for every unbounded field. AS-ING-6's ceiling argument applies to integration payloads too: a server that answers with a 300 MB result string must cost a truncation, not a process.

TrustLimits dataclass

TrustLimits(max_identifier_length: int = 128, max_description_length: int = 8192, max_schema_bytes: int = 65536, max_schema_depth: int = 16, max_tools_per_server: int = 512, max_result_length: int = 262144, max_result_parts: int = 64)

The ceilings every untrusted payload is read under.

Defaults are generous enough that a well-behaved server never notices them and small enough that a hostile one cannot exhaust the process. They are a value rather than module constants so a deployment facing an unusually chatty server can raise one without patching the framework.

sanitize_text

sanitize_text(value: Any, *, limit: int) -> tuple[str, bool]

Return (text, truncated) for an untrusted content string.

A non-string reads as absent rather than raising: a server that answers description: null has given us a tool without a description, which is poor but not hostile. Control characters are stripped, then the result is capped at limit and marked when it was shortened.

Source code in src/symfonic/capabilities/extensions/trust.py

def sanitize_text(value: Any, *, limit: int) -> tuple[str, bool]:
    """Return ``(text, truncated)`` for an untrusted content string.

    A non-string reads as absent rather than raising: a server that answers
    ``description: null`` has given us a tool without a description, which is
    poor but not hostile. Control characters are stripped, then the result is
    capped at ``limit`` and marked when it was shortened.
    """
    if not isinstance(value, str):
        return "", False
    cleaned = _CONTROL_CHARS.sub("", value)
    if len(cleaned) <= limit:
        return cleaned, False
    keep = max(limit - len(TRUNCATION_MARKER), 0)
    return cleaned[:keep] + TRUNCATION_MARKER, True

validate_identifier

validate_identifier(value: Any, *, field: str, origin: str, limits: TrustLimits, allow_dot: bool = True) -> str

Return value as a usable identifier, or refuse it.

Parameters:

Name	Type	Description	Default
value	Any	The raw payload field, of unknown type.	required
field	str	The field's name, for the error message.	required
origin	str	Who supplied it (server name, plugin name).	required
limits	TrustLimits	The ceilings to read it under.	required
allow_dot	bool	False for a name that will be joined to another with a .. A dotted prefix makes the joined name ambiguous, which is a routing bug wearing a naming bug's clothes.	True

Raises:

Type	Description
UntrustedPayloadError	If it is not a string, is empty, exceeds the length cap, or leaves the identifier charset.

Source code in src/symfonic/capabilities/extensions/trust.py

def validate_identifier(
    value: Any,
    *,
    field: str,
    origin: str,
    limits: TrustLimits,
    allow_dot: bool = True,
) -> str:
    """Return ``value`` as a usable identifier, or refuse it.

    Args:
        value: The raw payload field, of unknown type.
        field: The field's name, for the error message.
        origin: Who supplied it (server name, plugin name).
        limits: The ceilings to read it under.
        allow_dot: ``False`` for a name that will be joined to another with a
            ``.``. A dotted prefix makes the joined name ambiguous, which is a
            routing bug wearing a naming bug's clothes.

    Raises:
        UntrustedPayloadError: If it is not a string, is empty, exceeds the
            length cap, or leaves the identifier charset.
    """
    if not isinstance(value, str):
        raise UntrustedPayloadError(
            f"{origin}: {field} must be a string, got {type(value).__name__}."
        )
    if not value:
        raise UntrustedPayloadError(f"{origin}: {field} is empty.")
    if len(value) > limits.max_identifier_length:
        raise UntrustedPayloadError(
            f"{origin}: {field} is {len(value)} characters, over the "
            f"{limits.max_identifier_length}-character limit."
        )
    pattern = _ID_CHARSET if allow_dot else _NAMESPACE_CHARSET
    if not pattern.match(value):
        charset = "[A-Za-z0-9_.-]" if allow_dot else "[A-Za-z0-9_-]"
        raise UntrustedPayloadError(
            f"{origin}: {field} {value!r} is outside the permitted charset "
            f"{charset}; these names are rendered into manifests and used as "
            "routing keys, where a separator forges a boundary."
        )
    return value

validate_schema

validate_schema(value: Any, *, origin: str, field: str, limits: TrustLimits) -> dict[str, Any]

Return the payload's input schema as a plain dict, or refuse it.

An absent schema is legal and reads as {} — an MCP tool taking no arguments is ordinary. Anything present must be a JSON-shaped mapping within the depth and size bounds.

Raises:

Type	Description
UntrustedPayloadError	If the schema is not a mapping, is not JSON-serialisable, or exceeds a bound.

Source code in src/symfonic/capabilities/extensions/trust.py

def validate_schema(
    value: Any,
    *,
    origin: str,
    field: str,
    limits: TrustLimits,
) -> dict[str, Any]:
    """Return the payload's input schema as a plain dict, or refuse it.

    An absent schema is legal and reads as ``{}`` — an MCP tool taking no
    arguments is ordinary. Anything present must be a JSON-shaped mapping
    within the depth and size bounds.

    Raises:
        UntrustedPayloadError: If the schema is not a mapping, is not
            JSON-serialisable, or exceeds a bound.
    """
    if value is None:
        return {}
    if not isinstance(value, Mapping):
        raise UntrustedPayloadError(
            f"{origin}: {field} must be a JSON object, got {type(value).__name__}."
        )
    depth = _depth(value, budget=limits.max_schema_depth)
    if depth > limits.max_schema_depth:
        raise UntrustedPayloadError(
            f"{origin}: {field} nests deeper than the {limits.max_schema_depth}-level "
            "limit; the framework declines to bind arguments against a schema it "
            "cannot finish reading."
        )
    # Encoded incrementally and abandoned at the cap, rather than serialised
    # whole and measured afterwards: a 300 MB schema must cost a refusal, not
    # the 300 MB string that proves it was too big.
    encoded_length = 0
    try:
        for chunk in json.JSONEncoder(default=None).iterencode(value):
            encoded_length += len(chunk)
            if encoded_length > limits.max_schema_bytes:
                break
    except (TypeError, ValueError) as err:
        raise UntrustedPayloadError(
            f"{origin}: {field} is not JSON-serialisable ({err})."
        ) from err
    if encoded_length > limits.max_schema_bytes:
        raise UntrustedPayloadError(
            f"{origin}: {field} serialises to at least {encoded_length} bytes, "
            f"over the {limits.max_schema_bytes}-byte limit."
        )
    return dict(value)