routing

Header-driven routing: RoutingContext, sanitize_route and resolve_routing.

Request routing derived from the MCP client’s routing headers.

A client may advertise which model, agent, client and session a request belongs to through headers. This module turns those headers into a RoutingContext – a value that is only the result of parsing and whitelisting, never a filesystem path: every route string passes through sanitize_route() before it can be used as a directory name, and the header snapshot kept for auditing is stripped of every credential-bearing header. Nothing here performs I/O.

Security notes:

  • Header values are attacker-controlled. route is restricted to [A-Za-z0-9._-]{1,64} and values made only of dots (., ..) are rejected, so a route can never traverse out of its directory.

  • recorded_headers never contains authorization, cookie, set-cookie, proxy-authorization or any *-token / *-secret / *-password / *-key / *-credential header, and is capped so a hostile client cannot blow up the audit record.

novelai_image_mcp.routing.MAX_RECORDED_HEADERS = 64

Largest number of headers kept in recorded_headers.

novelai_image_mcp.routing.MAX_RECORDED_VALUE_LENGTH = 512

Longest header value kept in recorded_headers.

novelai_image_mcp.routing.MAX_ROUTE_LENGTH = 64

Longest accepted route string; longer values are rejected outright.

novelai_image_mcp.routing.ROUTE_STDIO = 'stdio'

Route used for the local stdio transport, which has no HTTP request.

novelai_image_mcp.routing.ROUTE_UNKNOWN = 'unknown'

Route used when no usable routing information is available.

class novelai_image_mcp.routing.RoutingContext(enabled, route, model=None, agent=None, client=None, session=None, transport='http', recorded_headers=<factory>)[source]

Bases: object

Routing decision for a single request.

enabled mirrors the insiders header: when it is False no routing was attempted, so every extracted field is None and route is ROUTE_UNKNOWN. transport always reflects the real transport, and recorded_headers is the sanitised header snapshot used for audit.

Parameters:
  • enabled (bool)

  • route (str)

  • model (str | None)

  • agent (str | None)

  • client (str | None)

  • session (str | None)

  • transport (Literal['http', 'stdio'])

  • recorded_headers (Mapping[str, str])

enabled: bool
route: str
model: str | None = None
agent: str | None = None
client: str | None = None
session: str | None = None
transport: Literal['http', 'stdio'] = 'http'
recorded_headers: Mapping[str, str]
classmethod disabled(*, transport='http')[source]

Return the context used when routing is switched off.

transport still carries the real transport so audit output stays truthful; the caller supplies it rather than relying on the default.

Parameters:

transport (Literal['http', 'stdio'])

Return type:

RoutingContext

novelai_image_mcp.routing.resolve_routing(headers, *, enabled, model_headers, agent_headers, client_headers, client_info_name=None, has_http_request=True)[source]

Resolve the routing context for one request.

model_headers / agent_headers / client_headers are ordered candidate lists: the first header with a non-empty value supplies the corresponding field, and agent_headers then client_headers also supply the route. session is taken from the first non-empty client_headers entry whose name contains session. Header lookup is case-insensitive (and treats _ as -).

When enabled is false no field is extracted at all: the request gets the disabled context, but still carries the real transport.

Parameters:
  • headers (Mapping[str, str])

  • enabled (bool)

  • model_headers (Sequence[str])

  • agent_headers (Sequence[str])

  • client_headers (Sequence[str])

  • client_info_name (str | None)

  • has_http_request (bool)

Return type:

RoutingContext

novelai_image_mcp.routing.sanitize_route(value)[source]

Return value if it is a safe directory name, else "unknown".

Accepts only [A-Za-z0-9._-]{1,64}. Path separators, whitespace, non-ASCII text, empty strings, over-long values and dot-only values (. / .., i.e. directory traversal) are all rejected.

Parameters:

value (str)

Return type:

str