mirror of
https://github.com/searxng/searxng.git
synced 2026-10-25 05:44:14 +00:00
[feat] brave: parse JavaScript IIFE to fix engine
- Should resolve #6819 by reverting Accept-Encoding headers back to the curl_cffi default value - Adds IIFE Parsing / Data Extraction that appears after headers update
This commit is contained in:
1 parent
62366b05c0
commit
d47cb96bfb
4 files changed
+309
-29
No files matched your search
+102
-21
@@ -118,6 +118,7 @@ Implementations
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import typing as t
|
||||
from collections.abc import Callable
|
||||
from urllib.parse import urlencode
|
||||
@@ -129,7 +130,14 @@ from searx.enginelib.traits import EngineTraits
|
||||
from searx.exceptions import SearxEngineResponseException
|
||||
from searx.result_types import EngineResults, MainResult, Video
|
||||
from searx.result_types.image import Image
|
||||
from searx.utils import html_to_text, js_obj_str_to_json_str, js_obj_str_to_python
|
||||
from searx.utils import (
|
||||
html_to_text,
|
||||
js_find_matching_bracket,
|
||||
js_obj_str_to_json_str,
|
||||
js_obj_str_to_python,
|
||||
js_split_args,
|
||||
js_find_variable_positions,
|
||||
)
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
from searx.extended_types import SXNG_Response
|
||||
@@ -188,6 +196,16 @@ time_range_map: dict[str, str] = {
|
||||
"year": "py",
|
||||
}
|
||||
|
||||
IIFE_RE = re.compile(
|
||||
r"""
|
||||
\(function\s*\( # (function(
|
||||
(?P<params>[^)]*) # formal parameters
|
||||
\)\s*\{\s*return\s* # ){ return
|
||||
(?P<body>\{) # the object literal starts
|
||||
""",
|
||||
re.VERBOSE | re.DOTALL,
|
||||
)
|
||||
|
||||
|
||||
def request(query: str, params: dict[str, t.Any]) -> None:
|
||||
|
||||
@@ -207,12 +225,10 @@ def request(query: str, params: dict[str, t.Any]) -> None:
|
||||
if brave_category == "goggles":
|
||||
args["goggles_id"] = Goggles
|
||||
|
||||
params["headers"]["Accept-Encoding"] = "gzip, deflate"
|
||||
params["url"] = f"{base_url}{brave_category}?{urlencode(args)}"
|
||||
logger.debug("url %s", params["url"])
|
||||
|
||||
# set properties in the cookies
|
||||
|
||||
params["cookies"]["safesearch"] = safesearch_map.get(params["safesearch"], "off")
|
||||
# the useLocation is IP based, we use cookie "country" for the region
|
||||
params["cookies"]["useLocation"] = "0"
|
||||
@@ -235,25 +251,90 @@ def _extract_published_date(published_date_raw: str | None):
|
||||
return None
|
||||
|
||||
|
||||
def extract_iife_object(data_src: str, match: re.Match[str]) -> tuple[str, int]:
|
||||
"""
|
||||
Extract and substitute the object literal from a JavaScript IIFE.
|
||||
Returns the transformed object literal and the position immediately after the IIFE call.
|
||||
"""
|
||||
body_start = match.start("body")
|
||||
body_end = js_find_matching_bracket(data_src, body_start, "{", "}")
|
||||
obj_str = data_src[body_start : body_end + 1]
|
||||
|
||||
# 1. Locate the argument call execution signature block: ...)(arg1, arg2)
|
||||
call_start = data_src.find("(", body_end)
|
||||
if call_start == -1:
|
||||
raise ValueError("IIFE missing call arguments")
|
||||
call_end = js_find_matching_bracket(data_src, call_start, "(", ")")
|
||||
|
||||
# 2. map arguments
|
||||
args = js_split_args(data_src[call_start + 1 : call_end])
|
||||
params = [p.strip() for p in match["params"].split(",") if p.strip()]
|
||||
mapping = dict(zip(params, args))
|
||||
|
||||
# 3. substitute idents
|
||||
variables = js_find_variable_positions(obj_str)
|
||||
for variable_name, pos in reversed(variables):
|
||||
if variable_name in mapping:
|
||||
value = mapping[variable_name]
|
||||
obj_str = obj_str[:pos] + value + obj_str[pos + len(variable_name) :]
|
||||
|
||||
return obj_str, call_end + 1
|
||||
|
||||
|
||||
def extract_json_data(text: str) -> dict[str, t.Any]:
|
||||
# Example script source containing the data:
|
||||
#
|
||||
# kit.start(app, element, {
|
||||
# node_ids: [0, 19],
|
||||
# data: [{type:"data",data: .... ["q","goggles_id"],route:1,url:1}}]
|
||||
# ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
# form: null,
|
||||
# error: null
|
||||
# });
|
||||
start = text.index("data: [{")
|
||||
newline = text.index("\n", start)
|
||||
end = text.rindex("}}]", start, newline)
|
||||
js_obj_str = "{" + text[start:end] + "}}]}"
|
||||
# js_obj_str = js_obj_str.replace("\xa0", "") # remove ASCII for
|
||||
# js_obj_str = js_obj_str.replace(r"\u003C", "<").replace(r"\u003c", "<") # fix broken HTML tags in strings
|
||||
json_str = js_obj_str_to_json_str(js_obj_str)
|
||||
data: dict[str, t.Any] = json.loads(json_str)
|
||||
return data
|
||||
"""
|
||||
Extract the SvelteKit data payload that Brave embeds in the page.
|
||||
|
||||
Handles:
|
||||
- plain-object form
|
||||
- pure IIFE form
|
||||
- mixed form (IIFE + plain object) that appears on some news pages
|
||||
"""
|
||||
marker = "data: ["
|
||||
start = text.find(marker)
|
||||
if start == -1:
|
||||
raise ValueError("Brave data payload not found (no 'data: [')")
|
||||
|
||||
array_start = start + len(marker) - 1 # Points directly at '['
|
||||
array_end = js_find_matching_bracket(text, array_start, "[", "]")
|
||||
data_src = text[array_start + 1 : array_end]
|
||||
|
||||
objects: list[str] = []
|
||||
pos = 0
|
||||
n = len(data_src)
|
||||
|
||||
while pos < n:
|
||||
# Move past whitespace and commas safely
|
||||
while pos < n and data_src[pos] in " \t\n\r,":
|
||||
pos += 1
|
||||
if pos >= n:
|
||||
break
|
||||
|
||||
# Case 1: IIFE Object Signature Matches
|
||||
if m := IIFE_RE.match(data_src, pos):
|
||||
obj_str, next_pos = extract_iife_object(data_src, m)
|
||||
objects.append(obj_str)
|
||||
pos = next_pos
|
||||
# Brave sometimes emits an extra ')' after the IIFE call
|
||||
while pos < n and data_src[pos] in " \t\n\r)":
|
||||
pos += 1
|
||||
continue
|
||||
|
||||
# Case 2: Standard raw plain JSON mapping block {type:"data", ...}
|
||||
if data_src[pos] == "{":
|
||||
obj_end = js_find_matching_bracket(data_src, pos, "{", "}")
|
||||
objects.append(data_src[pos : obj_end + 1])
|
||||
pos = obj_end + 1
|
||||
continue
|
||||
|
||||
raise ValueError(f"unexpected token in data array at offset {pos}: {data_src[pos : pos + 40]!r}")
|
||||
|
||||
if not objects:
|
||||
raise ValueError("no data objects found inside Brave payload")
|
||||
|
||||
combined = '{"data":[' + ",".join(objects) + "]}"
|
||||
json_str = js_obj_str_to_json_str(combined)
|
||||
return json.loads(json_str)
|
||||
|
||||
|
||||
def response(resp: "SXNG_Response") -> EngineResults:
|
||||
|
||||
+117
-6
@@ -7,6 +7,7 @@ import re
|
||||
import importlib
|
||||
import importlib.util
|
||||
import json
|
||||
import string
|
||||
import types
|
||||
|
||||
import typing as t
|
||||
@@ -440,7 +441,7 @@ def to_string(obj: t.Any) -> str:
|
||||
return repr(obj)
|
||||
|
||||
|
||||
def ecma_unescape(string: str) -> str:
|
||||
def ecma_unescape(text: str) -> str:
|
||||
"""Python implementation of the unescape javascript function
|
||||
|
||||
https://www.ecma-international.org/ecma-262/6.0/#sec-unescape-string
|
||||
@@ -455,20 +456,20 @@ def ecma_unescape(string: str) -> str:
|
||||
'ó'
|
||||
"""
|
||||
# "%u5409" becomes "吉"
|
||||
string = _ECMA_UNESCAPE4_RE.sub(lambda e: chr(int(e.group(1), 16)), string)
|
||||
text = _ECMA_UNESCAPE4_RE.sub(lambda e: chr(int(e.group(1), 16)), text)
|
||||
# "%20" becomes " ", "%F3" becomes "ó"
|
||||
string = _ECMA_UNESCAPE2_RE.sub(lambda e: chr(int(e.group(1), 16)), string)
|
||||
return string
|
||||
text = _ECMA_UNESCAPE2_RE.sub(lambda e: chr(int(e.group(1), 16)), text)
|
||||
return text
|
||||
|
||||
|
||||
def remove_pua_from_str(string: str):
|
||||
def remove_pua_from_str(text: str):
|
||||
"""Removes unicode's "PRIVATE USE CHARACTER"s (PUA_) from a string.
|
||||
|
||||
.. _PUA: https://en.wikipedia.org/wiki/Private_Use_Areas
|
||||
"""
|
||||
pua_ranges = ((0xE000, 0xF8FF), (0xF0000, 0xFFFFD), (0x100000, 0x10FFFD))
|
||||
s: list[str] = []
|
||||
for c in string:
|
||||
for c in text:
|
||||
i = ord(c)
|
||||
if any(a <= i <= b for (a, b) in pua_ranges):
|
||||
continue
|
||||
@@ -793,6 +794,116 @@ def js_obj_str_to_json_str(js_obj_str: str) -> str:
|
||||
return s
|
||||
|
||||
|
||||
def js_find_matching_bracket(text: str, start: int, open_char: str = "{", close_char: str = "}") -> int:
|
||||
"""Return index of the matching closing bracket, safely ignoring JS string literals."""
|
||||
depth = 0
|
||||
i = start
|
||||
in_string = None
|
||||
escape_ = False
|
||||
|
||||
while i < len(text):
|
||||
c = text[i]
|
||||
if in_string:
|
||||
if escape_:
|
||||
escape_ = False
|
||||
elif c == "\\":
|
||||
escape_ = True
|
||||
elif c == in_string:
|
||||
in_string = None
|
||||
elif c in ('"', "'", "`"):
|
||||
in_string = c
|
||||
elif c == open_char:
|
||||
depth += 1
|
||||
elif c == close_char:
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return i
|
||||
i += 1
|
||||
raise ValueError(f"unbalanced {open_char}{close_char} starting at {start}")
|
||||
|
||||
|
||||
def js_split_args(args_str: str) -> list[str]:
|
||||
"""Split a JS argument list, respecting strings and nested brackets."""
|
||||
args: list[str] = []
|
||||
last_pos = 0
|
||||
in_string = None
|
||||
escape_ = False
|
||||
depth = 0
|
||||
|
||||
for idx, c in enumerate(f"{args_str},"):
|
||||
if in_string:
|
||||
if escape_:
|
||||
escape_ = False
|
||||
elif c == "\\":
|
||||
escape_ = True
|
||||
elif c == in_string:
|
||||
in_string = None
|
||||
elif c in ('"', "'", "`"):
|
||||
in_string = c
|
||||
elif c in ("(", "[", "{"):
|
||||
depth += 1
|
||||
elif c in (")", "]", "}"):
|
||||
depth -= 1
|
||||
elif c == "," and depth == 0:
|
||||
if item := args_str[last_pos:idx].strip():
|
||||
args.append(item)
|
||||
last_pos = idx + 1
|
||||
return args
|
||||
|
||||
|
||||
def js_find_variable_positions(js_obj: str) -> list[tuple[str, int]]:
|
||||
"""Find all variables referenced in the JavaScript object."""
|
||||
# pylint: disable=too-many-branches
|
||||
# pylint: disable=unsubscriptable-object
|
||||
|
||||
matches: list[tuple[str, int]] = []
|
||||
|
||||
in_string = False
|
||||
in_escape = False
|
||||
|
||||
# flat array like [1,2,3], not e.g. [{"a": 3}]
|
||||
in_flat_array = False
|
||||
# if the current position is inside the place of
|
||||
# a JS value, e.g. the range of ` bar` in {"foo": bar}
|
||||
in_value = False
|
||||
current_var: tuple[str, int] | None = None
|
||||
|
||||
for pos, c in enumerate(js_obj):
|
||||
if in_string:
|
||||
if in_escape:
|
||||
in_escape = False
|
||||
elif c == "\\":
|
||||
in_escape = True
|
||||
elif c == in_string:
|
||||
in_string = None
|
||||
elif c in ("'", "\"", "`"):
|
||||
in_string = c
|
||||
elif c == ":":
|
||||
in_value = True
|
||||
elif c in (",", "{", "}", "[", "]"):
|
||||
# handle matching variables in arrays like [a,b,c]
|
||||
# i.e. if we encounter a "," in such a flat array,
|
||||
# we're still inside a value
|
||||
if c == "[":
|
||||
in_flat_array = True
|
||||
elif c != ",":
|
||||
in_flat_array = False
|
||||
in_value = in_flat_array
|
||||
if current_var:
|
||||
# reached end of variable name
|
||||
if current_var[0] not in ("void0", "true", "false", "null", "undefined"):
|
||||
matches.append(current_var)
|
||||
current_var = None
|
||||
elif in_value:
|
||||
if current_var and c in (string.ascii_letters + string.digits + "_$"):
|
||||
current_var = (current_var[0] + c, current_var[1])
|
||||
elif c in string.ascii_letters + "_$":
|
||||
# start of new variable name
|
||||
current_var = (c, pos)
|
||||
|
||||
return matches
|
||||
|
||||
|
||||
def parse_duration_string(duration_str: str) -> timedelta | None:
|
||||
"""Parse a time string in format MM:SS or HH:MM:SS and convert it to a
|
||||
:obj:`datetime.timedelta` object.
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
# pylint: disable=missing-module-docstring,line-too-long
|
||||
|
||||
import json
|
||||
|
||||
from parameterized import parameterized
|
||||
|
||||
from searx.engines import brave
|
||||
from tests import SearxTestCase
|
||||
|
||||
|
||||
class BraveEngineTests(SearxTestCase):
|
||||
'''Unit tests specific to the Brave engine'''
|
||||
|
||||
# extract_iife_object
|
||||
@parameterized.expand(
|
||||
[
|
||||
(
|
||||
"""(function (a, b, c){ return {"a": a, "b": b, "c": c };})("first", 2, {"third": 3})""", # basic shape, numeric, nested object
|
||||
{"a": "first", "b": 2, "c": {"third": 3}},
|
||||
),
|
||||
(
|
||||
'(function ($,aa,a,a1){return {"first": $, "second": true, "third": [aa,a,a1]};})("actual","x","y","z")', # non-alpha param, literal, and array
|
||||
{"first": "actual", "second": True, "third": ["x", "y", "z"]},
|
||||
),
|
||||
]
|
||||
)
|
||||
def test_javascript_iife_parser(self, js_code, expected_output):
|
||||
iife_match = brave.IIFE_RE.match(js_code)
|
||||
self.assertIsNotNone(iife_match)
|
||||
|
||||
iife_object, _ = brave.extract_iife_object(js_code, iife_match) # pyright: ignore[reportArgumentType]
|
||||
parsed = json.loads(iife_object)
|
||||
self.assertEqual(parsed, expected_output)
|
||||
@@ -58,7 +58,7 @@ class TestUtils(SearxTestCase):
|
||||
|
||||
def test_extract_text_error_empty(self):
|
||||
with self.assertRaises(ValueError):
|
||||
utils.extract_text({})
|
||||
utils.extract_text({}) # pyright: ignore[reportArgumentType]
|
||||
|
||||
def test_extract_url(self):
|
||||
def f(html_str, search_url):
|
||||
@@ -132,7 +132,7 @@ class TestXPathUtils(SearxTestCase): # pylint: disable=missing-class-docstring
|
||||
utils.get_xpath(lxml.etree.XPath('//a'))
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
utils.get_xpath([])
|
||||
utils.get_xpath([]) # pyright: ignore[reportArgumentType]
|
||||
|
||||
def test_get_xpath_invalid(self):
|
||||
invalid_xpath = '//a[0].text'
|
||||
@@ -214,3 +214,57 @@ class TestXPathUtils(SearxTestCase): # pylint: disable=missing-class-docstring
|
||||
key, counter = solution # pyright: ignore[reportGeneralTypeIssues]
|
||||
self.assertEqual(key, "71087e0b10d819181fc5463588a230b5744e6b9f19b86667c71ff73a9c7567aa")
|
||||
self.assertEqual(counter, 260)
|
||||
|
||||
|
||||
class TestJsHelpers(SearxTestCase):
|
||||
# js_find_matching_bracket
|
||||
@parameterized.expand(
|
||||
[
|
||||
# simple object
|
||||
('{"a": 1, "b": {"c": 2}}', 0, "{", "}", 22),
|
||||
# nested with strings that contain braces
|
||||
('{"a": "hello { world }", "b": 2}', 0, "{", "}", 31),
|
||||
# array
|
||||
("[1, 2, [3, 4], 5]", 0, "[", "]", 16),
|
||||
# string with escaped quote
|
||||
(r'{"a": "he said \"hi\"", "b": 1}', 0, "{", "}", 30),
|
||||
]
|
||||
)
|
||||
def test_js_find_matching_bracket(self, text, start, open_c, close_c, expected_end):
|
||||
end = utils.js_find_matching_bracket(text, start, open_c, close_c)
|
||||
self.assertEqual(end, expected_end)
|
||||
|
||||
def test_js_find_matching_bracket_unbalanced(self):
|
||||
with self.assertRaises(ValueError):
|
||||
utils.js_find_matching_bracket("{a: 1", 0)
|
||||
|
||||
# js_split_args
|
||||
@parameterized.expand(
|
||||
[
|
||||
('"first", 2, {"third": 3}', ['"first"', "2", '{"third": 3}']),
|
||||
('"a", "b,c", [1,2]', ['"a"', '"b,c"', "[1,2]"]),
|
||||
("$", ["$"]),
|
||||
("", []),
|
||||
('"only"', ['"only"']),
|
||||
('1, "two", true, null', ["1", '"two"', "true", "null"]),
|
||||
]
|
||||
)
|
||||
def test_js_split_args(self, args_str, expected):
|
||||
self.assertEqual(utils.js_split_args(args_str), expected)
|
||||
|
||||
# js_substitute_identifiers
|
||||
@parameterized.expand(
|
||||
[
|
||||
# basic
|
||||
('{"a": a, "b": b}', [("a", 6), ("b", 14)]),
|
||||
# $ and multi-char
|
||||
('{"first": $, "second": aa}', [("$", 10), ("aa", 23)]),
|
||||
# identifier inside a string must NOT be matched
|
||||
('{"msg": "hello a world", "val": a}', [("a", 32)]),
|
||||
# ints, booleans and null are no identifiers
|
||||
('{"msg": null, "val": true, "status": 1}', []),
|
||||
]
|
||||
)
|
||||
def test_js_find_variables(self, js_obj, expected):
|
||||
result = utils.js_find_variable_positions(js_obj)
|
||||
self.assertEqual(result, expected)
|
||||
Reference in new issue
Block a user