diff --git a/src/cloudflare/_utils/_path.py b/src/cloudflare/_utils/_path.py index 4d6e1e4cbca..cf9b3150243 100644 --- a/src/cloudflare/_utils/_path.py +++ b/src/cloudflare/_utils/_path.py @@ -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: @@ -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. @@ -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`. @@ -61,16 +80,19 @@ 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) @@ -78,6 +100,11 @@ def _interpolate( 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. @@ -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 diff --git a/src/cloudflare/resources/ai/ai.py b/src/cloudflare/resources/ai/ai.py index 43f5734a994..2cb58a1c26b 100644 --- a/src/cloudflare/resources/ai/ai.py +++ b/src/cloudflare/resources/ai/ai.py @@ -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( { @@ -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( { diff --git a/tests/test_utils/test_path.py b/tests/test_utils/test_path.py index 7105fc49bed..8db6a063cf7 100644 --- a/tests/test_utils/test_path.py +++ b/tests/test_utils/test_path.py @@ -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: @@ -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: