update

.config/yt-dlp/plugins/bgutil-ytdlp-pot-provider/yt_dlp_plugins/extractor/getpot_bgutil.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
-__version__ = '1.2.2'
+__version__ = '2.0.0'
 
 import abc
 import json
+import os
+from typing import TypeVar
 
 from yt_dlp.extractor.youtube.pot.provider import (
     ExternalRequestFeature,
@@ -15,6 +17,8 @@ from yt_dlp.extractor.youtube.pot.utils import WEBPO_CLIENTS
 from yt_dlp.utils import js_to_json
 from yt_dlp.utils.traversal import traverse_obj
 
+T = TypeVar('T')
+
 
 class BgUtilPTPBase(PoTokenProvider, abc.ABC):
     PROVIDER_VERSION = __version__
@@ -36,8 +40,6 @@ class BgUtilPTPBase(PoTokenProvider, abc.ABC):
         PoTokenContext.SUBS,
     )
     _GETPOT_TIMEOUT = 20.0
-    _GET_SERVER_VSN_TIMEOUT = 5.0
-    _MIN_NODE_VSN = (18, 0, 0)
 
     def _info_and_raise(self, msg, raise_from=None):
         self.logger.info(msg)
@@ -47,6 +49,25 @@ class BgUtilPTPBase(PoTokenProvider, abc.ABC):
         self.logger.warning(msg, once=once)
         raise PoTokenProviderRejectedRequest(msg) from raise_from
 
+    def _script_config_arg(self, key: str, default: T = None, *, casesense=True) -> str | T:
+        return self.ie._configuration_arg(
+            ie_key='youtubepot-bgutilscript', key=key, default=[default], casesense=casesense)[0]
+
+    @staticmethod
+    def _resolve_script_path(*ps: str):
+        # realpath resolves symlinks and internally calls abspath
+        return os.path.realpath(
+            os.path.expanduser(os.path.expandvars(os.path.join(*ps))))
+
+    def _script_path_provided(self) -> str | None:
+        if server_home := self._script_config_arg('server_home'):
+            return self._resolve_script_path(server_home)
+
+        if script_path := self._script_config_arg('script_path'):
+            return self._resolve_script_path(script_path, os.pardir, os.pardir)
+
+        return None
+
     def _check_version(self, got_version, *, default='unknown', name):
         def _major(version):
             return version.split('.', 1)[0]
@@ -58,34 +79,37 @@ class BgUtilPTPBase(PoTokenProvider, abc.ABC):
                 f'Please ensure they are on the same version. '
                 f'Otherwise, help will NOT be provided for any issues that arise. '
                 f'(plugin: {self.PROVIDER_VERSION}, {name}: {got_version or default})',
-                once=True,
-            )
+                once=True)
+
         if not got_version or _major(got_version) != _major(self.PROVIDER_VERSION):
             self._warn_and_raise(
                 f'Plugin and {name} major versions are mismatched. '
-                f'Update both the plugin and the {name} to the same version to proceed.'
-            )
+                f'Update both the plugin and the {name} to the same version to proceed.')
 
     def _get_attestation(self, webpage: str | None):
         if not webpage:
             return None
-        raw_challenge_data = self.ie._search_regex(
-            r"""(?sx)window\.ytAtR\s*=\s*(?P<raw_cd>(?P<q>['"])
-                (?:
-                    \\.|
-                    (?!(?P=q)).
-                )*
-            (?P=q))\s*;""",
-            webpage,
-            'raw challenge data',
-            default=None,
-            group='raw_cd',
-        )
-        att_txt = traverse_obj(raw_challenge_data, ({js_to_json}, {json.loads}, {json.loads}, 'bgChallenge'))
-        if not att_txt:
-            self.logger.warning('Failed to extract initial attestation from the webpage')
-            return None
-        return att_txt
+        raw_cd = (
+            traverse_obj(
+                self.ie._search_regex(
+                    r'''(?sx)window\s*\.\s*ytAtN\s*\(\s*
+                        (?P<js>\{.+?}\s*)
+                    \s*\)\s*;''', webpage, 'ytAtP challenge', default=None),
+                ({js_to_json}, {json.loads}, 'R'))
+            or traverse_obj(
+                self.ie._search_regex(
+                    r'''(?sx)window\.ytAtR\s*=\s*(?P<raw_cd>(?P<q>['"])
+                        (?:
+                            \\.|
+                            (?!(?P=q)).
+                        )*
+                    (?P=q))\s*;''', webpage, 'ytAtR challenge', default=None),
+                ({js_to_json}, {json.loads})))
+
+        if att_txt := traverse_obj(raw_cd, ({json.loads}, 'bgChallenge')):
+            return att_txt
+        self.logger.warning('Failed to extract initial attestation from the webpage')
+        return None
 
 
 __all__ = ['__version__']

.config/yt-dlp/plugins/bgutil-ytdlp-pot-provider/yt_dlp_plugins/extractor/getpot_bgutil_http.py

@@ -23,6 +23,7 @@ from yt_dlp_plugins.extractor.getpot_bgutil import BgUtilPTPBase
 class BgUtilHTTPPTP(BgUtilPTPBase):
     PROVIDER_NAME = 'bgutil:http'
     DEFAULT_BASE_URL = 'http://127.0.0.1:4416'
+    _GET_SERVER_VSN_TIMEOUT = 5.0
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -61,11 +62,8 @@ class BgUtilHTTPPTP(BgUtilPTPBase):
                 note=False))
         except TransportError as e:
             # the server may be down
-            script_path_provided = self.ie._configuration_arg(
-                ie_key='youtubepot-bgutilscript', key='script_path', default=[None])[0] is not None
-
             warning_base = f'Error reaching GET {self._base_url}/ping (caused by {e.__class__.__name__}). '
-            if script_path_provided:  # server down is expected, log info
+            if self._script_path_provided() is not None:  # server down is expected, log info
                 self._info_and_raise(
                     warning_base + 'This is expected if you are using the script method.')
             else:
@@ -108,18 +106,18 @@ class BgUtilHTTPPTP(BgUtilPTPBase):
         # used for CI check
         self.logger.trace('Generating POT via HTTP server')
 
-        disable_innertube = bool(self._configuration_arg('disable_innertube', default=[None])[0])
-        challenge = self._get_attestation(None if disable_innertube else request.video_webpage)
+        if self._configuration_arg('disable_innertube', default=[None])[0] is not None:
+            self._warn_and_raise(
+                "'youtubepot-bgutilhttp:disable_innertube' extractor arg is deprecated")
+
+        challenge = self._get_attestation(request.video_webpage)
         # The challenge is falsy when the webpage and the challenge are unavailable
         # In this case, we need to disable /att/get since it's broken for web_music
         if not challenge and request.internal_client_name == 'web_music':
-            if not disable_innertube:  # if not already set, warn the user
-                self.logger.warning(
-                    'BotGuard challenges could not be obtained from the webpage, '
-                    'overriding disable_innertube=True because InnerTube challenges '
-                    'are currently broken for the web_music client. '
-                    'Pass disable_innertube=1 to suppress this warning.')
-            disable_innertube = True
+            self._warn_and_raise(
+                'BotGuard challenges could not be obtained from the webpage, '
+                'a PO Token cannot be generated because InnerTube challenges '
+                'are currently broken for the web_music client. ')
 
         try:
             response = self._request_webpage(
@@ -128,7 +126,6 @@ class BgUtilHTTPPTP(BgUtilPTPBase):
                         'bypass_cache': request.bypass_cache,
                         'challenge': challenge,
                         'content_binding': get_webpo_content_binding(request)[0],
-                        'disable_innertube': disable_innertube,
                         'disable_tls_verification': not request.request_verify_tls,
                         'proxy': request.request_proxy,
                         'innertube_context': request.innertube_context,

.config/yt-dlp/plugins/bgutil-ytdlp-pot-provider/yt_dlp_plugins/extractor/getpot_bgutil_script.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import abc
 import functools
 import json
-import os.path
+import os
 import re
-import shutil
 import subprocess
+import sys
+import sysconfig
+from typing import Iterable
 
 from yt_dlp.extractor.youtube.pot.provider import (
     PoTokenProviderError,
@@ -15,106 +18,201 @@ from yt_dlp.extractor.youtube.pot.provider import (
     register_provider,
 )
 from yt_dlp.extractor.youtube.pot.utils import get_webpo_content_binding
-from yt_dlp.utils import Popen
+from yt_dlp.utils import Popen, int_or_none, shell_quote
+from yt_dlp.utils.traversal import traverse_obj
 
 from yt_dlp_plugins.extractor.getpot_bgutil import BgUtilPTPBase
 
+_FALLBACK_PATHEXT = ('.COM', '.EXE', '.BAT', '.CMD')
 
-@register_provider
-class BgUtilScriptPTP(BgUtilPTPBase):
-    PROVIDER_NAME = 'bgutil:script'
+
+# Copied from https://github.com/yt-dlp/yt-dlp/blob/d9813a3da6959662841dfb34cad0ee6c07a65d1e/yt_dlp/utils/_jsruntime.py#L16-L64
+# NOTE: keep in sync with upstream
+def _find_exe(basename: str) -> str:
+    # Check in Python "scripts" path, e.g. for pipx-installed binaries
+    binary = os.path.join(
+        sysconfig.get_path('scripts'),
+        basename + sysconfig.get_config_var('EXE'))
+    if os.access(binary, os.F_OK | os.X_OK) and not os.path.isdir(binary):
+        return binary
+
+    if os.name != 'nt':
+        return basename
+
+    paths: list[str] = []
+
+    # binary dir
+    if getattr(sys, 'frozen', False):
+        paths.append(os.path.dirname(sys.executable))
+    # cwd
+    paths.append(os.getcwd())
+    # PATH items
+    if path := os.environ.get('PATH'):
+        paths.extend(filter(None, path.split(os.path.pathsep)))
+
+    pathext = os.environ.get('PATHEXT')
+    if pathext is None:
+        exts = _FALLBACK_PATHEXT
+    else:
+        exts = tuple(ext for ext in pathext.split(os.pathsep) if ext)
+
+    visited = set()
+    for path in map(os.path.realpath, paths):
+        normed = os.path.normcase(path)
+        if normed in visited:
+            continue
+        visited.add(normed)
+
+        for ext in exts:
+            binary = os.path.join(path, f'{basename}{ext}')
+            if os.access(binary, os.F_OK | os.X_OK) and not os.path.isdir(binary):
+                return binary
+
+    return basename
+
+
+def _determine_runtime_path(path, basename):
+    if not path:
+        return _find_exe(basename)
+    if os.path.isdir(path):
+        return os.path.join(path, basename)
+    return path
+
+
+class BgUtilScriptPTPBase(BgUtilPTPBase, abc.ABC):
+    _GET_SCRIPT_VSN_TIMEOUT = 15.0
+
+    @staticmethod
+    def _jsrt_vsn_tup(v: str):
+        return tuple(int_or_none(x, default=0) for x in v.split('.'))
+
+    def __init_subclass__(cls):
+        super().__init_subclass__()
+        pref = cls._JSRT_PREF
+        register_preference(cls)(lambda provider, request: pref)
+
+    _SCRIPT_BASENAME: str
+    _JSRT_NAME: str  # Name of the JS Runtime shown in logs
+    _JSRT_EXEC: str  # Name of the executable, and the name used in yt-dlp
+    _JSRT_VSN_REGEX: str
+    _JSRT_MIN_VER: tuple[int, ...]
+    _JSRT_PREF: int
+
+    @abc.abstractmethod
+    def _script_path_impl(self) -> str:
+        raise NotImplementedError
+
+    def _jsrt_args(self) -> Iterable[str]:
+        return ()
+
+    def _jsrt_envs(self) -> dict:
+        return os.environ.copy()
+
+    def _jsrt_path_impl(self) -> str | None:
+        jsrt_path = _determine_runtime_path(
+            traverse_obj(self.ie.get_param('js_runtimes'), (self._JSRT_EXEC, 'path')),
+            self._JSRT_EXEC)
+        try:
+            output, _, returncode = Popen.run(
+                [jsrt_path, '--version'], env=self._jsrt_envs(), timeout=5.0,
+                text=True, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            output = output.strip()
+        except subprocess.TimeoutExpired:
+            self.logger.debug(
+                f'Failed to check {self._JSRT_NAME} version: {self._JSRT_NAME} process '
+                'did not finish in 5.0 seconds', once=True)
+            return None
+        except FileNotFoundError:
+            self.logger.debug(
+                f'{self._JSRT_NAME} executable not found. Please ensure {self._JSRT_NAME} is '
+                'installed and available in PATH or passed to yt-dlp with --js-runtimes.', once=True)
+            return None
+        mobj = re.search(self._JSRT_VSN_REGEX, output)
+        if returncode or not mobj:
+            self.logger.debug(
+                f'Failed to check {self._JSRT_NAME} version. '
+                f'{self._JSRT_NAME} returned {returncode} exit status. '
+                f'Process output:\n{output}', once=True)
+            return None
+        if self._jsrt_has_support(mobj.group(1)):
+            return jsrt_path
+
+    def _jsrt_has_support(self, v: str) -> bool:
+        if self._jsrt_vsn_tup(v) >= self._JSRT_MIN_VER:
+            self.logger.trace(f'{self._JSRT_NAME} version: {v}')
+            return True
+        else:
+            min_vsn_str = '.'.join(map(str, self._JSRT_MIN_VER))
+            self.logger.debug(
+                f'{self._JSRT_NAME} version too low. '
+                f'(got {v}, but at least {min_vsn_str} is required)', once=True)
+            return False
+
+    @functools.cached_property
+    def _script_path(self) -> str:
+        return self._script_path_impl()
+
+    @functools.cached_property
+    def _jsrt_path(self) -> str | None:
+        return self._jsrt_path_impl()
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._check_script = functools.cache(self._check_script_impl)
 
     @functools.cached_property
-    def _script_path(self):
-        script_path = self._configuration_arg(
-            'script_path', casesense=True, default=[None])[0]
-
-        if script_path:
-            return os.path.expandvars(script_path)
-
-        # check deprecated arg
-        deprecated_script_path = self.ie._configuration_arg(
-            ie_key='youtube', key='getpot_bgutil_script', default=[None])[0]
-
-        if deprecated_script_path:
-            self._warn_and_raise(
-                "'youtube:getpot_bgutil_script' extractor arg is deprecated, use 'youtubepot-bgutilscript:script_path' instead")
+    def _server_home(self) -> str:
+        if path := self._script_path_provided():
+            return path
 
         # default if no arg was passed
-        home = os.path.expanduser('~')
-        default_path = os.path.join(
-            home, 'bgutil-ytdlp-pot-provider', 'server', 'build', 'generate_once.js')
+        default_home = self._resolve_script_path('~', 'bgutil-ytdlp-pot-provider', 'server')
         self.logger.debug(
-            f'No script path passed, defaulting to {default_path}')
-        return default_path
-
-    def is_available(self):
-        return self._check_script(self._script_path)
+            f'No server_home or script_path passed, defaulting to {default_home}', once=True)
+        return default_home
 
     @functools.cached_property
-    def _node_path(self):
-        node_path = shutil.which('node')
-        if node_path is None:
-            self.logger.trace('node is not in PATH')
-        vsn = self._check_node_version(node_path)
-        if vsn:
-            self.logger.trace(f'Node version: {vsn}')
-            return node_path
+    def _script_cache_dir(self) -> str:
+        # don't use _HOMEDIR as the server is coded this way and accepts HOME and USERPROFILE regardless of the OS
+        home_dir = os.getenv('HOME') or os.getenv('USERPROFILE')
+        if (xdg_cache := os.getenv('XDG_CACHE_HOME')) is not None:
+            return os.path.abspath(os.path.join(xdg_cache, 'bgutil-ytdlp-pot-provider'))
+        elif home_dir:
+            return os.path.abspath(os.path.join(home_dir, '.cache', 'bgutil-ytdlp-pot-provider'))
+        else:
+            return self._server_home
 
-    def _check_script_impl(self, script_path):
+    def is_available(self) -> bool:
+        return self._check_script(self._script_path)
+
+    def _check_script_impl(self, script_path) -> bool:
         if not os.path.isfile(script_path):
             self.logger.debug(
-                f"Script path doesn't exist: {script_path}")
+                f"Script path doesn't exist: {script_path}", once=True)
             return False
-        if os.path.basename(script_path) != 'generate_once.js':
+        if os.path.basename(script_path) != self._SCRIPT_BASENAME:
             self.logger.warning(
-                'Incorrect script passed to extractor args. Path to generate_once.js required', once=True)
+                f'The script path passed in the extractor argument '
+                f'has a wrong base name, expected {self._SCRIPT_BASENAME}.', once=True)
             return False
-        node_path = self._node_path
-        if not node_path:
+        if not self._jsrt_path:
             return False
-        stdout, stderr, returncode = Popen.run(
-            [self._node_path, script_path, '--version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-            timeout=self._GET_SERVER_VSN_TIMEOUT)
+        stdout, _, returncode = Popen.run(
+            [self._jsrt_path, *self._jsrt_args(), script_path, '--version'],
+            env=self._jsrt_envs(), timeout=self._GET_SCRIPT_VSN_TIMEOUT,
+            text=True, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE)
+        stdout = stdout.strip()
         if returncode:
             self.logger.warning(
                 f'Failed to check script version. '
                 f'Script returned {returncode} exit status. '
-                f'Script stdout: {stdout}; Script stderr: {stderr}',
+                f'Script stdout:\n{stdout}',
                 once=True)
             return False
         else:
-            self._check_version(stdout.strip(), name='script')
+            self._check_version(stdout, name='script')
             return True
 
-    def _check_node_version(self, node_path):
-        try:
-            stdout, stderr, returncode = Popen.run(
-                [node_path, '--version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-                timeout=self._GET_SERVER_VSN_TIMEOUT)
-            stdout = stdout.strip()
-            mobj = re.match(r'v(\d+)\.(\d+)\.(\d+)', stdout)
-            if returncode or not mobj:
-                raise ValueError
-            node_vsn = tuple(map(int, mobj.groups()))
-            if node_vsn >= self._MIN_NODE_VSN:
-                return node_vsn
-            raise RuntimeError
-        except RuntimeError:
-            min_vsn_str = 'v' + '.'.join(str(v) for v in self._MIN_NODE_VSN)
-            self.logger.warning(
-                f'Node version too low. '
-                f'(got {stdout}, but at least {min_vsn_str} is required)')
-        except (subprocess.TimeoutExpired, ValueError):
-            self.logger.warning(
-                f'Failed to check node version. '
-                f'Node returned {returncode} exit status. '
-                f'Node stdout: {stdout}; Node stderr: {stderr}')
-
     def _real_request_pot(
         self,
         request: PoTokenRequest,
@@ -123,10 +221,11 @@ class BgUtilScriptPTP(BgUtilPTPBase):
         self.logger.trace(
             f'Generating POT via script: {self._script_path}')
 
-        command_args = [self._node_path, self._script_path]
+        command_args = [self._jsrt_path, *self._jsrt_args(), self._script_path]
         if proxy := request.request_proxy:
             command_args.extend(['-p', proxy])
         command_args.extend(['-c', get_webpo_content_binding(request)[0]])
+        command_args.extend(['--innertube-context', json.dumps(request.innertube_context)])
         if request.bypass_cache:
             command_args.append('--bypass-cache')
         if request.request_source_address:
@@ -140,12 +239,14 @@ class BgUtilScriptPTP(BgUtilPTPBase):
             f'{request.internal_client_name} client via bgutil script',
         )
         self.logger.debug(
-            f'Executing command to get POT via script: {" ".join(command_args)}')
+            f'Executing command to get POT via script: {" ".join(map(shell_quote, command_args))}')
 
         try:
-            stdout, stderr, returncode = Popen.run(
-                command_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-                timeout=self._GETPOT_TIMEOUT)
+            stdout, _, returncode = Popen.run(
+                command_args, env=self._jsrt_envs(), timeout=self._GETPOT_TIMEOUT,
+                text=True, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE)
+            stdout_lines = stdout.strip().splitlines()
+            json_resp = stdout_lines.pop()
         except subprocess.TimeoutExpired as e:
             raise PoTokenProviderError(
                 f'_get_pot_via_script failed: Timeout expired when trying to run script (caused by {e!r})')
@@ -153,20 +254,13 @@ class BgUtilScriptPTP(BgUtilPTPBase):
             raise PoTokenProviderError(
                 f'_get_pot_via_script failed: Unable to run script (caused by {e!r})') from e
 
-        msg = ''
-        if stdout_extra := stdout.strip().splitlines()[:-1]:
-            msg = f'stdout:\n{stdout_extra}\n'
-        if stderr_stripped := stderr.strip():  # Empty strings are falsy
-            msg += f'stderr:\n{stderr_stripped}\n'
-        msg = msg.strip()
-        if msg:
-            self.logger.trace(msg)
+        if stdout_extra := '\n'.join(stdout_lines):
+            self.logger.debug(f'script stdout:\n{stdout_extra}')
         if returncode:
             raise PoTokenProviderError(
                 f'_get_pot_via_script failed with returncode {returncode}')
 
         try:
-            json_resp = stdout.splitlines()[-1]
             self.logger.trace(f'JSON response:\n{json_resp}')
             # The JSON response is always the last line
             script_data_resp = json.loads(json_resp)
@@ -179,10 +273,55 @@ class BgUtilScriptPTP(BgUtilPTPBase):
         return PoTokenResponse(po_token=script_data_resp['poToken'])
 
 
-@register_preference(BgUtilScriptPTP)
-def bgutil_script_getpot_preference(provider, request):
-    return 1
+@register_provider
+class BgUtilScriptNodePTP(BgUtilScriptPTPBase):
+    PROVIDER_NAME = 'bgutil:script-node'
+    _SCRIPT_BASENAME = 'generate_once.js'
+    _JSRT_NAME = 'Node.js'
+    _JSRT_EXEC = 'node'
+    _JSRT_VSN_REGEX = r'^v(\S+)'
+    _JSRT_MIN_VER = (22, 0, 0)
+    _JSRT_PREF = 10
+
+    def _script_path_impl(self) -> str:
+        return os.path.join(
+            self._server_home, 'build', self._SCRIPT_BASENAME)
 
 
-__all__ = [BgUtilScriptPTP.__name__,
-           bgutil_script_getpot_preference.__name__]
+@register_provider
+class BgUtilScriptDenoPTP(BgUtilScriptPTPBase):
+    PROVIDER_NAME = 'bgutil:script-deno'
+    _SCRIPT_BASENAME = 'generate_once.ts'
+    _JSRT_NAME = 'Deno'
+    _JSRT_EXEC = 'deno'
+    _JSRT_VSN_REGEX = r'^deno (\S+)'
+    _JSRT_MIN_VER = (2, 4, 3)
+    _JSRT_PREF = 20
+
+    def _script_path_impl(self) -> str:
+        return os.path.join(
+            self._server_home, 'src', self._SCRIPT_BASENAME)
+
+    def _jsrt_args(self) -> Iterable[str]:
+        def escpath(*strs: str):
+            return ','.join(s.replace(',', ',,') for s in strs)
+        node_mods_path = os.path.join(self._server_home, 'node_modules')
+        return (
+            'run', '--allow-env', '--allow-net',
+            f'--allow-ffi={escpath(node_mods_path)}',
+            f'--allow-write={escpath(self._script_cache_dir)}',
+            f'--allow-read={escpath(self._script_cache_dir, node_mods_path)}',
+        )
+
+    def _jsrt_envs(self) -> dict:
+        process_env = os.environ.copy()
+        process_env['DENO_NO_PROMPT'] = '1'
+        process_env['DENO_NO_UPDATE_CHECK'] = '1'
+        process_env['FORCE_COLOR'] = 'false'
+        return process_env
+
+
+__all__ = [
+    BgUtilScriptNodePTP.__name__,
+    BgUtilScriptDenoPTP.__name__,
+]