yt_dlp/jsinterp.py

   1 from collections.abc import MutableMapping
   2 import json
   3 import operator
   4 import re
   5
   6 from .utils import (
   7     ExtractorError,
   8     remove_quotes,
   9 )
  10
  11 _OPERATORS = [
  12     ('|', operator.or_),
  13     ('^', operator.xor),
  14     ('&', operator.and_),
  15     ('>>', operator.rshift),
  16     ('<<', operator.lshift),
  17     ('-', operator.sub),
  18     ('+', operator.add),
  19     ('%', operator.mod),
  20     ('/', operator.truediv),
  21     ('*', operator.mul),
  22 ]
  23 _ASSIGN_OPERATORS = [(op + '=', opfunc) for op, opfunc in _OPERATORS]
  24 _ASSIGN_OPERATORS.append(('=', (lambda cur, right: right)))
  25
  26 _NAME_RE = r'[a-zA-Z_$][a-zA-Z_$0-9]*'
  27
  28
  29 class JS_Break(ExtractorError):
  30     def __init__(self):
  31         ExtractorError.__init__(self, 'Invalid break')
  32
  33
  34 class JS_Continue(ExtractorError):
  35     def __init__(self):
  36         ExtractorError.__init__(self, 'Invalid continue')
  37
  38
  39 class LocalNameSpace(MutableMapping):
  40     def __init__(self, *stack):
  41         self.stack = tuple(stack)
  42
  43     def __getitem__(self, key):
  44         for scope in self.stack:
  45             if key in scope:
  46                 return scope[key]
  47         raise KeyError(key)
  48
  49     def __setitem__(self, key, value):
  50         for scope in self.stack:
  51             if key in scope:
  52                 scope[key] = value
  53                 break
  54         else:
  55             self.stack[0][key] = value
  56         return value
  57
  58     def __delitem__(self, key):
  59         raise NotImplementedError('Deleting is not supported')
  60
  61     def __iter__(self):
  62         for scope in self.stack:
  63             yield from scope
  64
  65     def __len__(self, key):
  66         return len(iter(self))
  67
  68     def __repr__(self):
  69         return f'LocalNameSpace{self.stack}'
  70
  71
  72 class JSInterpreter(object):
  73     def __init__(self, code, objects=None):
  74         if objects is None:
  75             objects = {}
  76         self.code = code
  77         self._functions = {}
  78         self._objects = objects
  79         self.__named_object_counter = 0
  80
  81     def _named_object(self, namespace, obj):
  82         self.__named_object_counter += 1
  83         name = f'__yt_dlp_jsinterp_obj{self.__named_object_counter}'
  84         namespace[name] = obj
  85         return name
  86
  87     @staticmethod
  88     def _seperate(expr, delim=',', max_split=None):
  89         if not expr:
  90             return
  91         parens = {'(': 0, '{': 0, '[': 0, ']': 0, '}': 0, ')': 0}
  92         start, splits, pos, max_pos = 0, 0, 0, len(delim) - 1
  93         for idx, char in enumerate(expr):
  94             if char in parens:
  95                 parens[char] += 1
  96             is_in_parens = (parens['['] - parens[']']
  97                             or parens['('] - parens[')']
  98                             or parens['{'] - parens['}'])
  99             if char == delim[pos] and not is_in_parens:
 100                 if pos == max_pos:
 101                     pos = 0
 102                     yield expr[start: idx - max_pos]
 103                     start = idx + 1
 104                     splits += 1
 105                     if max_split and splits >= max_split:
 106                         break
 107                 else:
 108                     pos += 1
 109             else:
 110                 pos = 0
 111         yield expr[start:]
 112
 113     @staticmethod
 114     def _seperate_at_paren(expr, delim):
 115         seperated = list(JSInterpreter._seperate(expr, delim, 1))
 116         if len(seperated) < 2:
 117             raise ExtractorError(f'No terminating paren {delim} in {expr}')
 118         return seperated[0][1:].strip(), seperated[1].strip()
 119
 120     def interpret_statement(self, stmt, local_vars, allow_recursion=100):
 121         if allow_recursion < 0:
 122             raise ExtractorError('Recursion limit reached')
 123
 124         sub_statements = list(self._seperate(stmt, ';'))
 125         stmt = (sub_statements or ['']).pop()
 126         for sub_stmt in sub_statements:
 127             ret, should_abort = self.interpret_statement(sub_stmt, local_vars, allow_recursion - 1)
 128             if should_abort:
 129                 return ret
 130
 131         should_abort = False
 132         stmt = stmt.lstrip()
 133         stmt_m = re.match(r'var\s', stmt)
 134         if stmt_m:
 135             expr = stmt[len(stmt_m.group(0)):]
 136         else:
 137             return_m = re.match(r'return(?:\s+|$)', stmt)
 138             if return_m:
 139                 expr = stmt[len(return_m.group(0)):]
 140                 should_abort = True
 141             else:
 142                 # Try interpreting it as an expression
 143                 expr = stmt
 144
 145         v = self.interpret_expression(expr, local_vars, allow_recursion)
 146         return v, should_abort
 147
 148     def interpret_expression(self, expr, local_vars, allow_recursion):
 149         expr = expr.strip()
 150         if expr == '':  # Empty expression
 151             return None
 152
 153         if expr.startswith('{'):
 154             inner, outer = self._seperate_at_paren(expr, '}')
 155             inner, should_abort = self.interpret_statement(inner, local_vars, allow_recursion - 1)
 156             if not outer or should_abort:
 157                 return inner
 158             else:
 159                 expr = json.dumps(inner) + outer
 160
 161         if expr.startswith('('):
 162             inner, outer = self._seperate_at_paren(expr, ')')
 163             inner = self.interpret_expression(inner, local_vars, allow_recursion)
 164             if not outer:
 165                 return inner
 166             else:
 167                 expr = json.dumps(inner) + outer
 168
 169         if expr.startswith('['):
 170             inner, outer = self._seperate_at_paren(expr, ']')
 171             name = self._named_object(local_vars, [
 172                 self.interpret_expression(item, local_vars, allow_recursion)
 173                 for item in self._seperate(inner)])
 174             expr = name + outer
 175
 176         m = re.match(r'try\s*', expr)
 177         if m:
 178             if expr[m.end()] == '{':
 179                 try_expr, expr = self._seperate_at_paren(expr[m.end():], '}')
 180             else:
 181                 try_expr, expr = expr[m.end() - 1:], ''
 182             ret, should_abort = self.interpret_statement(try_expr, local_vars, allow_recursion - 1)
 183             if should_abort:
 184                 return ret
 185             return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
 186
 187         m = re.match(r'catch\s*\(', expr)
 188         if m:
 189             # We ignore the catch block
 190             _, expr = self._seperate_at_paren(expr, '}')
 191             return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
 192
 193         m = re.match(r'for\s*\(', expr)
 194         if m:
 195             constructor, remaining = self._seperate_at_paren(expr[m.end() - 1:], ')')
 196             if remaining.startswith('{'):
 197                 body, expr = self._seperate_at_paren(remaining, '}')
 198             else:
 199                 m = re.match(r'switch\s*\(', remaining)  # FIXME
 200                 if m:
 201                     switch_val, remaining = self._seperate_at_paren(remaining[m.end() - 1:], ')')
 202                     body, expr = self._seperate_at_paren(remaining, '}')
 203                     body = 'switch(%s){%s}' % (switch_val, body)
 204                 else:
 205                     body, expr = remaining, ''
 206             start, cndn, increment = self._seperate(constructor, ';')
 207             if self.interpret_statement(start, local_vars, allow_recursion - 1)[1]:
 208                 raise ExtractorError(
 209                     f'Premature return in the initialization of a for loop in {constructor!r}')
 210             while True:
 211                 if not self.interpret_expression(cndn, local_vars, allow_recursion):
 212                     break
 213                 try:
 214                     ret, should_abort = self.interpret_statement(body, local_vars, allow_recursion - 1)
 215                     if should_abort:
 216                         return ret
 217                 except JS_Break:
 218                     break
 219                 except JS_Continue:
 220                     pass
 221                 if self.interpret_statement(increment, local_vars, allow_recursion - 1)[1]:
 222                     raise ExtractorError(
 223                         f'Premature return in the initialization of a for loop in {constructor!r}')
 224             return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
 225
 226         m = re.match(r'switch\s*\(', expr)
 227         if m:
 228             switch_val, remaining = self._seperate_at_paren(expr[m.end() - 1:], ')')
 229             switch_val = self.interpret_expression(switch_val, local_vars, allow_recursion)
 230             body, expr = self._seperate_at_paren(remaining, '}')
 231             body, default = body.split('default:') if 'default:' in body else (body, None)
 232             items = body.split('case ')[1:]
 233             if default:
 234                 items.append(f'default:{default}')
 235             matched = False
 236             for item in items:
 237                 case, stmt = [i.strip() for i in self._seperate(item, ':', 1)]
 238                 matched = matched or case == 'default' or switch_val == self.interpret_expression(case, local_vars, allow_recursion)
 239                 if matched:
 240                     try:
 241                         ret, should_abort = self.interpret_statement(stmt, local_vars, allow_recursion - 1)
 242                         if should_abort:
 243                             return ret
 244                     except JS_Break:
 245                         break
 246             return self.interpret_statement(expr, local_vars, allow_recursion - 1)[0]
 247
 248         # Comma seperated statements
 249         sub_expressions = list(self._seperate(expr))
 250         expr = sub_expressions.pop().strip() if sub_expressions else ''
 251         for sub_expr in sub_expressions:
 252             self.interpret_expression(sub_expr, local_vars, allow_recursion)
 253
 254         for m in re.finditer(rf'''(?x)
 255                 (?P<pre_sign>\+\+|--)(?P<var1>{_NAME_RE})|
 256                 (?P<var2>{_NAME_RE})(?P<post_sign>\+\+|--)''', expr):
 257             var = m.group('var1') or m.group('var2')
 258             start, end = m.span()
 259             sign = m.group('pre_sign') or m.group('post_sign')
 260             ret = local_vars[var]
 261             local_vars[var] += 1 if sign[0] == '+' else -1
 262             if m.group('pre_sign'):
 263                 ret = local_vars[var]
 264             expr = expr[:start] + json.dumps(ret) + expr[end:]
 265
 266         for op, opfunc in _ASSIGN_OPERATORS:
 267             m = re.match(r'''(?x)
 268                 (?P<out>%s)(?:\[(?P<index>[^\]]+?)\])?
 269                 \s*%s
 270                 (?P<expr>.*)$''' % (_NAME_RE, re.escape(op)), expr)
 271             if not m:
 272                 continue
 273             right_val = self.interpret_expression(m.group('expr'), local_vars, allow_recursion)
 274
 275             if m.groupdict().get('index'):
 276                 lvar = local_vars[m.group('out')]
 277                 idx = self.interpret_expression(m.group('index'), local_vars, allow_recursion)
 278                 if not isinstance(idx, int):
 279                     raise ExtractorError(f'List indices must be integers: {idx}')
 280                 cur = lvar[idx]
 281                 val = opfunc(cur, right_val)
 282                 lvar[idx] = val
 283                 return val
 284             else:
 285                 cur = local_vars.get(m.group('out'))
 286                 val = opfunc(cur, right_val)
 287                 local_vars[m.group('out')] = val
 288                 return val
 289
 290         if expr.isdigit():
 291             return int(expr)
 292
 293         if expr == 'break':
 294             raise JS_Break()
 295         elif expr == 'continue':
 296             raise JS_Continue()
 297
 298         var_m = re.match(
 299             r'(?!if|return|true|false|null)(?P<name>%s)$' % _NAME_RE,
 300             expr)
 301         if var_m:
 302             return local_vars[var_m.group('name')]
 303
 304         try:
 305             return json.loads(expr)
 306         except ValueError:
 307             pass
 308
 309         m = re.match(
 310             r'(?P<in>%s)\[(?P<idx>.+)\]$' % _NAME_RE, expr)
 311         if m:
 312             val = local_vars[m.group('in')]
 313             idx = self.interpret_expression(m.group('idx'), local_vars, allow_recursion)
 314             return val[idx]
 315
 316         for op, opfunc in _OPERATORS:
 317             seperated = list(self._seperate(expr, op))
 318             if len(seperated) < 2:
 319                 continue
 320             right_val = seperated.pop()
 321             left_val = op.join(seperated)
 322             left_val, should_abort = self.interpret_statement(
 323                 left_val, local_vars, allow_recursion - 1)
 324             if should_abort:
 325                 raise ExtractorError(f'Premature left-side return of {op} in {expr!r}')
 326             right_val, should_abort = self.interpret_statement(
 327                 right_val, local_vars, allow_recursion - 1)
 328             if should_abort:
 329                 raise ExtractorError(f'Premature right-side return of {op} in {expr!r}')
 330             return opfunc(left_val or 0, right_val)
 331
 332         m = re.match(
 333             r'(?P<var>%s)(?:\.(?P<member>[^(]+)|\[(?P<member2>[^]]+)\])\s*' % _NAME_RE,
 334             expr)
 335         if m:
 336             variable = m.group('var')
 337             member = remove_quotes(m.group('member') or m.group('member2'))
 338             arg_str = expr[m.end():]
 339             if arg_str.startswith('('):
 340                 arg_str, remaining = self._seperate_at_paren(arg_str, ')')
 341             else:
 342                 arg_str, remaining = None, arg_str
 343
 344             def assertion(cndn, msg):
 345                 """ assert, but without risk of getting optimized out """
 346                 if not cndn:
 347                     raise ExtractorError(f'{member} {msg}: {expr}')
 348
 349             def eval_method():
 350                 nonlocal member
 351                 if variable == 'String':
 352                     obj = str
 353                 elif variable in local_vars:
 354                     obj = local_vars[variable]
 355                 else:
 356                     if variable not in self._objects:
 357                         self._objects[variable] = self.extract_object(variable)
 358                     obj = self._objects[variable]
 359
 360                 if arg_str is None:
 361                     # Member access
 362                     if member == 'length':
 363                         return len(obj)
 364                     return obj[member]
 365
 366                 # Function call
 367                 argvals = [
 368                     self.interpret_expression(v, local_vars, allow_recursion)
 369                     for v in self._seperate(arg_str)]
 370
 371                 if obj == str:
 372                     if member == 'fromCharCode':
 373                         assertion(argvals, 'takes one or more arguments')
 374                         return ''.join(map(chr, argvals))
 375                     raise ExtractorError(f'Unsupported string method {member}')
 376
 377                 if member == 'split':
 378                     assertion(argvals, 'takes one or more arguments')
 379                     assertion(argvals == [''], 'with arguments is not implemented')
 380                     return list(obj)
 381                 elif member == 'join':
 382                     assertion(isinstance(obj, list), 'must be applied on a list')
 383                     assertion(len(argvals) == 1, 'takes exactly one argument')
 384                     return argvals[0].join(obj)
 385                 elif member == 'reverse':
 386                     assertion(not argvals, 'does not take any arguments')
 387                     obj.reverse()
 388                     return obj
 389                 elif member == 'slice':
 390                     assertion(isinstance(obj, list), 'must be applied on a list')
 391                     assertion(len(argvals) == 1, 'takes exactly one argument')
 392                     return obj[argvals[0]:]
 393                 elif member == 'splice':
 394                     assertion(isinstance(obj, list), 'must be applied on a list')
 395                     assertion(argvals, 'takes one or more arguments')
 396                     index, howMany = (argvals + [len(obj)])[:2]
 397                     if index < 0:
 398                         index += len(obj)
 399                     add_items = argvals[2:]
 400                     res = []
 401                     for i in range(index, min(index + howMany, len(obj))):
 402                         res.append(obj.pop(index))
 403                     for i, item in enumerate(add_items):
 404                         obj.insert(index + i, item)
 405                     return res
 406                 elif member == 'unshift':
 407                     assertion(isinstance(obj, list), 'must be applied on a list')
 408                     assertion(argvals, 'takes one or more arguments')
 409                     for item in reversed(argvals):
 410                         obj.insert(0, item)
 411                     return obj
 412                 elif member == 'pop':
 413                     assertion(isinstance(obj, list), 'must be applied on a list')
 414                     assertion(not argvals, 'does not take any arguments')
 415                     if not obj:
 416                         return
 417                     return obj.pop()
 418                 elif member == 'push':
 419                     assertion(argvals, 'takes one or more arguments')
 420                     obj.extend(argvals)
 421                     return obj
 422                 elif member == 'forEach':
 423                     assertion(argvals, 'takes one or more arguments')
 424                     assertion(len(argvals) <= 2, 'takes at-most 2 arguments')
 425                     f, this = (argvals + [''])[:2]
 426                     return [f((item, idx, obj), this=this) for idx, item in enumerate(obj)]
 427                 elif member == 'indexOf':
 428                     assertion(argvals, 'takes one or more arguments')
 429                     assertion(len(argvals) <= 2, 'takes at-most 2 arguments')
 430                     idx, start = (argvals + [0])[:2]
 431                     try:
 432                         return obj.index(idx, start)
 433                     except ValueError:
 434                         return -1
 435
 436                 if isinstance(obj, list):
 437                     member = int(member)
 438                 return obj[member](argvals)
 439
 440             if remaining:
 441                 return self.interpret_expression(
 442                     self._named_object(local_vars, eval_method()) + remaining,
 443                     local_vars, allow_recursion)
 444             else:
 445                 return eval_method()
 446
 447         m = re.match(r'^(?P<func>%s)\((?P<args>[a-zA-Z0-9_$,]*)\)$' % _NAME_RE, expr)
 448         if m:
 449             fname = m.group('func')
 450             argvals = tuple([
 451                 int(v) if v.isdigit() else local_vars[v]
 452                 for v in self._seperate(m.group('args'))])
 453             if fname in local_vars:
 454                 return local_vars[fname](argvals)
 455             elif fname not in self._functions:
 456                 self._functions[fname] = self.extract_function(fname)
 457             return self._functions[fname](argvals)
 458
 459         if expr:
 460             raise ExtractorError('Unsupported JS expression %r' % expr)
 461
 462     def extract_object(self, objname):
 463         _FUNC_NAME_RE = r'''(?:[a-zA-Z$0-9]+|"[a-zA-Z$0-9]+"|'[a-zA-Z$0-9]+')'''
 464         obj = {}
 465         obj_m = re.search(
 466             r'''(?x)
 467                 (?<!this\.)%s\s*=\s*{\s*
 468                     (?P<fields>(%s\s*:\s*function\s*\(.*?\)\s*{.*?}(?:,\s*)?)*)
 469                 }\s*;
 470             ''' % (re.escape(objname), _FUNC_NAME_RE),
 471             self.code)
 472         fields = obj_m.group('fields')
 473         # Currently, it only supports function definitions
 474         fields_m = re.finditer(
 475             r'''(?x)
 476                 (?P<key>%s)\s*:\s*function\s*\((?P<args>[a-z,]+)\){(?P<code>[^}]+)}
 477             ''' % _FUNC_NAME_RE,
 478             fields)
 479         for f in fields_m:
 480             argnames = f.group('args').split(',')
 481             obj[remove_quotes(f.group('key'))] = self.build_function(argnames, f.group('code'))
 482
 483         return obj
 484
 485     def extract_function_code(self, funcname):
 486         """ @returns argnames, code """
 487         func_m = re.search(
 488             r'''(?x)
 489                 (?:function\s+%s|[{;,]\s*%s\s*=\s*function|var\s+%s\s*=\s*function)\s*
 490                 \((?P<args>[^)]*)\)\s*
 491                 (?P<code>\{(?:(?!};)[^"]|"([^"]|\\")*")+\})''' % (
 492                 re.escape(funcname), re.escape(funcname), re.escape(funcname)),
 493             self.code)
 494         code, _ = self._seperate_at_paren(func_m.group('code'), '}')  # refine the match
 495         if func_m is None:
 496             raise ExtractorError('Could not find JS function %r' % funcname)
 497         return func_m.group('args').split(','), code
 498
 499     def extract_function(self, funcname):
 500         return self.extract_function_from_code(*self.extract_function_code(funcname))
 501
 502     def extract_function_from_code(self, argnames, code, *global_stack):
 503         local_vars = {}
 504         while True:
 505             mobj = re.search(r'function\((?P<args>[^)]*)\)\s*{', code)
 506             if mobj is None:
 507                 break
 508             start, body_start = mobj.span()
 509             body, remaining = self._seperate_at_paren(code[body_start - 1:], '}')
 510             name = self._named_object(
 511                 local_vars,
 512                 self.extract_function_from_code(
 513                     [str.strip(x) for x in mobj.group('args').split(',')],
 514                     body, local_vars, *global_stack))
 515             code = code[:start] + name + remaining
 516         return self.build_function(argnames, code, local_vars, *global_stack)
 517
 518     def call_function(self, funcname, *args):
 519         return self.extract_function(funcname)(args)
 520
 521     def build_function(self, argnames, code, *global_stack):
 522         global_stack = list(global_stack) or [{}]
 523         local_vars = global_stack.pop(0)
 524
 525         def resf(args, **kwargs):
 526             local_vars.update({
 527                 **dict(zip(argnames, args)),
 528                 **kwargs
 529             })
 530             var_stack = LocalNameSpace(local_vars, *global_stack)
 531             for stmt in self._seperate(code.replace('\n', ''), ';'):
 532                 ret, should_abort = self.interpret_statement(stmt, var_stack)
 533                 if should_abort:
 534                     break
 535             return ret
 536         return resf