|
|
""" |
|
|
This module contains all the logic needed to find the signature functions. |
|
|
|
|
|
YouTube's strategy to restrict downloading videos is to send a ciphered version |
|
|
of the signature to the client, along with the decryption algorithm obfuscated |
|
|
in JavaScript. For the clients to play the videos, JavaScript must take the |
|
|
ciphered version, cycle it through a series of "transform functions," and then |
|
|
signs the media URL with the output. |
|
|
|
|
|
This module is responsible for (1) finding these "transformations |
|
|
functions" (2) sends them to be interpreted by jsinterp.py |
|
|
""" |
|
|
import logging |
|
|
import re |
|
|
|
|
|
from pytubefix.exceptions import RegexMatchError, InterpretationError |
|
|
from pytubefix.jsinterp import JSInterpreter, extract_player_js_global_var |
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
class Cipher: |
|
|
def __init__(self, js: str, js_url: str): |
|
|
|
|
|
self.js_url = js_url |
|
|
|
|
|
self.signature_function_name = get_initial_function_name(js, js_url) |
|
|
self.throttling_function_name = get_throttling_function_name(js, js_url) |
|
|
|
|
|
self.calculated_n = None |
|
|
|
|
|
self.js_interpreter = JSInterpreter(js) |
|
|
|
|
|
def get_throttling(self, n: str): |
|
|
"""Interpret the function that throttles download speed. |
|
|
:param str n: |
|
|
Contains the parameter that must be transformed. |
|
|
:rtype: str |
|
|
:returns: |
|
|
Returns the transformed value "n". |
|
|
""" |
|
|
try: |
|
|
return self.js_interpreter.call_function(self.throttling_function_name, n) |
|
|
except: |
|
|
raise InterpretationError(js_url=self.js_url) |
|
|
|
|
|
def get_signature(self, ciphered_signature: str) -> str: |
|
|
"""interprets the function that signs the streams. |
|
|
The lack of this signature generates the 403 forbidden error. |
|
|
:param str ciphered_signature: |
|
|
Contains the signature that must be transformed. |
|
|
:rtype: str |
|
|
:returns: |
|
|
Returns the correct stream signature. |
|
|
""" |
|
|
try: |
|
|
return self.js_interpreter.call_function(self.signature_function_name, ciphered_signature) |
|
|
except: |
|
|
raise InterpretationError(js_url=self.js_url) |
|
|
|
|
|
|
|
|
def get_initial_function_name(js: str, js_url: str) -> str: |
|
|
"""Extract the name of the function responsible for computing the signature. |
|
|
:param str js: |
|
|
The contents of the base.js asset file. |
|
|
:param str js_url: |
|
|
Full base.js url |
|
|
:rtype: str |
|
|
:returns: |
|
|
Function name from regex match |
|
|
""" |
|
|
|
|
|
function_patterns = [ |
|
|
r'(?P<sig>[a-zA-Z0-9_$]+)\s*=\s*function\(\s*(?P<arg>[a-zA-Z0-9_$]+)\s*\)\s*{\s*(?P=arg)\s*=\s*(?P=arg)\.split\(\s*[a-zA-Z0-9_\$\"\[\]]+\s*\)\s*;\s*[^}]+;\s*return\s+(?P=arg)\.join\(\s*[a-zA-Z0-9_\$\"\[\]]+\s*\)', |
|
|
r'(?:\b|[^a-zA-Z0-9_$])(?P<sig>[a-zA-Z0-9_$]{2,})\s*=\s*function\(\s*a\s*\)\s*{\s*a\s*=\s*a\.split\(\s*""\s*\)(?:;[a-zA-Z0-9_$]{2}\.[a-zA-Z0-9_$]{2}\(a,\d+\))?', |
|
|
r'\b(?P<var>[a-zA-Z0-9_$]+)&&\((?P=var)=(?P<sig>[a-zA-Z0-9_$]{2,})\(decodeURIComponent\((?P=var)\)\)', |
|
|
|
|
|
r'\b[cs]\s*&&\s*[adf]\.set\([^,]+\s*,\s*encodeURIComponent\s*\(\s*(?P<sig>[a-zA-Z0-9$]+)\(', |
|
|
r'\b[a-zA-Z0-9]+\s*&&\s*[a-zA-Z0-9]+\.set\([^,]+\s*,\s*encodeURIComponent\s*\(\s*(?P<sig>[a-zA-Z0-9$]+)\(', |
|
|
r'\bm=(?P<sig>[a-zA-Z0-9$]{2,})\(decodeURIComponent\(h\.s\)\)', |
|
|
|
|
|
r'("|\')signature\1\s*,\s*(?P<sig>[a-zA-Z0-9$]+)\(', |
|
|
r'\.sig\|\|(?P<sig>[a-zA-Z0-9$]+)\(', |
|
|
r'yt\.akamaized\.net/\)\s*\|\|\s*.*?\s*[cs]\s*&&\s*[adf]\.set\([^,]+\s*,\s*(?:encodeURIComponent\s*\()?\s*(?P<sig>[a-zA-Z0-9$]+)\(', |
|
|
r'\b[cs]\s*&&\s*[adf]\.set\([^,]+\s*,\s*(?P<sig>[a-zA-Z0-9$]+)\(', |
|
|
r'\bc\s*&&\s*[a-zA-Z0-9]+\.set\([^,]+\s*,\s*\([^)]*\)\s*\(\s*(?P<sig>[a-zA-Z0-9$]+)\(' |
|
|
] |
|
|
logger.debug("looking for signature cipher name") |
|
|
for pattern in function_patterns: |
|
|
regex = re.compile(pattern) |
|
|
function_match = regex.search(js) |
|
|
if function_match: |
|
|
sig = function_match.group('sig') |
|
|
logger.debug("finished regex search, matched: %s", pattern) |
|
|
logger.debug(f'Signature cipher function name: {sig}') |
|
|
return sig |
|
|
|
|
|
raise RegexMatchError( |
|
|
caller="get_initial_function_name", pattern=f"multiple in {js_url}" |
|
|
) |
|
|
|
|
|
|
|
|
def get_throttling_function_name(js: str, js_url: str) -> str: |
|
|
"""Extract the name of the function that computes the throttling parameter. |
|
|
|
|
|
:param str js: |
|
|
The contents of the base.js asset file. |
|
|
:param str js_url: |
|
|
Full base.js url |
|
|
:rtype: str |
|
|
:returns: |
|
|
The name of the function used to compute the throttling parameter. |
|
|
""" |
|
|
|
|
|
logger.debug("looking for nsig name") |
|
|
try: |
|
|
|
|
|
global_obj, varname, code = extract_player_js_global_var(js) |
|
|
if global_obj and varname and code: |
|
|
logger.debug(f"Global Obj name is: {varname}") |
|
|
global_obj = JSInterpreter(js).interpret_expression(code, {}, 100) |
|
|
logger.debug("Successfully interpreted global object") |
|
|
for k, v in enumerate(global_obj): |
|
|
if v.endswith('_w8_'): |
|
|
logger.debug(f"_w8_ found in index {k}") |
|
|
pattern = r'''(?xs) |
|
|
[;\n](?: |
|
|
(?P<f>function\s+)| |
|
|
(?:var\s+)? |
|
|
)(?P<funcname>[a-zA-Z0-9_$]+)\s*(?(f)|=\s*function\s*) |
|
|
\((?P<argname>[a-zA-Z0-9_$]+)\)\s*\{ |
|
|
(?:(?!\};(?![\]\)])).)+ |
|
|
\}\s*catch\(\s*[a-zA-Z0-9_$]+\s*\)\s* |
|
|
\{\s*return\s+%s\[%d\]\s*\+\s*(?P=argname)\s*\}\s*return\s+[^}]+\}[;\n] |
|
|
''' % (re.escape(varname), k) |
|
|
func_name = re.search(pattern, js) |
|
|
if func_name: |
|
|
n_func = func_name.group("funcname") |
|
|
logger.debug(f"Nfunc name is: {n_func}") |
|
|
return n_func |
|
|
except: |
|
|
pass |
|
|
|
|
|
pattern = r'''(?x) |
|
|
(?: |
|
|
\.get\("n"\)\)&&\(b=| |
|
|
(?: |
|
|
b=String\.fromCharCode\(110\)| |
|
|
(?P<str_idx>[a-zA-Z0-9_$.]+)&&\(b="nn"\[\+(?P=str_idx)\] |
|
|
) |
|
|
(?: |
|
|
,[a-zA-Z0-9_$]+\(a\))?,c=a\. |
|
|
(?: |
|
|
get\(b\)| |
|
|
[a-zA-Z0-9_$]+\[b\]\|\|null |
|
|
)\)&&\(c=| |
|
|
\b(?P<var>[a-zA-Z0-9_$]+)= |
|
|
)(?P<nfunc>[a-zA-Z0-9_$]+)(?:\[(?P<idx>\d+)\])?\([a-zA-Z]\) |
|
|
(?(var),[a-zA-Z0-9_$]+\.set\((?:"n+"|[a-zA-Z0-9_$]+)\,(?P=var)\))''' |
|
|
|
|
|
logger.debug('Finding throttling function name') |
|
|
|
|
|
regex = re.compile(pattern) |
|
|
function_match = regex.search(js) |
|
|
if function_match: |
|
|
logger.debug("finished regex search, matched: %s", pattern) |
|
|
|
|
|
func = function_match.group('nfunc') |
|
|
idx = function_match.group('idx') |
|
|
|
|
|
logger.debug(f'func is: {func}') |
|
|
logger.debug(f'idx is: {idx}') |
|
|
|
|
|
logger.debug('Checking throttling function name') |
|
|
if idx: |
|
|
n_func_check_pattern = fr'var {re.escape(func)}\s*=\s*\[(.+?)];' |
|
|
n_func_found = re.search(n_func_check_pattern, js) |
|
|
|
|
|
if n_func_found: |
|
|
throttling_function = n_func_found.group(1) |
|
|
logger.debug(f'Throttling function name is: {throttling_function}') |
|
|
return throttling_function |
|
|
|
|
|
raise RegexMatchError( |
|
|
caller="get_throttling_function_name", pattern=f"{n_func_check_pattern} in {js_url}" |
|
|
) |
|
|
|
|
|
raise RegexMatchError( |
|
|
caller="get_throttling_function_name", pattern=f"{pattern} in {js_url}" |
|
|
) |
|
|
|
