Verified Commit 2cc44e8f authored by Kevin Morris's avatar Kevin Morris
Browse files

fix(rpc): perform regex match against callback name



Since we're in the hot path, a constant re.compiled
JSONP_EXPR is defined for checks against the callback.

Additionally, reorganized `content_type` and `content`
to avoid performing a DB query when we encounter a
regex mismatch.
Signed-off-by: Kevin Morris's avatarKevin Morris <kevr@0cost.org>
parent 12b4269b
import hashlib
import re
from http import HTTPStatus
from typing import List, Optional
......@@ -61,6 +62,9 @@ def parse_args(request: Request):
return args
JSONP_EXPR = re.compile(r'^[a-zA-Z0-9()_.]{1,128}$')
@router.get("/rpc")
async def rpc(request: Request,
v: Optional[int] = Query(default=None),
......@@ -78,6 +82,16 @@ async def rpc(request: Request,
return JSONResponse(rpc.error("Rate limit reached"),
status_code=int(HTTPStatus.TOO_MANY_REQUESTS))
# If `callback` was provided, produce a text/javascript response
# valid for the jsonp callback. Otherwise, by default, return
# application/json containing `output`.
content_type = "application/json"
if callback:
if not re.match(JSONP_EXPR, callback):
return rpc.error("Invalid callback name.")
content_type = "text/javascript"
# Prepare list of arguments for input. If 'arg' was given, it'll
# be a list with one element.
arguments = parse_args(request)
......@@ -92,21 +106,14 @@ async def rpc(request: Request,
md5.update(content)
etag = md5.hexdigest()
# If `callback` was provided, produce a text/javascript response
# valid for the jsonp callback. Otherwise, by default, return
# application/json containing `output`.
# Note: Being the API hot path, `content` is not defaulted to
# avoid copying the JSON string in the case callback is provided.
content_type = "application/json"
if callback:
print("callback called")
content_type = "text/javascript"
content = f"/**/{callback}({content.decode()})"
# The ETag header expects quotes to surround any identifier.
# https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/ETag
headers = {
"Content-Type": content_type,
"ETag": f'"{etag}"'
}
if callback:
content = f"/**/{callback}({content.decode()})"
return Response(content, headers=headers)
......@@ -624,3 +624,9 @@ def test_rpc_jsonp_callback():
"/rpc?v=5&type=search&arg=big&callback=jsonCallback")
assert response.headers.get("content-type") == "text/javascript"
assert re.search(r'^/\*\*/jsonCallback\(.*\)$', response.text) is not None
# Test an invalid callback name; we get an application/json error.
response = make_request(
"/rpc?v=5&type=search&arg=big&callback=jsonCallback!")
assert response.headers.get("content-type") == "application/json"
assert response.json().get("error") == "Invalid callback name."
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment