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
41 changes: 35 additions & 6 deletions src/cloudflare/_utils/_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,12 @@
# Matches '.' or '..' where each dot is either literal or percent-encoded (%2e / %2E).
_DOT_SEGMENT_RE = re.compile(r"^(?:\.|%2[eE]){1,2}$")

_PLACEHOLDER_RE = re.compile(r"\{(\w+)\}")
# A placeholder name may optionally be prefixed with '+', matching the RFC 6570
# "reserved expansion" operator. This is used for path segments where the value
# itself is allowed to contain additional '/' separators (e.g. AI model names
# like `@cf/meta/llama-3.1-8b-instruct`).
# https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.3
_PLACEHOLDER_RE = re.compile(r"\{(\+?\w+)\}")


def _quote_path_segment_part(value: str) -> str:
Expand All @@ -26,6 +31,17 @@ def _quote_path_segment_part(value: str) -> str:
return quote(value, safe="!$&'()*+,;=:@")


def _quote_path_segment_part_reserved(value: str) -> str:
"""Percent-encode `value` for use in a URI path segment, allowing '/' through unescaped.

Same as `_quote_path_segment_part` but additionally treats '/' as safe, per the
RFC 6570 reserved expansion operator (`{+name}`). Dot-segments are still rejected
by `path_template()` after interpolation, so this does not weaken path traversal
protection.
"""
return quote(value, safe="!$&'()*+,;=:@/")


def _quote_query_part(value: str) -> str:
"""Percent-encode `value` for use in a URI query string.

Expand All @@ -48,10 +64,13 @@ def _interpolate(
template: str,
values: Mapping[str, Any],
quoter: Callable[[str], str],
reserved_quoter: Callable[[str], str] | None = None,
) -> str:
"""Replace {name} placeholders in `template`, quoting each value with `quoter`.

Placeholder names are looked up in `values`.
Placeholder names are looked up in `values`. A placeholder written as `{+name}`
uses `reserved_quoter` instead of `quoter`, matching the RFC 6570 reserved
expansion operator (falls back to `quoter` if `reserved_quoter` is not given).

Raises:
KeyError: If a placeholder is not found in `values`.
Expand All @@ -61,23 +80,31 @@ def _interpolate(
parts = _PLACEHOLDER_RE.split(template)

for i in range(1, len(parts), 2):
name = parts[i]
raw_name = parts[i]
is_reserved = raw_name.startswith("+")
name = raw_name[1:] if is_reserved else raw_name
if name not in values:
raise KeyError(f"a value for placeholder {{{name}}} was not provided")
raise KeyError(f"a value for placeholder {{{raw_name}}} was not provided")
val = values[name]
if val is None:
parts[i] = "null"
elif isinstance(val, bool):
parts[i] = "true" if val else "false"
else:
parts[i] = quoter(str(values[name]))
active_quoter = reserved_quoter if is_reserved and reserved_quoter is not None else quoter
parts[i] = active_quoter(str(val))

return "".join(parts)


def path_template(template: str, /, **kwargs: Any) -> str:
"""Interpolate {name} placeholders in `template` from keyword arguments.

A placeholder can be written as `{+name}` (RFC 6570 reserved expansion) in the
path portion of the template to allow the interpolated value to contain
additional unescaped `/` separators, for parameters whose values are themselves
composed of multiple path segments (e.g. AI model names).

Args:
template: The template string containing {name} placeholders.
**kwargs: Keyword arguments to interpolate into the template.
Expand Down Expand Up @@ -107,7 +134,9 @@ def path_template(template: str, /, **kwargs: Any) -> str:
path_template = rest

# Interpolate each portion with the appropriate quoting rules.
path_result = _interpolate(path_template, kwargs, _quote_path_segment_part)
path_result = _interpolate(
path_template, kwargs, _quote_path_segment_part, reserved_quoter=_quote_path_segment_part_reserved
)

# Reject dot-segments (. and ..) in the final assembled path. The check
# runs after interpolation so that adjacent placeholders or a mix of static
Expand Down
4 changes: 2 additions & 2 deletions src/cloudflare/resources/ai/ai.py
Original file line number Diff line number Diff line change
Expand Up @@ -991,7 +991,7 @@ def run(
Optional[AIRunResponse],
self._post(
path_template(
"/accounts/{account_id}/ai/run/{model_name}", account_id=account_id, model_name=model_name
"/accounts/{account_id}/ai/run/{+model_name}", account_id=account_id, model_name=model_name
),
body=maybe_transform(
{
Expand Down Expand Up @@ -1971,7 +1971,7 @@ async def run(
Optional[AIRunResponse],
await self._post(
path_template(
"/accounts/{account_id}/ai/run/{model_name}", account_id=account_id, model_name=model_name
"/accounts/{account_id}/ai/run/{+model_name}", account_id=account_id, model_name=model_name
),
body=await async_maybe_transform(
{
Expand Down
14 changes: 14 additions & 0 deletions tests/test_utils/test_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,18 @@
),
("/v1/{val}", dict(val="x?admin=true"), "/v1/x%3Fadmin=true"), # query injection
("/v1/{val}", dict(val="x#admin"), "/v1/x%23admin"), # fragment injection
# Reserved expansion ({+name}): '/' is preserved in the path portion
(
"/accounts/{account_id}/ai/run/{+model_name}",
dict(account_id="a1", model_name="@cf/vendor/model"),
"/accounts/a1/ai/run/@cf/vendor/model",
),
("/v1/{+v}", dict(v="a/b/c"), "/v1/a/b/c"),
("/v1/{+v}", dict(v="a b"), "/v1/a%20b"), # other characters still escaped
("/v1/{+v}", dict(v="a?b"), "/v1/a%3Fb"), # '?' still escaped, only '/' is preserved
("/v1/{+v}?q={v}", dict(v="a/b"), "/v1/a/b?q=a/b"), # reserved flag only affects the path portion
("/v1/{+v}", dict(v=None), "/v1/null"),
("/v1/{+v}", dict(v=True), "/v1/true"),
],
)
def test_interpolation(template: str, kwargs: dict[str, Any], expected: str) -> None:
Expand Down Expand Up @@ -82,6 +94,8 @@ def test_missing_kwarg_raises_key_error() -> None:
("/v1/.%2E/{x}", dict(x="ok")), # mixed encoded ".." in static
("/v1/{v}?q=1", dict(v="..")),
("/v1/{v}#frag", dict(v="..")),
("/v1/{+v}", dict(v="..")), # reserved expansion still rejects dot-segments
("/v1/{+v}", dict(v="../../secret")),
],
)
def test_dot_segment_rejected(template: str, kwargs: dict[str, Any]) -> None:
Expand Down