Back-port JS interpreter upgrade from yt-dlp PR #1437

pull/30577/head
df 3 years ago committed by dirkf
parent 5f5de51a49
commit 96f87aaa3b

@ -112,6 +112,57 @@ class TestJSInterpreter(unittest.TestCase):
''') ''')
self.assertEqual(jsi.call_function('z'), 5) self.assertEqual(jsi.call_function('z'), 5)
def test_for_loop(self):
# function x() { a=0; for (i=0; i-10; i++) {a++} a }
jsi = JSInterpreter('''
function x() { a=0; for (i=0; i-10; i = i + 1) {a++} a }
''')
self.assertEqual(jsi.call_function('x'), 10)
def test_switch(self):
jsi = JSInterpreter('''
function x(f) { switch(f){
case 1:f+=1;
case 2:f+=2;
case 3:f+=3;break;
case 4:f+=4;
default:f=0;
} return f }
''')
self.assertEqual(jsi.call_function('x', 1), 7)
self.assertEqual(jsi.call_function('x', 3), 6)
self.assertEqual(jsi.call_function('x', 5), 0)
def test_try(self):
jsi = JSInterpreter('''
function x() { try{return 10} catch(e){return 5} }
''')
self.assertEqual(jsi.call_function('x'), 10)
def test_for_loop_continue(self):
jsi = JSInterpreter('''
function x() { a=0; for (i=0; i-10; i++) { continue; a++ } a }
''')
self.assertEqual(jsi.call_function('x'), 0)
def test_for_loop_break(self):
jsi = JSInterpreter('''
function x() { a=0; for (i=0; i-10; i++) { break; a++ } a }
''')
self.assertEqual(jsi.call_function('x'), 0)
def test_literal_list(self):
jsi = JSInterpreter('''
function x() { [1, 2, "asdf", [5, 6, 7]][3] }
''')
self.assertEqual(jsi.call_function('x'), [5, 6, 7])
def test_comma(self):
jsi = JSInterpreter('''
function x() { a=5; a -= 1, a+=3; return a }
''')
self.assertEqual(jsi.call_function('x'), 7)
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()

@ -21,6 +21,10 @@ import subprocess
import sys import sys
import xml.etree.ElementTree import xml.etree.ElementTree
try:
import collections.abc as compat_collections_abc
except ImportError:
import collections as compat_collections_abc
try: try:
import urllib.request as compat_urllib_request import urllib.request as compat_urllib_request
@ -3025,6 +3029,7 @@ __all__ = [
'compat_b64decode', 'compat_b64decode',
'compat_basestring', 'compat_basestring',
'compat_chr', 'compat_chr',
'compat_collections_abc',
'compat_cookiejar', 'compat_cookiejar',
'compat_cookiejar_Cookie', 'compat_cookiejar_Cookie',
'compat_cookies', 'compat_cookies',

@ -8,6 +8,15 @@ from .utils import (
ExtractorError, ExtractorError,
remove_quotes, remove_quotes,
) )
from .compat import (
compat_collections_abc
)
MutableMapping = compat_collections_abc.MutableMapping
class Nonlocal:
pass
_OPERATORS = [ _OPERATORS = [
('|', operator.or_), ('|', operator.or_),
@ -22,11 +31,55 @@ _OPERATORS = [
('*', operator.mul), ('*', operator.mul),
] ]
_ASSIGN_OPERATORS = [(op + '=', opfunc) for op, opfunc in _OPERATORS] _ASSIGN_OPERATORS = [(op + '=', opfunc) for op, opfunc in _OPERATORS]
_ASSIGN_OPERATORS.append(('=', lambda cur, right: right)) _ASSIGN_OPERATORS.append(('=', (lambda cur, right: right)))
_NAME_RE = r'[a-zA-Z_$][a-zA-Z_$0-9]*' _NAME_RE = r'[a-zA-Z_$][a-zA-Z_$0-9]*'
class JS_Break(ExtractorError):
def __init__(self):
ExtractorError.__init__(self, 'Invalid break')
class JS_Continue(ExtractorError):
def __init__(self):
ExtractorError.__init__(self, 'Invalid continue')
class LocalNameSpace(MutableMapping):
def __init__(self, *stack):
self.stack = tuple(stack)
def __getitem__(self, key):
for scope in self.stack:
if key in scope:
return scope[key]
raise KeyError(key)
def __setitem__(self, key, value):
for scope in self.stack:
if key in scope:
scope[key] = value
break
else:
self.stack[0][key] = value
return value
def __delitem__(self, key):
raise NotImplementedError('Deleting is not supported')
def __iter__(self):
for scope in self.stack:
for scope_item in iter(scope):
yield scope_item
def __len__(self, key):
return len(iter(self))
def __repr__(self):
return 'LocalNameSpace%s' % (self.stack, )
class JSInterpreter(object): class JSInterpreter(object):
def __init__(self, code, objects=None): def __init__(self, code, objects=None):
if objects is None: if objects is None:
@ -34,11 +87,58 @@ class JSInterpreter(object):
self.code = code self.code = code
self._functions = {} self._functions = {}
self._objects = objects self._objects = objects
self.__named_object_counter = 0
def _named_object(self, namespace, obj):
self.__named_object_counter += 1
name = '__youtube_dl_jsinterp_obj%s' % (self.__named_object_counter, )
namespace[name] = obj
return name
@staticmethod
def _separate(expr, delim=',', max_split=None):
if not expr:
return
parens = {'(': 0, '{': 0, '[': 0, ']': 0, '}': 0, ')': 0}
start, splits, pos, max_pos = 0, 0, 0, len(delim) - 1
for idx, char in enumerate(expr):
if char in parens:
parens[char] += 1
is_in_parens = (parens['['] - parens[']']
or parens['('] - parens[')']
or parens['{'] - parens['}'])
if char == delim[pos] and not is_in_parens:
if pos == max_pos:
pos = 0
yield expr[start: idx - max_pos]
start = idx + 1
splits += 1
if max_split and splits >= max_split:
break
else:
pos += 1
else:
pos = 0
yield expr[start:]
@staticmethod
def _separate_at_paren(expr, delim):
separated = list(JSInterpreter._separate(expr, delim, 1))
if len(separated) < 2:
raise ExtractorError('No terminating paren {0} in {1}'.format(delim, expr))
return separated[0][1:].strip(), separated[1].strip()
def interpret_statement(self, stmt, local_vars, allow_recursion=100): def interpret_statement(self, stmt, local_vars, allow_recursion=100):
if allow_recursion < 0: if allow_recursion < 0:
raise ExtractorError('Recursion limit reached') raise ExtractorError('Recursion limit reached')
sub_statements = list(self._separate(stmt, ';'))
stmt = (sub_statements or ['']).pop()
for sub_stmt in sub_statements:
ret, should_abort = self.interpret_statement(sub_stmt, local_vars, allow_recursion - 1)
if should_abort:
return ret
should_abort = False should_abort = False
stmt = stmt.lstrip() stmt = stmt.lstrip()
stmt_m = re.match(r'var\s', stmt) stmt_m = re.match(r'var\s', stmt)
@ -61,25 +161,119 @@ class JSInterpreter(object):
if expr == '': # Empty expression if expr == '': # Empty expression
return None return None
if expr.startswith('{'):
inner, outer = self._separate_at_paren(expr, '}')
inner, should_abort = self.interpret_statement(inner, local_vars, allow_recursion - 1)
if not outer or should_abort:
return inner
else:
expr = json.dumps(inner) + outer
if expr.startswith('('): if expr.startswith('('):
parens_count = 0 inner, outer = self._separate_at_paren(expr, ')')
for m in re.finditer(r'[()]', expr): inner = self.interpret_expression(inner, local_vars, allow_recursion)
if m.group(0) == '(': if not outer:
parens_count += 1 return inner
else:
expr = json.dumps(inner) + outer
if expr.startswith('['):
inner, outer = self._separate_at_paren(expr, ']')
name = self._named_object(local_vars, [
self.interpret_expression(item, local_vars, allow_recursion)
for item in self._separate(inner)])
expr = name + outer
m = re.match(r'try\s*', expr)
if m:
if expr[m.end()] == '{':
try_expr, expr = self._separate_at_paren(expr[m.end():], '}')
else:
try_expr, expr = expr[m.end() - 1:], ''
ret, should_abort = self.interpret_statement(try_expr, local_vars, allow_recursion - 1)
if should_abort:
return ret
return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
m = re.match(r'(?:(?P<catch>catch)|(?P<for>for)|(?P<switch>switch))\s*\(', expr)
md = m.groupdict() if m else {}
if md.get('catch'):
# We ignore the catch block
_, expr = self._separate_at_paren(expr, '}')
return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
elif md.get('for'):
def raise_constructor_error(c):
raise ExtractorError(
'Premature return in the initialization of a for loop in {0!r}'.format(c))
constructor, remaining = self._separate_at_paren(expr[m.end() - 1:], ')')
if remaining.startswith('{'):
body, expr = self._separate_at_paren(remaining, '}')
else:
m = re.match(r'switch\s*\(', remaining) # FIXME
if m:
switch_val, remaining = self._separate_at_paren(remaining[m.end() - 1:], ')')
body, expr = self._separate_at_paren(remaining, '}')
body = 'switch(%s){%s}' % (switch_val, body)
else: else:
parens_count -= 1 body, expr = remaining, ''
if parens_count == 0: start, cndn, increment = self._separate(constructor, ';')
sub_expr = expr[1:m.start()] if self.interpret_statement(start, local_vars, allow_recursion - 1)[1]:
sub_result = self.interpret_expression( raise_constructor_error(constructor)
sub_expr, local_vars, allow_recursion) while True:
remaining_expr = expr[m.end():].strip() if not self.interpret_expression(cndn, local_vars, allow_recursion):
if not remaining_expr: break
return sub_result try:
else: ret, should_abort = self.interpret_statement(body, local_vars, allow_recursion - 1)
expr = json.dumps(sub_result) + remaining_expr if should_abort:
return ret
except JS_Break:
break
except JS_Continue:
pass
if self.interpret_statement(increment, local_vars, allow_recursion - 1)[1]:
raise_constructor_error(constructor)
return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
elif md.get('switch'):
switch_val, remaining = self._separate_at_paren(expr[m.end() - 1:], ')')
switch_val = self.interpret_expression(switch_val, local_vars, allow_recursion)
body, expr = self._separate_at_paren(remaining, '}')
body, default = body.split('default:') if 'default:' in body else (body, None)
items = body.split('case ')[1:]
if default:
items.append('default:%s' % (default, ))
matched = False
for item in items:
case, stmt = [i.strip() for i in self._separate(item, ':', 1)]
matched = matched or case == 'default' or switch_val == self.interpret_expression(case, local_vars, allow_recursion)
if matched:
try:
ret, should_abort = self.interpret_statement(stmt, local_vars, allow_recursion - 1)
if should_abort:
return ret
except JS_Break:
break break
else: return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
raise ExtractorError('Premature end of parens in %r' % expr)
# Comma separated statements
sub_expressions = list(self._separate(expr))
expr = sub_expressions.pop().strip() if sub_expressions else ''
for sub_expr in sub_expressions:
self.interpret_expression(sub_expr, local_vars, allow_recursion)
for m in re.finditer(r'''(?x)
(?P<pre_sign>\+\+|--)(?P<var1>%(_NAME_RE)s)|
(?P<var2>%(_NAME_RE)s)(?P<post_sign>\+\+|--)''' % globals(), expr):
var = m.group('var1') or m.group('var2')
start, end = m.span()
sign = m.group('pre_sign') or m.group('post_sign')
ret = local_vars[var]
local_vars[var] += 1 if sign[0] == '+' else -1
if m.group('pre_sign'):
ret = local_vars[var]
expr = expr[:start] + json.dumps(ret) + expr[end:]
for op, opfunc in _ASSIGN_OPERATORS: for op, opfunc in _ASSIGN_OPERATORS:
m = re.match(r'''(?x) m = re.match(r'''(?x)
@ -88,14 +282,13 @@ class JSInterpreter(object):
(?P<expr>.*)$''' % (_NAME_RE, re.escape(op)), expr) (?P<expr>.*)$''' % (_NAME_RE, re.escape(op)), expr)
if not m: if not m:
continue continue
right_val = self.interpret_expression( right_val = self.interpret_expression(m.group('expr'), local_vars, allow_recursion)
m.group('expr'), local_vars, allow_recursion - 1)
if m.groupdict().get('index'): if m.groupdict().get('index'):
lvar = local_vars[m.group('out')] lvar = local_vars[m.group('out')]
idx = self.interpret_expression( idx = self.interpret_expression(m.group('index'), local_vars, allow_recursion)
m.group('index'), local_vars, allow_recursion) if not isinstance(idx, int):
assert isinstance(idx, int) raise ExtractorError('List indices must be integers: %s' % (idx, ))
cur = lvar[idx] cur = lvar[idx]
val = opfunc(cur, right_val) val = opfunc(cur, right_val)
lvar[idx] = val lvar[idx] = val
@ -109,8 +302,13 @@ class JSInterpreter(object):
if expr.isdigit(): if expr.isdigit():
return int(expr) return int(expr)
if expr == 'break':
raise JS_Break()
elif expr == 'continue':
raise JS_Continue()
var_m = re.match( var_m = re.match(
r'(?!if|return|true|false)(?P<name>%s)$' % _NAME_RE, r'(?!if|return|true|false|null)(?P<name>%s)$' % _NAME_RE,
expr) expr)
if var_m: if var_m:
return local_vars[var_m.group('name')] return local_vars[var_m.group('name')]
@ -124,91 +322,161 @@ class JSInterpreter(object):
r'(?P<in>%s)\[(?P<idx>.+)\]$' % _NAME_RE, expr) r'(?P<in>%s)\[(?P<idx>.+)\]$' % _NAME_RE, expr)
if m: if m:
val = local_vars[m.group('in')] val = local_vars[m.group('in')]
idx = self.interpret_expression( idx = self.interpret_expression(m.group('idx'), local_vars, allow_recursion)
m.group('idx'), local_vars, allow_recursion - 1)
return val[idx] return val[idx]
def raise_expr_error(where, op, exp):
raise ExtractorError('Premature {0} return of {1} in {2!r}'.format(where, op, exp))
for op, opfunc in _OPERATORS:
separated = list(self._separate(expr, op))
if len(separated) < 2:
continue
right_val = separated.pop()
left_val = op.join(separated)
left_val, should_abort = self.interpret_statement(
left_val, local_vars, allow_recursion - 1)
if should_abort:
raise_expr_error('left-side', op, expr)
right_val, should_abort = self.interpret_statement(
right_val, local_vars, allow_recursion - 1)
if should_abort:
raise_expr_error('right-side', op, expr)
return opfunc(left_val or 0, right_val)
m = re.match( m = re.match(
r'(?P<var>%s)(?:\.(?P<member>[^(]+)|\[(?P<member2>[^]]+)\])\s*(?:\(+(?P<args>[^()]*)\))?$' % _NAME_RE, r'(?P<var>%s)(?:\.(?P<member>[^(]+)|\[(?P<member2>[^]]+)\])\s*' % _NAME_RE,
expr) expr)
if m: if m:
variable = m.group('var') variable = m.group('var')
member = remove_quotes(m.group('member') or m.group('member2')) nl = Nonlocal()
arg_str = m.group('args')
if variable in local_vars: nl.member = remove_quotes(m.group('member') or m.group('member2'))
obj = local_vars[variable] arg_str = expr[m.end():]
else: if arg_str.startswith('('):
if variable not in self._objects: arg_str, remaining = self._separate_at_paren(arg_str, ')')
self._objects[variable] = self.extract_object(variable)
obj = self._objects[variable]
if arg_str is None:
# Member access
if member == 'length':
return len(obj)
return obj[member]
assert expr.endswith(')')
# Function call
if arg_str == '':
argvals = tuple()
else: else:
argvals = tuple([ arg_str, remaining = None, arg_str
def assertion(cndn, msg):
""" assert, but without risk of getting optimized out """
if not cndn:
raise ExtractorError('{0} {1}: {2}'.format(nl.member, msg, expr))
def eval_method():
# nonlocal member
member = nl.member
if variable == 'String':
obj = str
elif variable in local_vars:
obj = local_vars[variable]
else:
if variable not in self._objects:
self._objects[variable] = self.extract_object(variable)
obj = self._objects[variable]
if arg_str is None:
# Member access
if member == 'length':
return len(obj)
return obj[member]
# Function call
argvals = [
self.interpret_expression(v, local_vars, allow_recursion) self.interpret_expression(v, local_vars, allow_recursion)
for v in arg_str.split(',')]) for v in self._separate(arg_str)]
if member == 'split': if obj == str:
assert argvals == ('',) if member == 'fromCharCode':
return list(obj) assertion(argvals, 'takes one or more arguments')
if member == 'join': return ''.join(map(chr, argvals))
assert len(argvals) == 1 raise ExtractorError('Unsupported string method %s' % (member, ))
return argvals[0].join(obj)
if member == 'reverse': if member == 'split':
assert len(argvals) == 0 assertion(argvals, 'takes one or more arguments')
obj.reverse() assertion(argvals == [''], 'with arguments is not implemented')
return obj return list(obj)
if member == 'slice': elif member == 'join':
assert len(argvals) == 1 assertion(isinstance(obj, list), 'must be applied on a list')
return obj[argvals[0]:] assertion(len(argvals) == 1, 'takes exactly one argument')
if member == 'splice': return argvals[0].join(obj)
assert isinstance(obj, list) elif member == 'reverse':
index, howMany = argvals assertion(not argvals, 'does not take any arguments')
res = [] obj.reverse()
for i in range(index, min(index + howMany, len(obj))): return obj
res.append(obj.pop(index)) elif member == 'slice':
return res assertion(isinstance(obj, list), 'must be applied on a list')
assertion(len(argvals) == 1, 'takes exactly one argument')
return obj[member](argvals) return obj[argvals[0]:]
elif member == 'splice':
for op, opfunc in _OPERATORS: assertion(isinstance(obj, list), 'must be applied on a list')
m = re.match(r'(?P<x>.+?)%s(?P<y>.+)' % re.escape(op), expr) assertion(argvals, 'takes one or more arguments')
if not m: index, howMany = (argvals + [len(obj)])[:2]
continue if index < 0:
x, abort = self.interpret_statement( index += len(obj)
m.group('x'), local_vars, allow_recursion - 1) add_items = argvals[2:]
if abort: res = []
raise ExtractorError( for i in range(index, min(index + howMany, len(obj))):
'Premature left-side return of %s in %r' % (op, expr)) res.append(obj.pop(index))
y, abort = self.interpret_statement( for i, item in enumerate(add_items):
m.group('y'), local_vars, allow_recursion - 1) obj.insert(index + i, item)
if abort: return res
raise ExtractorError( elif member == 'unshift':
'Premature right-side return of %s in %r' % (op, expr)) assertion(isinstance(obj, list), 'must be applied on a list')
return opfunc(x, y) assertion(argvals, 'takes one or more arguments')
for item in reversed(argvals):
obj.insert(0, item)
return obj
elif member == 'pop':
assertion(isinstance(obj, list), 'must be applied on a list')
assertion(not argvals, 'does not take any arguments')
if not obj:
return
return obj.pop()
elif member == 'push':
assertion(argvals, 'takes one or more arguments')
obj.extend(argvals)
return obj
elif member == 'forEach':
assertion(argvals, 'takes one or more arguments')
assertion(len(argvals) <= 2, 'takes at-most 2 arguments')
f, this = (argvals + [''])[:2]
return [f((item, idx, obj), this=this) for idx, item in enumerate(obj)]
elif member == 'indexOf':
assertion(argvals, 'takes one or more arguments')
assertion(len(argvals) <= 2, 'takes at-most 2 arguments')
idx, start = (argvals + [0])[:2]
try:
return obj.index(idx, start)
except ValueError:
return -1
if isinstance(obj, list):
member = int(member)
nl.member = member
return obj[member](argvals)
if remaining:
return self.interpret_expression(
self._named_object(local_vars, eval_method()) + remaining,
local_vars, allow_recursion)
else:
return eval_method()
m = re.match( m = re.match(r'^(?P<func>%s)\((?P<args>[a-zA-Z0-9_$,]*)\)$' % _NAME_RE, expr)
r'^(?P<func>%s)\((?P<args>[a-zA-Z0-9_$,]*)\)$' % _NAME_RE, expr)
if m: if m:
fname = m.group('func') fname = m.group('func')
argvals = tuple([ argvals = tuple([
int(v) if v.isdigit() else local_vars[v] int(v) if v.isdigit() else local_vars[v]
for v in m.group('args').split(',')]) if len(m.group('args')) > 0 else tuple() for v in self._separate(m.group('args'))])
if fname not in self._functions: if fname in local_vars:
return local_vars[fname](argvals)
elif fname not in self._functions:
self._functions[fname] = self.extract_function(fname) self._functions[fname] = self.extract_function(fname)
return self._functions[fname](argvals) return self._functions[fname](argvals)
raise ExtractorError('Unsupported JS expression %r' % expr) if expr:
raise ExtractorError('Unsupported JS expression %r' % expr)
def extract_object(self, objname): def extract_object(self, objname):
_FUNC_NAME_RE = r'''(?:[a-zA-Z$0-9]+|"[a-zA-Z$0-9]+"|'[a-zA-Z$0-9]+')''' _FUNC_NAME_RE = r'''(?:[a-zA-Z$0-9]+|"[a-zA-Z$0-9]+"|'[a-zA-Z$0-9]+')'''
@ -233,30 +501,52 @@ class JSInterpreter(object):
return obj return obj
def extract_function(self, funcname): def extract_function_code(self, funcname):
""" @returns argnames, code """
func_m = re.search( func_m = re.search(
r'''(?x) r'''(?x)
(?:function\s+%s|[{;,]\s*%s\s*=\s*function|var\s+%s\s*=\s*function)\s* (?:function\s+%(f_n)s|[{;,]\s*%(f_n)s\s*=\s*function|var\s+%(f_n)s\s*=\s*function)\s*
\((?P<args>[^)]*)\)\s* \((?P<args>[^)]*)\)\s*
\{(?P<code>[^}]+)\}''' % ( (?P<code>\{(?:(?!};)[^"]|"([^"]|\\")*")+\})''' % {'f_n': re.escape(funcname), },
re.escape(funcname), re.escape(funcname), re.escape(funcname)),
self.code) self.code)
code, _ = self._separate_at_paren(func_m.group('code'), '}') # refine the match
if func_m is None: if func_m is None:
raise ExtractorError('Could not find JS function %r' % funcname) raise ExtractorError('Could not find JS function %r' % funcname)
argnames = func_m.group('args').split(',') return func_m.group('args').split(','), code
return self.build_function(argnames, func_m.group('code')) def extract_function(self, funcname):
return self.extract_function_from_code(*self.extract_function_code(funcname))
def extract_function_from_code(self, argnames, code, *global_stack):
local_vars = {}
while True:
mobj = re.search(r'function\((?P<args>[^)]*)\)\s*{', code)
if mobj is None:
break
start, body_start = mobj.span()
body, remaining = self._separate_at_paren(code[body_start - 1:], '}')
name = self._named_object(
local_vars,
self.extract_function_from_code(
[str.strip(x) for x in mobj.group('args').split(',')],
body, local_vars, *global_stack))
code = code[:start] + name + remaining
return self.build_function(argnames, code, local_vars, *global_stack)
def call_function(self, funcname, *args): def call_function(self, funcname, *args):
f = self.extract_function(funcname) return self.extract_function(funcname)(args)
return f(args)
def build_function(self, argnames, code, *global_stack):
def build_function(self, argnames, code): global_stack = list(global_stack) or [{}]
def resf(args): local_vars = global_stack.pop(0)
local_vars = dict(zip(argnames, args))
for stmt in code.split(';'): def resf(args, **kwargs):
res, abort = self.interpret_statement(stmt, local_vars) local_vars.update(dict(zip(argnames, args)))
if abort: local_vars.update(kwargs)
var_stack = LocalNameSpace(local_vars, *global_stack)
for stmt in self._separate(code.replace('\n', ''), ';'):
ret, should_abort = self.interpret_statement(stmt, var_stack)
if should_abort:
break break
return res return ret
return resf return resf

Loading…
Cancel
Save