Repository: yt-dlp/yt-dlp
File: yt_dlp/downloader/external.py (Python)

Code before:
            cmd += ['--out', f'.{os.path.sep}{os.path.basename(tmpfilename)}']
        cmd += ['--auto-file-renaming=false']

        if 'fragments' in info_dict:
            cmd += ['--file-allocation=none', '--uri-selector=inorder']
            url_list_file = '%s.frag.urls' % tmpfilename
            url_list = []
            for frag_index, fragment in enumerate(info_dict['fragments']):
                fragment_filename = '%s-Frag%d' % (os.path.basename(tmpfilename), frag_index)
                url_list.append('%s\n\tout=%s' % (fragment['url'], fragment_filename))
            stream, _ = self.sanitize_open(url_list_file, 'wb')
            stream.write('\n'.join(url_list).encode())
            stream.close()
            cmd += ['-i', url_list_file]
        else:
            cmd += ['--', info_dict['url']]
        return cmd

    def _call_downloader(self, tmpfilename, info_dict):
        info_dict.pop('__rpc_port', None)
        info_dict.pop('__rpc_secret', None)
        if self._ENABLE_PROGRESS:
            import uuid
            info_dict = info_dict.copy()
            info_dict['__rpc_port'] = find_available_port() or 19190
            info_dict['__rpc_secret'] = str(uuid.uuid4())
        return super()._call_downloader(tmpfilename, info_dict)

    def _call_process(self, cmd, info_dict):
        if '__rpc_port' not in info_dict:
            return super()._call_process(cmd, info_dict)

        from tempfile import TemporaryFile
        import json
        import uuid

        rpc_port = info_dict['__rpc_port']
        rpc_secret = info_dict['__rpc_secret']
        nr_frags = len(info_dict['fragments']) if 'fragments' in info_dict else -1

        def aria2c_rpc(method, params, secret=True):
            if secret and rpc_secret:
                params = [f'token:{rpc_secret}', *params]
            # note: there's no need to be UUID (it can even a numeric value), but that's easier
            sanitycheck = str(uuid.uuid4())
            d = json.dumps({
                'jsonrpc': '2.0',
                'id': sanitycheck,
                'method': method,
                'params': params,
            }).encode('utf-8')

Review comment (security):
```suggestion
        info_dict['__rpc_port'] = find_available_port() or 19190
        info_dict['__rpc_secret'] = str(uuid.uuid4())
```
* why's _ENABLE_PROGRESS needed?
* why the pop and copy?
* Why import uuid locally?

Code after:
        # See: https://github.com/yt-dlp/yt-dlp/issues/276
        # https://github.com/ytdl-org/youtube-dl/issues/20312
        # https://github.com/aria2/aria2/issues/1373
        dn = os.path.dirname(tmpfilename)
        if dn:
            cmd += ['--dir', self._aria2c_filename(dn) + os.path.sep]
        if 'fragments' not in info_dict:
            cmd += ['--out', self._aria2c_filename(os.path.basename(tmpfilename))]
        cmd += ['--auto-file-renaming=false']

        if 'fragments' in info_dict:
            cmd += ['--file-allocation=none', '--uri-selector=inorder']
            url_list_file = '%s.frag.urls' % tmpfilename
            url_list = []
            for frag_index, fragment in enumerate(info_dict['fragments']):
                fragment_filename = '%s-Frag%d' % (os.path.basename(tmpfilename), frag_index)
                url_list.append('%s\n\tout=%s' % (fragment['url'], self._aria2c_filename(fragment_filename)))
            stream, _ = self.sanitize_open(url_list_file, 'wb')
            stream.write('\n'.join(url_list).encode())
            stream.close()
            cmd += ['-i', self._aria2c_filename(url_list_file)]
        else:
            cmd += ['--', info_dict['url']]
        return cmd

    def aria2c_rpc(self, rpc_port, rpc_secret, method, params=()):
        # Does not actually need to be UUID, just unique
        sanitycheck = str(uuid.uuid4())
        d = json.dumps({
            'jsonrpc': '2.0',
            'id': sanitycheck,
            'method': method,
            'params': [f'token:{rpc_secret}', *params],
        }).encode('utf-8')
        request = sanitized_Request(
            f'http://localhost:{rpc_port}/jsonrpc',
            data=d, headers={
                'Content-Type': 'application/json',
                'Content-Length': f'{len(d)}',
                'Ytdl-request-proxy': '__noproxy__',
            })
        with self.ydl.urlopen(request) as r:
            resp = json.load(r)
        assert resp.get('id') == sanitycheck, 'Something went wrong with RPC server'
        return resp['result']

    def _call_process(self, cmd, info_dict):
        if '__rpc' not in info_dict:
            return super()._call_process(cmd, info_dict)

        send_rpc = functools.partial(self.aria2c_rpc, info_dict['__rpc']['port'], info_dict['__rpc']['secret'])