yt-dlp `--exec` command injection when using `%q` in yt-dlp on Windows

@@ -142,5 +142,10 @@
         "when": "e3a3ed8a981d9395c4859b6ef56cd02bc3148db2",
         "short": "[cleanup:ie] No `from` stdlib imports in extractors",
         "authors": ["pukkandan"]
+    },
+    {
+        "action": "add",
+        "when": "9590cc6b4768e190183d7d071a6c78170889116a",
+        "short": "[priority] Security: [[CVE-2024-22423](https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-22423)] [Prevent RCE when using `--exec` with `%q` on Windows](https://github.com/yt-dlp/yt-dlp/security/advisories/GHSA-hjq6-52gw-2g7p)\n    - The shell escape function now properly escapes `%`, `\\` and `\\n`.\n    - `utils.Popen` has been patched accordingly."
     }
 ]

@@ -2069,6 +2069,10 @@ def run_shell(args):
 
         # Test escaping
         assert run_shell(['echo', 'test"&']) == '"test""&"\n'
+        assert run_shell(['echo', '%CMDCMDLINE:~-1%&']) == '"%CMDCMDLINE:~-1%&"\n'
+        assert run_shell(['echo', 'a\nb']) == '"a"\n"b"\n'
+        assert run_shell(['echo', '"']) == '""""\n'
+        assert run_shell(['echo', '\\']) == '\\\n'
         # Test if delayed expansion is disabled
         assert run_shell(['echo', '^!']) == '"^!"\n'
         assert run_shell('echo "^!"') == '"^!"\n'

@@ -27,12 +27,9 @@ def compat_etree_fromstring(text):
 compat_os_name = os._name if os.name == 'java' else os.name
 
 
-if compat_os_name == 'nt':
-    def compat_shlex_quote(s):
-        import re
-        return s if re.match(r'^[-_\w./]+$', s) else s.replace('"', '""').join('""')
-else:
-    from shlex import quote as compat_shlex_quote  # noqa: F401
+def compat_shlex_quote(s):
+    from ..utils import shell_quote
+    return shell_quote(s)
 
 
 def compat_ord(c):

@@ -50,7 +50,6 @@
     compat_expanduser,
     compat_HTMLParseError,
     compat_os_name,
-    compat_shlex_quote,
 )
 from ..dependencies import xattr
 
@@ -836,9 +835,11 @@ def __init__(self, args, *remaining, env=None, text=False, shell=False, **kwargs
 
         if shell and compat_os_name == 'nt' and kwargs.get('executable') is None:
             if not isinstance(args, str):
-                args = ' '.join(compat_shlex_quote(a) for a in args)
+                args = shell_quote(args, shell=True)
             shell = False
-            args = f'{self.__comspec()} /Q /S /D /V:OFF /C "{args}"'
+            # Set variable for `cmd.exe` newline escaping (see `utils.shell_quote`)
+            env['='] = '"^\n\n"'
+            args = f'{self.__comspec()} /Q /S /D /V:OFF /E:ON /C "{args}"'
 
         super().__init__(args, *remaining, env=env, shell=shell, **kwargs, startupinfo=self._startupinfo)
 
@@ -1637,15 +1638,38 @@ def get_filesystem_encoding():
     return encoding if encoding is not None else 'utf-8'
 
 
-def shell_quote(args):
-    quoted_args = []
-    encoding = get_filesystem_encoding()
-    for a in args:
-        if isinstance(a, bytes):
-            # We may get a filename encoded with 'encodeFilename'
-            a = a.decode(encoding)
-        quoted_args.append(compat_shlex_quote(a))
-    return ' '.join(quoted_args)
+_WINDOWS_QUOTE_TRANS = str.maketrans({'"': '\\"', '\\': '\\\\'})
+_CMD_QUOTE_TRANS = str.maketrans({
+    # Keep quotes balanced by replacing them with `""` instead of `\\"`
+    '"': '""',
+    # Requires a variable `=` containing `"^\n\n"` (set in `utils.Popen`)
+    # `=` should be unique since variables containing `=` cannot be set using cmd
+    '\n': '%=%',
+    # While we are only required to escape backslashes immediately before quotes,
+    # we instead escape all of 'em anyways to be consistent
+    '\\': '\\\\',
+    # Use zero length variable replacement so `%` doesn't get expanded
+    # `cd` is always set as long as extensions are enabled (`/E:ON` in `utils.Popen`)
+    '%': '%%cd:~,%',
+})
+
+
+def shell_quote(args, *, shell=False):
+    args = list(variadic(args))
+    if any(isinstance(item, bytes) for item in args):
+        deprecation_warning('Passing bytes to utils.shell_quote is deprecated')
+        encoding = get_filesystem_encoding()
+        for index, item in enumerate(args):
+            if isinstance(item, bytes):
+                args[index] = item.decode(encoding)
+
+    if compat_os_name != 'nt':
+        return shlex.join(args)
+
+    trans = _CMD_QUOTE_TRANS if shell else _WINDOWS_QUOTE_TRANS
+    return ' '.join(
+        s if re.fullmatch(r'[\w#$*\-+./:?@\\]+', s, re.ASCII) else s.translate(trans).join('""')
+        for s in args)
 
 
 def smuggle_url(url, data):
@@ -2849,7 +2873,7 @@ def ytdl_is_updateable():
 
 def args_to_str(args):
     # Get a short string representation for a subprocess command
-    return ' '.join(compat_shlex_quote(a) for a in args)
+    return shell_quote(args)
 
 
 def error_to_str(err):

@@ -25,7 +25,7 @@
 
 from .cache import Cache
 from .compat import functools, urllib  # isort: split
-from .compat import compat_os_name, compat_shlex_quote, urllib_req_to_req
+from .compat import compat_os_name, urllib_req_to_req
 from .cookies import LenientSimpleCookie, load_cookies
 from .downloader import FFmpegFD, get_suitable_downloader, shorten_protocol_name
 from .downloader.rtmp import rtmpdump_version
@@ -102,7 +102,6 @@
     UserNotLive,
     YoutubeDLError,
     age_restricted,
-    args_to_str,
     bug_reports_message,
     date_from_str,
     deprecation_warning,
@@ -141,6 +140,7 @@
     sanitize_filename,
     sanitize_path,
     sanitize_url,
+    shell_quote,
     str_or_none,
     strftime_or_none,
     subtitles_filename,
@@ -823,7 +823,7 @@ def warn_if_short_id(self, argv):
             self.report_warning(
                 'Long argument string detected. '
                 'Use -- to separate parameters and URLs, like this:\n%s' %
-                args_to_str(correct_argv))
+                shell_quote(correct_argv))
 
     def add_info_extractor(self, ie):
         """Add an InfoExtractor object to the end of the list."""
@@ -1355,7 +1355,7 @@ def create_key(outer_mobj):
                 value, fmt = escapeHTML(str(value)), str_fmt
             elif fmt[-1] == 'q':  # quoted
                 value = map(str, variadic(value) if '#' in flags else [value])
-                value, fmt = ' '.join(map(compat_shlex_quote, value)), str_fmt
+                value, fmt = shell_quote(value, shell=True), str_fmt
             elif fmt[-1] == 'B':  # bytes
                 value = f'%{str_fmt}'.encode() % str(value).encode()
                 value, fmt = value.decode('utf-8', 'ignore'), 's'

@@ -93,5 +93,10 @@
         "action": "add",
         "when": "c1d71d0d9f41db5e4306c86af232f5f6220a130b",
         "short": "[priority] **The minimum *recommended* Python version has been raised to 3.8**\nSince Python 3.7 has reached end-of-life, support for it will be dropped soon. [Read more](https://github.com/yt-dlp/yt-dlp/issues/7803)"
+    },
+    {
+        "action": "add",
+        "when": "61bdf15fc7400601c3da1aa7a43917310a5bf391",
+        "short": "[priority] Security: [[CVE-2023-40581](https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2023-40581)] [Prevent RCE when using `--exec` with `%q` on Windows](https://github.com/yt-dlp/yt-dlp/security/advisories/GHSA-42h4-v29r-42qg)\n    - The shell escape function is now using `\"\"` instead of `\\\"`.\n    - `utils.Popen` has been patched to properly quote commands."
     }
 ]

@@ -14,6 +14,7 @@
 import io
 import itertools
 import json
+import subprocess
 import xml.etree.ElementTree
 
 from yt_dlp.compat import (
@@ -28,6 +29,7 @@
     InAdvancePagedList,
     LazyList,
     OnDemandPagedList,
+    Popen,
     age_restricted,
     args_to_str,
     base_url,
@@ -2388,6 +2390,20 @@ def test_extract_basic_auth(self):
         assert extract_basic_auth('http://user:@foo.bar') == ('http://foo.bar', 'Basic dXNlcjo=')
         assert extract_basic_auth('http://user:pass@foo.bar') == ('http://foo.bar', 'Basic dXNlcjpwYXNz')
 
+    @unittest.skipUnless(compat_os_name == 'nt', 'Only relevant on Windows')
+    def test_Popen_windows_escaping(self):
+        def run_shell(args):
+            stdout, stderr, error = Popen.run(
+                args, text=True, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            assert not stderr
+            assert not error
+            return stdout
+
+        # Test escaping
+        assert run_shell(['echo', 'test"&']) == '"test""&"\n'
+        # Test if delayed expansion is disabled
+        assert run_shell(['echo', '^!']) == '"^!"\n'
+        assert run_shell('echo "^!"') == '"^!"\n'
 
 if __name__ == '__main__':
     unittest.main()

@@ -784,9 +784,9 @@ def expect_same_infodict(out):
         test('%(title4)#S', 'foo_bar_test')
         test('%(title4).10S', ('foo ＂bar＂ ', 'foo ＂bar＂' + ('#' if compat_os_name == 'nt' else ' ')))
         if compat_os_name == 'nt':
-            test('%(title4)q', ('"foo \\"bar\\" test"', "＂foo ⧹＂bar⧹＂ test＂"))
-            test('%(formats.:.id)#q', ('"id 1" "id 2" "id 3"', '＂id 1＂ ＂id 2＂ ＂id 3＂'))
-            test('%(formats.0.id)#q', ('"id 1"', '＂id 1＂'))
+            test('%(title4)q', ('"foo ""bar"" test"', None))
+            test('%(formats.:.id)#q', ('"id 1" "id 2" "id 3"', None))
+            test('%(formats.0.id)#q', ('"id 1"', None))
         else:
             test('%(title4)q', ('\'foo "bar" test\'', '\'foo ＂bar＂ test\''))
             test('%(formats.:.id)#q', "'id 1' 'id 2' 'id 3'")

@@ -30,7 +30,7 @@ def compat_etree_fromstring(text):
 if compat_os_name == 'nt':
     def compat_shlex_quote(s):
         import re
-        return s if re.match(r'^[-_\w./]+$', s) else '"%s"' % s.replace('"', '\\"')
+        return s if re.match(r'^[-_\w./]+$', s) else s.replace('"', '""').join('""')
 else:
     from shlex import quote as compat_shlex_quote  # noqa: F401
 

@@ -1,8 +1,6 @@
-import subprocess
-
 from .common import PostProcessor
 from ..compat import compat_shlex_quote
-from ..utils import PostProcessingError, encodeArgument, variadic
+from ..utils import Popen, PostProcessingError, variadic
 
 
 class ExecPP(PostProcessor):
@@ -27,10 +25,10 @@ def parse_cmd(self, cmd, info):
     def run(self, info):
         for tmpl in self.exec_cmd:
             cmd = self.parse_cmd(tmpl, info)
-            self.to_screen('Executing command: %s' % cmd)
-            retCode = subprocess.call(encodeArgument(cmd), shell=True)
-            if retCode != 0:
-                raise PostProcessingError('Command returned error code %d' % retCode)
+            self.to_screen(f'Executing command: {cmd}')
+            _, _, return_code = Popen.run(cmd, shell=True)
+            if return_code != 0:
+                raise PostProcessingError(f'Command returned error code {return_code}')
         return [], info
 
 

@@ -825,7 +825,7 @@ def _fix(key):
         _fix('LD_LIBRARY_PATH')  # Linux
         _fix('DYLD_LIBRARY_PATH')  # macOS
 
-    def __init__(self, *args, env=None, text=False, **kwargs):
+    def __init__(self, args, *remaining, env=None, text=False, shell=False, **kwargs):
         if env is None:
             env = os.environ.copy()
         self._fix_pyinstaller_ld_path(env)
@@ -835,7 +835,21 @@ def __init__(self, *args, env=None, text=False, **kwargs):
             kwargs['universal_newlines'] = True  # For 3.6 compatibility
             kwargs.setdefault('encoding', 'utf-8')
             kwargs.setdefault('errors', 'replace')
-        super().__init__(*args, env=env, **kwargs, startupinfo=self._startupinfo)
+
+        if shell and compat_os_name == 'nt' and kwargs.get('executable') is None:
+            if not isinstance(args, str):
+                args = ' '.join(compat_shlex_quote(a) for a in args)
+            shell = False
+            args = f'{self.__comspec()} /Q /S /D /V:OFF /C "{args}"'
+
+        super().__init__(args, *remaining, env=env, shell=shell, **kwargs, startupinfo=self._startupinfo)
+
+    def __comspec(self):
+        comspec = os.environ.get('ComSpec') or os.path.join(
+            os.environ.get('SystemRoot', ''), 'System32', 'cmd.exe')
+        if os.path.isabs(comspec):
+            return comspec
+        raise FileNotFoundError('shell not found: neither %ComSpec% nor %SystemRoot% is set')
 
     def communicate_or_kill(self, *args, **kwargs):
         try: