Repository: yt-dlp/yt-dlp

Issue:
[Dropbox] Unable to extract mountComponent
### DO NOT REMOVE OR SKIP THE ISSUE TEMPLATE

- [X] I understand that I will be **blocked** if I *intentionally* remove or skip any mandatory\* field

### Checklist

- [X] I'm reporting that a **supported** site is broken
- [X] I've verified that I'm running yt-dlp version **2023.03.04** ([update instructions](https://github.com/yt-dlp/yt-dlp#update)) or later (specify commit)
- [X] I've checked that all provided URLs are playable in a browser with the same IP and same login details
- [X] I've checked that all URLs and arguments with special characters are [properly quoted or escaped](https://github.com/yt-dlp/yt-dlp/wiki/FAQ#video-url-contains-an-ampersand--and-im-getting-some-strange-output-1-2839-or-v-is-not-recognized-as-an-internal-or-external-command)
- [X] I've searched [known issues](https://github.com/yt-dlp/yt-dlp/issues/3766) and the [bugtracker](https://github.com/yt-dlp/yt-dlp/issues?q=) for similar issues **including closed ones**. DO NOT post duplicates
- [X] I've read the [guidelines for opening an issue](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#opening-an-issue)
- [ ] I've read about [sharing account credentials](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#are-you-willing-to-share-account-details-if-needed) and I'm willing to share it if required

### Region

_No response_

### Provide a description that is worded well enough to be understood

hello,
cannot extract from dropbox.com. demo link [NSFW]: REDACTED


### Provide verbose output that clearly demonstrates the problem

- [X] Run **your** yt-dlp command with **-vU** flag added (`yt-dlp -vU <your command line>`)
- [ ] If using API, add `'verbose': True` to `YoutubeDL` params instead
- [X] Copy the WHOLE output (starting with `[debug] Command-line config`) and insert it below

### Complete Verbose Output

```shell
C:\Users\MASTERX>youtube-dl "REDACTED" -FUv
[debug] Command-line config: ['REDACTED', '-FUv']
[debug] Encodings: locale cp1252, fs utf-8, pref cp1252, out utf-8, error utf-8, screen utf-8
[debug] yt-dlp version stable@2023.03.04 [392389b7d] (win_exe)
[debug] Python 3.8.10 (CPython AMD64 64bit) - Windows-10-10.0.19045-SP0 (OpenSSL 1.1.1k  25 Mar 2021)
[debug] exe versions: ffmpeg N-110072-g073ec3b9da-20230325 (setts), ffprobe N-110072-g073ec3b9da-20230325
[debug] Optional libraries: Cryptodome-3.17, brotli-1.0.9, certifi-2022.12.07, mutagen-1.46.0, sqlite3-2.6.0, websockets-10.4
[debug] Proxy map: {}
[debug] Loaded 1786 extractors
[debug] Fetching release info: https://api.github.com/repos/yt-dlp/yt-dlp/releases/latest
Available version: stable@2023.03.04, Current version: stable@2023.03.04
Current Build Hash: 5590c57bd0433ed239a2deaaf92e2ad6f37fe50f53664c821575cafe106a9421
yt-dlp is up to date (stable@2023.03.04)
[Dropbox] Extracting URL: REDACTED
[Dropbox] q9n7yl55jdub33o: Downloading webpage
ERROR: [Dropbox] q9n7yl55jdub33o: Unable to extract mountComponent; please report this issue on  https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using  yt-dlp -U
  File "yt_dlp\extractor\common.py", line 694, in extract
  File "yt_dlp\extractor\dropbox.py", line 56, in _real_extract
  File "yt_dlp\extractor\common.py", line 1256, in _search_json
  File "yt_dlp\extractor\common.py", line 1242, in _search_regex
```


Discussion:
bashonly: Add a test for each of the new URL formats
DoubiTe: @bashonly & @nathantouze this PR is partially fix the extractor because it still generate an error "mountComponent" #7005
Is it possible to fix-it @pukkandan ? Thanks a lot

Patch:
diff --git a/CONTRIBUTORS b/CONTRIBUTORS
index 6ccd08931..6b9b9f470 100644
--- a/CONTRIBUTORS
+++ b/CONTRIBUTORS
@@ -2,7 +2,6 @@ pukkandan (owner)
 shirt-dev (collaborator)
 coletdjnz/colethedj (collaborator)
 Ashish0804 (collaborator)
-nao20010128nao/Lesmiscore (collaborator)
 bashonly (collaborator)
 Grub4K (collaborator)
 h-h-h-h
@@ -467,3 +466,4 @@ nnoboa
 rdamas
 RfadnjdExt
 urectanc
+nao20010128nao/Lesmiscore
diff --git a/Collaborators.md b/Collaborators.md
index a0976dd8c..70ab616f1 100644
--- a/Collaborators.md
+++ b/Collaborators.md
@@ -44,16 +44,6 @@ ## [Ashish0804](https://github.com/Ashish0804) <sub><sup>[Inactive]</sup></sub>
 * Improved/fixed support for HiDive, HotStar, Hungama, LBRY, LinkedInLearning, Mxplayer, SonyLiv, TV2, Vimeo, VLive etc
 
 
-## [Lesmiscore](https://github.com/Lesmiscore)
-
-**Bitcoin**: bc1qfd02r007cutfdjwjmyy9w23rjvtls6ncve7r3s  
-**Monacoin**: mona1q3tf7dzvshrhfe3md379xtvt2n22duhglv5dskr
-
-* Download live from start to end for YouTube
-* Added support for new websites AbemaTV, mildom, PixivSketch, skeb, radiko, voicy, mirrativ, openrec, whowatch, damtomo, 17.live, mixch etc
-* Improved/fixed support for fc2, YahooJapanNews, tver, iwara etc
-
-
 ## [bashonly](https://github.com/bashonly)
 
 * `--update-to`, automated release, nightly builds
diff --git a/yt_dlp/extractor/dropbox.py b/yt_dlp/extractor/dropbox.py
index 214b309bf..ec86d7ad2 100644
--- a/yt_dlp/extractor/dropbox.py
+++ b/yt_dlp/extractor/dropbox.py
@@ -1,3 +1,4 @@
+import base64
 import os.path
 import re
 
@@ -5,14 +6,13 @@
 from ..compat import compat_urllib_parse_unquote
 from ..utils import (
     ExtractorError,
-    traverse_obj,
-    try_get,
+    update_url_query,
     url_basename,
 )
 
 
 class DropboxIE(InfoExtractor):
-    _VALID_URL = r'https?://(?:www\.)?dropbox[.]com/sh?/(?P<id>[a-zA-Z0-9]{15})/.*'
+    _VALID_URL = r'https?://(?:www\.)?dropbox\.com/(?:(?:e/)?scl/fi|sh?)/(?P<id>\w+)'
     _TESTS = [
         {
             'url': 'https://www.dropbox.com/s/nelirfsxnmcfbfh/youtube-dl%20test%20video%20%27%C3%A4%22BaW_jenozKc.mp4?dl=0',
@@ -22,7 +22,16 @@ class DropboxIE(InfoExtractor):
                 'title': 'youtube-dl test video \'ä"BaW_jenozKc'
             }
         }, {
-            'url': 'https://www.dropbox.com/sh/662glsejgzoj9sr/AAByil3FGH9KFNZ13e08eSa1a/Pregame%20Ceremony%20Program%20PA%2020140518.m4v',
+            'url': 'https://www.dropbox.com/s/nelirfsxnmcfbfh',
+            'only_matching': True,
+        }, {
+            'url': 'https://www.dropbox.com/sh/2mgpiuq7kv8nqdf/AABy-fW4dkydT4GmWi2mdOUDa?dl=0&preview=Drone+Shot.mp4',
+            'only_matching': True,
+        }, {
+            'url': 'https://www.dropbox.com/scl/fi/r2kd2skcy5ylbbta5y1pz/DJI_0003.MP4?dl=0&rlkey=wcdgqangn7t3lnmmv6li9mu9h',
+            'only_matching': True,
+        }, {
+            'url': 'https://www.dropbox.com/e/scl/fi/r2kd2skcy5ylbbta5y1pz/DJI_0003.MP4?dl=0&rlkey=wcdgqangn7t3lnmmv6li9mu9h',
             'only_matching': True,
         },
     ]
@@ -53,16 +62,25 @@ def _real_extract(self, url):
             else:
                 raise ExtractorError('Password protected video, use --video-password <password>', expected=True)
 
-        info_json = self._search_json(r'InitReact\.mountComponent\(.*?,', webpage, 'mountComponent', video_id,
-                                      contains_pattern=r'{.+?"preview".+?}', end_pattern=r'\)')['props']
-        transcode_url = traverse_obj(info_json, ((None, 'preview'), 'file', 'preview', 'content', 'transcode_url'), get_all=False)
-        formats, subtitles = self._extract_m3u8_formats_and_subtitles(transcode_url, video_id)
+        formats, subtitles, has_anonymous_download = [], {}, False
+        for encoded in reversed(re.findall(r'registerStreamedPrefetch\s*\(\s*"[\w/+=]+"\s*,\s*"([\w/+=]+)"', webpage)):
+            decoded = base64.b64decode(encoded).decode('utf-8', 'ignore')
+            transcode_url = self._search_regex(
+                r'\n\x03(https://[^\x12\x03\n]+\.m3u8)', decoded, 'transcode url', default=None)
+            if not transcode_url:
+                continue
+            formats, subtitles = self._extract_m3u8_formats_and_subtitles(transcode_url, video_id)
+            has_anonymous_download = self._search_regex(r'(anonymous:\tanonymous)', decoded, 'anonymous', default=False)
+            break
 
         # downloads enabled we can get the original file
-        if 'anonymous' in (try_get(info_json, lambda x: x['sharePermission']['canDownloadRoles']) or []):
-            video_url = re.sub(r'[?&]dl=0', '', url)
-            video_url += ('?' if '?' not in video_url else '&') + 'dl=1'
-            formats.append({'url': video_url, 'format_id': 'original', 'format_note': 'Original', 'quality': 1})
+        if has_anonymous_download:
+            formats.append({
+                'url': update_url_query(url, {'dl': '1'}),
+                'format_id': 'original',
+                'format_note': 'Original',
+                'quality': 1
+            })
 
         return {
             'id': video_id,
diff --git a/yt_dlp/extractor/mzaalo.py b/yt_dlp/extractor/mzaalo.py
index c6f420cea..1996368cc 100644
--- a/yt_dlp/extractor/mzaalo.py
+++ b/yt_dlp/extractor/mzaalo.py
@@ -8,7 +8,7 @@
 
 
 class MzaaloIE(InfoExtractor):
-    _VALID_URL = r'https?://(?:www\.)?mzaalo\.com/play/(?P<type>movie|original|clip)/(?P<id>[a-fA-F0-9-]+)/[\w-]+'
+    _VALID_URL = r'(?i)https?://(?:www\.)?mzaalo\.com/(?:play|watch)/(?P<type>movie|original|clip)/(?P<id>[a-f0-9-]+)/[\w-]+'
     _TESTS = [{
         # Movies
         'url': 'https://www.mzaalo.com/play/movie/c0958d9f-f90e-4503-a755-44358758921d/Jamun',
@@ -55,6 +55,9 @@ class MzaaloIE(InfoExtractor):
             'language': 'hin',
         },
         'params': {'skip_download': 'm3u8'}
+    }, {
+        'url': 'https://mzaalo.com/watch/MOVIE/389c892d-0b65-4019-bf73-d4edcb1c014f/Chalo-Dilli',
+        'only_matching': True,
     }]
 
     def _real_extract(self, url):
diff --git a/yt_dlp/extractor/stageplus.py b/yt_dlp/extractor/stageplus.py
index adb4ebbc2..4bed4d646 100644
--- a/yt_dlp/extractor/stageplus.py
+++ b/yt_dlp/extractor/stageplus.py
@@ -484,18 +484,15 @@ def _real_extract(self, url):
             'url': 'url',
         })) or None
 
-        m3u8_headers = {'jwt': self._TOKEN}
-
         entries = []
         for idx, video in enumerate(traverse_obj(data, (
                 'performanceWorks', lambda _, v: v['id'] and url_or_none(v['stream']['url']))), 1):
             formats, subtitles = self._extract_m3u8_formats_and_subtitles(
-                video['stream']['url'], video['id'], 'mp4', m3u8_id='hls', headers=m3u8_headers)
+                video['stream']['url'], video['id'], 'mp4', m3u8_id='hls', query={'token': self._TOKEN})
             entries.append({
                 'id': video['id'],
                 'formats': formats,
                 'subtitles': subtitles,
-                'http_headers': m3u8_headers,
                 'album': metadata.get('title'),
                 'album_artist': metadata.get('artist'),
                 'track_number': idx,
diff --git a/yt_dlp/extractor/weverse.py b/yt_dlp/extractor/weverse.py
index 9a08b8e43..bbf62856a 100644
--- a/yt_dlp/extractor/weverse.py
+++ b/yt_dlp/extractor/weverse.py
@@ -70,10 +70,8 @@ def _real_initialize(self):
             return
 
         token = try_call(lambda: self._get_cookies('https://weverse.io/')['we2_access_token'].value)
-        if not token:
-            self.raise_login_required()
-
-        WeverseBaseIE._API_HEADERS['Authorization'] = f'Bearer {token}'
+        if token:
+            WeverseBaseIE._API_HEADERS['Authorization'] = f'Bearer {token}'
 
     def _call_api(self, ep, video_id, data=None, note='Downloading API JSON'):
         # Ref: https://ssl.pstatic.net/static/wevweb/2_3_2_11101725/public/static/js/2488.a09b41ff.chunk.js
@@ -101,11 +99,14 @@ def _call_api(self, ep, video_id, data=None, note='Downloading API JSON'):
                 self.raise_login_required(
                     'Session token has expired. Log in again or refresh cookies in browser')
             elif isinstance(e.cause, HTTPError) and e.cause.status == 403:
-                raise ExtractorError('Your account does not have access to this content', expected=True)
+                if 'Authorization' in self._API_HEADERS:
+                    raise ExtractorError('Your account does not have access to this content', expected=True)
+                self.raise_login_required()
             raise
 
     def _call_post_api(self, video_id):
-        return self._call_api(f'/post/v1.0/post-{video_id}?fieldSet=postV1', video_id)
+        path = '' if 'Authorization' in self._API_HEADERS else '/preview'
+        return self._call_api(f'/post/v1.0/post-{video_id}{path}?fieldSet=postV1', video_id)
 
     def _get_community_id(self, channel):
         return str(self._call_api(
diff --git a/yt_dlp/networking/_urllib.py b/yt_dlp/networking/_urllib.py
index 0c4794954..5a804d99b 100644
--- a/yt_dlp/networking/_urllib.py
+++ b/yt_dlp/networking/_urllib.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import functools
-import gzip
 import http.client
 import io
 import socket
@@ -155,20 +154,9 @@ def brotli(data):
 
     @staticmethod
     def gz(data):
-        gz = gzip.GzipFile(fileobj=io.BytesIO(data), mode='rb')
-        try:
-            return gz.read()
-        except OSError as original_oserror:
-            # There may be junk add the end of the file
-            # See http://stackoverflow.com/q/4928560/35070 for details
-            for i in range(1, 1024):
-                try:
-                    gz = gzip.GzipFile(fileobj=io.BytesIO(data[:-i]), mode='rb')
-                    return gz.read()
-                except OSError:
-                    continue
-            else:
-                raise original_oserror
+        # There may be junk added the end of the file
+        # We ignore it by only ever decoding a single gzip payload
+        return zlib.decompress(data, wbits=zlib.MAX_WBITS | 16)
 
     def http_request(self, req):
         # According to RFC 3986, URLs can not contain non-ASCII characters, however this is not


Test patch:
diff --git a/test/conftest.py b/test/conftest.py
new file mode 100644
index 000000000..15549d30b
--- /dev/null
+++ b/test/conftest.py
@@ -0,0 +1,21 @@
+import functools
+import inspect
+
+import pytest
+
+from yt_dlp.networking import RequestHandler
+from yt_dlp.networking.common import _REQUEST_HANDLERS
+from yt_dlp.utils._utils import _YDLLogger as FakeLogger
+
+
+@pytest.fixture
+def handler(request):
+    RH_KEY = request.param
+    if inspect.isclass(RH_KEY) and issubclass(RH_KEY, RequestHandler):
+        handler = RH_KEY
+    elif RH_KEY in _REQUEST_HANDLERS:
+        handler = _REQUEST_HANDLERS[RH_KEY]
+    else:
+        pytest.skip(f'{RH_KEY} request handler is not available')
+
+    return functools.partial(handler, logger=FakeLogger)
diff --git a/test/test_networking.py b/test/test_networking.py
index 2622d24da..5308c8d6f 100644
--- a/test/test_networking.py
+++ b/test/test_networking.py
@@ -8,12 +8,10 @@
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-import functools
 import gzip
 import http.client
 import http.cookiejar
 import http.server
-import inspect
 import io
 import pathlib
 import random
@@ -40,7 +38,6 @@
     Response,
 )
 from yt_dlp.networking._urllib import UrllibRH
-from yt_dlp.networking.common import _REQUEST_HANDLERS
 from yt_dlp.networking.exceptions import (
     CertificateVerifyError,
     HTTPError,
@@ -307,19 +304,6 @@ def setup_class(cls):
         cls.https_server_thread.start()
 
 
-@pytest.fixture
-def handler(request):
-    RH_KEY = request.param
-    if inspect.isclass(RH_KEY) and issubclass(RH_KEY, RequestHandler):
-        handler = RH_KEY
-    elif RH_KEY in _REQUEST_HANDLERS:
-        handler = _REQUEST_HANDLERS[RH_KEY]
-    else:
-        pytest.skip(f'{RH_KEY} request handler is not available')
-
-    return functools.partial(handler, logger=FakeLogger)
-
-
 class TestHTTPRequestHandler(TestRequestHandlerBase):
     @pytest.mark.parametrize('handler', ['Urllib'], indirect=True)
     def test_verify_cert(self, handler):
diff --git a/test/test_socks.py b/test/test_socks.py
index 6651290d2..95ffce275 100644
--- a/test/test_socks.py
+++ b/test/test_socks.py
@@ -1,113 +1,470 @@
 #!/usr/bin/env python3
-
 # Allow direct execution
 import os
 import sys
+import threading
 import unittest
 
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import pytest
 
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+import abc
+import contextlib
+import enum
+import functools
+import http.server
+import json
 import random
-import subprocess
-import urllib.request
+import socket
+import struct
+import time
+from socketserver import (
+    BaseRequestHandler,
+    StreamRequestHandler,
+    ThreadingTCPServer,
+)
 
-from test.helper import FakeYDL, get_params, is_download_test
+from test.helper import http_server_port
+from yt_dlp.networking import Request
+from yt_dlp.networking.exceptions import ProxyError, TransportError
+from yt_dlp.socks import (
+    SOCKS4_REPLY_VERSION,
+    SOCKS4_VERSION,
+    SOCKS5_USER_AUTH_SUCCESS,
+    SOCKS5_USER_AUTH_VERSION,
+    SOCKS5_VERSION,
+    Socks5AddressType,
+    Socks5Auth,
+)
 
+SOCKS5_USER_AUTH_FAILURE = 0x1
 
-@is_download_test
-class TestMultipleSocks(unittest.TestCase):
-    @staticmethod
-    def _check_params(attrs):
-        params = get_params()
-        for attr in attrs:
-            if attr not in params:
-                print('Missing %s. Skipping.' % attr)
-                return
-        return params
 
-    def test_proxy_http(self):
-        params = self._check_params(['primary_proxy', 'primary_server_ip'])
-        if params is None:
+class Socks4CD(enum.IntEnum):
+    REQUEST_GRANTED = 90
+    REQUEST_REJECTED_OR_FAILED = 91
+    REQUEST_REJECTED_CANNOT_CONNECT_TO_IDENTD = 92
+    REQUEST_REJECTED_DIFFERENT_USERID = 93
+
+
+class Socks5Reply(enum.IntEnum):
+    SUCCEEDED = 0x0
+    GENERAL_FAILURE = 0x1
+    CONNECTION_NOT_ALLOWED = 0x2
+    NETWORK_UNREACHABLE = 0x3
+    HOST_UNREACHABLE = 0x4
+    CONNECTION_REFUSED = 0x5
+    TTL_EXPIRED = 0x6
+    COMMAND_NOT_SUPPORTED = 0x7
+    ADDRESS_TYPE_NOT_SUPPORTED = 0x8
+
+
+class SocksTestRequestHandler(BaseRequestHandler):
+
+    def __init__(self, *args, socks_info=None, **kwargs):
+        self.socks_info = socks_info
+        super().__init__(*args, **kwargs)
+
+
+class SocksProxyHandler(BaseRequestHandler):
+    def __init__(self, request_handler_class, socks_server_kwargs, *args, **kwargs):
+        self.socks_kwargs = socks_server_kwargs or {}
+        self.request_handler_class = request_handler_class
+        super().__init__(*args, **kwargs)
+
+
+class Socks5ProxyHandler(StreamRequestHandler, SocksProxyHandler):
+
+    # SOCKS5 protocol https://tools.ietf.org/html/rfc1928
+    # SOCKS5 username/password authentication https://tools.ietf.org/html/rfc1929
+
+    def handle(self):
+        sleep = self.socks_kwargs.get('sleep')
+        if sleep:
+            time.sleep(sleep)
+        version, nmethods = self.connection.recv(2)
+        assert version == SOCKS5_VERSION
+        methods = list(self.connection.recv(nmethods))
+
+        auth = self.socks_kwargs.get('auth')
+
+        if auth is not None and Socks5Auth.AUTH_USER_PASS not in methods:
+            self.connection.sendall(struct.pack('!BB', SOCKS5_VERSION, Socks5Auth.AUTH_NO_ACCEPTABLE))
+            self.server.close_request(self.request)
             return
-        ydl = FakeYDL({
-            'proxy': params['primary_proxy']
-        })
-        self.assertEqual(
-            ydl.urlopen('http://yt-dl.org/ip').read().decode(),
-            params['primary_server_ip'])
-
-    def test_proxy_https(self):
-        params = self._check_params(['primary_proxy', 'primary_server_ip'])
-        if params is None:
+
+        elif Socks5Auth.AUTH_USER_PASS in methods:
+            self.connection.sendall(struct.pack("!BB", SOCKS5_VERSION, Socks5Auth.AUTH_USER_PASS))
+
+            _, user_len = struct.unpack('!BB', self.connection.recv(2))
+            username = self.connection.recv(user_len).decode()
+            pass_len = ord(self.connection.recv(1))
+            password = self.connection.recv(pass_len).decode()
+
+            if username == auth[0] and password == auth[1]:
+                self.connection.sendall(struct.pack('!BB', SOCKS5_USER_AUTH_VERSION, SOCKS5_USER_AUTH_SUCCESS))
+            else:
+                self.connection.sendall(struct.pack('!BB', SOCKS5_USER_AUTH_VERSION, SOCKS5_USER_AUTH_FAILURE))
+                self.server.close_request(self.request)
+                return
+
+        elif Socks5Auth.AUTH_NONE in methods:
+            self.connection.sendall(struct.pack('!BB', SOCKS5_VERSION, Socks5Auth.AUTH_NONE))
+        else:
+            self.connection.sendall(struct.pack('!BB', SOCKS5_VERSION, Socks5Auth.AUTH_NO_ACCEPTABLE))
+            self.server.close_request(self.request)
             return
-        ydl = FakeYDL({
-            'proxy': params['primary_proxy']
-        })
-        self.assertEqual(
-            ydl.urlopen('https://yt-dl.org/ip').read().decode(),
-            params['primary_server_ip'])
-
-    def test_secondary_proxy_http(self):
-        params = self._check_params(['secondary_proxy', 'secondary_server_ip'])
-        if params is None:
+
+        version, command, _, address_type = struct.unpack('!BBBB', self.connection.recv(4))
+        socks_info = {
+            'version': version,
+            'auth_methods': methods,
+            'command': command,
+            'client_address': self.client_address,
+            'ipv4_address': None,
+            'domain_address': None,
+            'ipv6_address': None,
+        }
+        if address_type == Socks5AddressType.ATYP_IPV4:
+            socks_info['ipv4_address'] = socket.inet_ntoa(self.connection.recv(4))
+        elif address_type == Socks5AddressType.ATYP_DOMAINNAME:
+            socks_info['domain_address'] = self.connection.recv(ord(self.connection.recv(1))).decode()
+        elif address_type == Socks5AddressType.ATYP_IPV6:
+            socks_info['ipv6_address'] = socket.inet_ntop(socket.AF_INET6, self.connection.recv(16))
+        else:
+            self.server.close_request(self.request)
+
+        socks_info['port'] = struct.unpack('!H', self.connection.recv(2))[0]
+
+        # dummy response, the returned IP is just a placeholder
+        self.connection.sendall(struct.pack(
+            '!BBBBIH', SOCKS5_VERSION, self.socks_kwargs.get('reply', Socks5Reply.SUCCEEDED), 0x0, 0x1, 0x7f000001, 40000))
+
+        self.request_handler_class(self.request, self.client_address, self.server, socks_info=socks_info)
+
+
+class Socks4ProxyHandler(StreamRequestHandler, SocksProxyHandler):
+
+    # SOCKS4 protocol http://www.openssh.com/txt/socks4.protocol
+    # SOCKS4A protocol http://www.openssh.com/txt/socks4a.protocol
+
+    def _read_until_null(self):
+        return b''.join(iter(functools.partial(self.connection.recv, 1), b'\x00'))
+
+    def handle(self):
+        sleep = self.socks_kwargs.get('sleep')
+        if sleep:
+            time.sleep(sleep)
+        socks_info = {
+            'version': SOCKS4_VERSION,
+            'command': None,
+            'client_address': self.client_address,
+            'ipv4_address': None,
+            'port': None,
+            'domain_address': None,
+        }
+        version, command, dest_port, dest_ip = struct.unpack('!BBHI', self.connection.recv(8))
+        socks_info['port'] = dest_port
+        socks_info['command'] = command
+        if version != SOCKS4_VERSION:
+            self.server.close_request(self.request)
             return
-        ydl = FakeYDL()
-        req = urllib.request.Request('http://yt-dl.org/ip')
-        req.add_header('Ytdl-request-proxy', params['secondary_proxy'])
-        self.assertEqual(
-            ydl.urlopen(req).read().decode(),
-            params['secondary_server_ip'])
-
-    def test_secondary_proxy_https(self):
-        params = self._check_params(['secondary_proxy', 'secondary_server_ip'])
-        if params is None:
+        use_remote_dns = False
+        if 0x0 < dest_ip <= 0xFF:
+            use_remote_dns = True
+        else:
+            socks_info['ipv4_address'] = socket.inet_ntoa(struct.pack("!I", dest_ip))
+
+        user_id = self._read_until_null().decode()
+        if user_id != (self.socks_kwargs.get('user_id') or ''):
+            self.connection.sendall(struct.pack(
+                '!BBHI', SOCKS4_REPLY_VERSION, Socks4CD.REQUEST_REJECTED_DIFFERENT_USERID, 0x00, 0x00000000))
+            self.server.close_request(self.request)
             return
-        ydl = FakeYDL()
-        req = urllib.request.Request('https://yt-dl.org/ip')
-        req.add_header('Ytdl-request-proxy', params['secondary_proxy'])
-        self.assertEqual(
-            ydl.urlopen(req).read().decode(),
-            params['secondary_server_ip'])
 
+        if use_remote_dns:
+            socks_info['domain_address'] = self._read_until_null().decode()
 
-@is_download_test
-class TestSocks(unittest.TestCase):
-    _SKIP_SOCKS_TEST = True
+        # dummy response, the returned IP is just a placeholder
+        self.connection.sendall(
+            struct.pack(
+                '!BBHI', SOCKS4_REPLY_VERSION,
+                self.socks_kwargs.get('cd_reply', Socks4CD.REQUEST_GRANTED), 40000, 0x7f000001))
 
-    def setUp(self):
-        if self._SKIP_SOCKS_TEST:
-            return
+        self.request_handler_class(self.request, self.client_address, self.server, socks_info=socks_info)
 
-        self.port = random.randint(20000, 30000)
-        self.server_process = subprocess.Popen([
-            'srelay', '-f', '-i', '127.0.0.1:%d' % self.port],
-            stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
-    def tearDown(self):
-        if self._SKIP_SOCKS_TEST:
-            return
+class IPv6ThreadingTCPServer(ThreadingTCPServer):
+    address_family = socket.AF_INET6
+
+
+class SocksHTTPTestRequestHandler(http.server.BaseHTTPRequestHandler, SocksTestRequestHandler):
+    def do_GET(self):
+        if self.path == '/socks_info':
+            payload = json.dumps(self.socks_info.copy())
+            self.send_response(200)
+            self.send_header('Content-Type', 'application/json; charset=utf-8')
+            self.send_header('Content-Length', str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload.encode())
+
+
+@contextlib.contextmanager
+def socks_server(socks_server_class, request_handler, bind_ip=None, **socks_server_kwargs):
+    server = server_thread = None
+    try:
+        bind_address = bind_ip or '127.0.0.1'
+        server_type = ThreadingTCPServer if '.' in bind_address else IPv6ThreadingTCPServer
+        server = server_type(
+            (bind_address, 0), functools.partial(socks_server_class, request_handler, socks_server_kwargs))
+        server_port = http_server_port(server)
+        server_thread = threading.Thread(target=server.serve_forever)
+        server_thread.daemon = True
+        server_thread.start()
+        if '.' not in bind_address:
+            yield f'[{bind_address}]:{server_port}'
+        else:
+            yield f'{bind_address}:{server_port}'
+    finally:
+        server.shutdown()
+        server.server_close()
+        server_thread.join(2.0)
+
+
+class SocksProxyTestContext(abc.ABC):
+    REQUEST_HANDLER_CLASS = None
+
+    def socks_server(self, server_class, *args, **kwargs):
+        return socks_server(server_class, self.REQUEST_HANDLER_CLASS, *args, **kwargs)
+
+    @abc.abstractmethod
+    def socks_info_request(self, handler, target_domain=None, target_port=None, **req_kwargs) -> dict:
+        """return a dict of socks_info"""
+
+
+class HTTPSocksTestProxyContext(SocksProxyTestContext):
+    REQUEST_HANDLER_CLASS = SocksHTTPTestRequestHandler
+
+    def socks_info_request(self, handler, target_domain=None, target_port=None, **req_kwargs):
+        request = Request(f'http://{target_domain or "127.0.0.1"}:{target_port or "40000"}/socks_info', **req_kwargs)
+        handler.validate(request)
+        return json.loads(handler.send(request).read().decode())
+
+
+CTX_MAP = {
+    'http': HTTPSocksTestProxyContext,
+}
+
+
+@pytest.fixture(scope='module')
+def ctx(request):
+    return CTX_MAP[request.param]()
+
+
+class TestSocks4Proxy:
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks4_no_auth(self, handler, ctx):
+        with handler() as rh:
+            with ctx.socks_server(Socks4ProxyHandler) as server_address:
+                response = ctx.socks_info_request(
+                    rh, proxies={'all': f'socks4://{server_address}'})
+                assert response['version'] == 4
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks4_auth(self, handler, ctx):
+        with handler() as rh:
+            with ctx.socks_server(Socks4ProxyHandler, user_id='user') as server_address:
+                with pytest.raises(ProxyError):
+                    ctx.socks_info_request(rh, proxies={'all': f'socks4://{server_address}'})
+                response = ctx.socks_info_request(
+                    rh, proxies={'all': f'socks4://user:@{server_address}'})
+                assert response['version'] == 4
+
+    @pytest.mark.parametrize('handler,ctx', [
+        pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
+            reason='socks4a implementation currently broken when destination is not a domain name'))
+    ], indirect=True)
+    def test_socks4a_ipv4_target(self, handler, ctx):
+        with ctx.socks_server(Socks4ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks4a://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='127.0.0.1')
+                assert response['version'] == 4
+                assert response['ipv4_address'] == '127.0.0.1'
+                assert response['domain_address'] is None
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks4a_domain_target(self, handler, ctx):
+        with ctx.socks_server(Socks4ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks4a://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='localhost')
+                assert response['version'] == 4
+                assert response['ipv4_address'] is None
+                assert response['domain_address'] == 'localhost'
+
+    @pytest.mark.parametrize('handler,ctx', [
+        pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
+            reason='source_address is not yet supported for socks4 proxies'))
+    ], indirect=True)
+    def test_ipv4_client_source_address(self, handler, ctx):
+        with ctx.socks_server(Socks4ProxyHandler) as server_address:
+            source_address = f'127.0.0.{random.randint(5, 255)}'
+            with handler(proxies={'all': f'socks4://{server_address}'},
+                         source_address=source_address) as rh:
+                response = ctx.socks_info_request(rh)
+                assert response['client_address'][0] == source_address
+                assert response['version'] == 4
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    @pytest.mark.parametrize('reply_code', [
+        Socks4CD.REQUEST_REJECTED_OR_FAILED,
+        Socks4CD.REQUEST_REJECTED_CANNOT_CONNECT_TO_IDENTD,
+        Socks4CD.REQUEST_REJECTED_DIFFERENT_USERID,
+    ])
+    def test_socks4_errors(self, handler, ctx, reply_code):
+        with ctx.socks_server(Socks4ProxyHandler, cd_reply=reply_code) as server_address:
+            with handler(proxies={'all': f'socks4://{server_address}'}) as rh:
+                with pytest.raises(ProxyError):
+                    ctx.socks_info_request(rh)
+
+    @pytest.mark.parametrize('handler,ctx', [
+        pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
+            reason='IPv6 socks4 proxies are not yet supported'))
+    ], indirect=True)
+    def test_ipv6_socks4_proxy(self, handler, ctx):
+        with ctx.socks_server(Socks4ProxyHandler, bind_ip='::1') as server_address:
+            with handler(proxies={'all': f'socks4://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='127.0.0.1')
+                assert response['client_address'][0] == '::1'
+                assert response['ipv4_address'] == '127.0.0.1'
+                assert response['version'] == 4
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_timeout(self, handler, ctx):
+        with ctx.socks_server(Socks4ProxyHandler, sleep=2) as server_address:
+            with handler(proxies={'all': f'socks4://{server_address}'}, timeout=1) as rh:
+                with pytest.raises(TransportError):
+                    ctx.socks_info_request(rh)
+
+
+class TestSocks5Proxy:
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks5_no_auth(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh)
+                assert response['auth_methods'] == [0x0]
+                assert response['version'] == 5
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks5_user_pass(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler, auth=('test', 'testpass')) as server_address:
+            with handler() as rh:
+                with pytest.raises(ProxyError):
+                    ctx.socks_info_request(rh, proxies={'all': f'socks5://{server_address}'})
+
+                response = ctx.socks_info_request(
+                    rh, proxies={'all': f'socks5://test:testpass@{server_address}'})
+
+                assert response['auth_methods'] == [Socks5Auth.AUTH_NONE, Socks5Auth.AUTH_USER_PASS]
+                assert response['version'] == 5
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks5_ipv4_target(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='127.0.0.1')
+                assert response['ipv4_address'] == '127.0.0.1'
+                assert response['version'] == 5
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks5_domain_target(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='localhost')
+                assert response['ipv4_address'] == '127.0.0.1'
+                assert response['version'] == 5
+
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks5h_domain_target(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks5h://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='localhost')
+                assert response['ipv4_address'] is None
+                assert response['domain_address'] == 'localhost'
+                assert response['version'] == 5
 
-        self.server_process.terminate()
-        self.server_process.communicate()
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_socks5h_ip_target(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks5h://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='127.0.0.1')
+                assert response['ipv4_address'] == '127.0.0.1'
+                assert response['domain_address'] is None
+                assert response['version'] == 5
 
-    def _get_ip(self, protocol):
-        if self._SKIP_SOCKS_TEST:
-            return '127.0.0.1'
+    @pytest.mark.parametrize('handler,ctx', [
+        pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
+            reason='IPv6 destination addresses are not yet supported'))
+    ], indirect=True)
+    def test_socks5_ipv6_destination(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler) as server_address:
+            with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='[::1]')
+                assert response['ipv6_address'] == '::1'
+                assert response['port'] == 80
+                assert response['version'] == 5
 
-        ydl = FakeYDL({
-            'proxy': '%s://127.0.0.1:%d' % (protocol, self.port),
-        })
-        return ydl.urlopen('http://yt-dl.org/ip').read().decode()
+    @pytest.mark.parametrize('handler,ctx', [
+        pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
+            reason='IPv6 socks5 proxies are not yet supported'))
+    ], indirect=True)
+    def test_ipv6_socks5_proxy(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler, bind_ip='::1') as server_address:
+            with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
+                response = ctx.socks_info_request(rh, target_domain='127.0.0.1')
+                assert response['client_address'][0] == '::1'
+                assert response['ipv4_address'] == '127.0.0.1'
+                assert response['version'] == 5
 
-    def test_socks4(self):
-        self.assertTrue(isinstance(self._get_ip('socks4'), str))
+    # XXX: is there any feasible way of testing IPv6 source addresses?
+    # Same would go for non-proxy source_address test...
+    @pytest.mark.parametrize('handler,ctx', [
+        pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
+            reason='source_address is not yet supported for socks5 proxies'))
+    ], indirect=True)
+    def test_ipv4_client_source_address(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler) as server_address:
+            source_address = f'127.0.0.{random.randint(5, 255)}'
+            with handler(proxies={'all': f'socks5://{server_address}'}, source_address=source_address) as rh:
+                response = ctx.socks_info_request(rh)
+                assert response['client_address'][0] == source_address
+                assert response['version'] == 5
 
-    def test_socks4a(self):
-        self.assertTrue(isinstance(self._get_ip('socks4a'), str))
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    @pytest.mark.parametrize('reply_code', [
+        Socks5Reply.GENERAL_FAILURE,
+        Socks5Reply.CONNECTION_NOT_ALLOWED,
+        Socks5Reply.NETWORK_UNREACHABLE,
+        Socks5Reply.HOST_UNREACHABLE,
+        Socks5Reply.CONNECTION_REFUSED,
+        Socks5Reply.TTL_EXPIRED,
+        Socks5Reply.COMMAND_NOT_SUPPORTED,
+        Socks5Reply.ADDRESS_TYPE_NOT_SUPPORTED,
+    ])
+    def test_socks5_errors(self, handler, ctx, reply_code):
+        with ctx.socks_server(Socks5ProxyHandler, reply=reply_code) as server_address:
+            with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
+                with pytest.raises(ProxyError):
+                    ctx.socks_info_request(rh)
 
-    def test_socks5(self):
-        self.assertTrue(isinstance(self._get_ip('socks5'), str))
+    @pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
+    def test_timeout(self, handler, ctx):
+        with ctx.socks_server(Socks5ProxyHandler, sleep=2) as server_address:
+            with handler(proxies={'all': f'socks5://{server_address}'}, timeout=1) as rh:
+                with pytest.raises(TransportError):
+                    ctx.socks_info_request(rh)
 
 
 if __name__ == '__main__':