Repository: yt-dlp/yt-dlp
Pull request #7142 (2023-05-27 09:09:50)

[core] Support decoding multiple content encodings
Authored by: coletdjnz

**IMPORTANT**: PRs without the template will be CLOSED

### Description of your *pull request* and other information

<!--

Explanation of your *pull request* in arbitrary form goes here. Please **make sure the description explains the purpose and effect** of your *pull request* and is worded well enough to be understood. Provide as much **context and examples** as possible

-->

ADD DESCRIPTION HERE

Inspired by [a comment by @dirkf](https://github.com/yt-dlp/yt-dlp/issues/5855#issuecomment-1363539612) on that urllib does not have support for multi content decoding, as defined in [RFC9110](https://datatracker.ietf.org/doc/html/rfc9110#name-content-encoding). Decided to implement this as an exercise as it was fairly simple while needing to fix the below.

This also fixes the errors that occur in the odd case the webserver returns a brotli encoded response when the user does not have brotli support installed (https://github.com/yt-dlp/yt-dlp/issues/4748, https://github.com/yt-dlp/yt-dlp/issues/5855). Instead, it will return the raw brotli encoded content in line with the behaviour of the others.



<details open><summary>Template</summary> <!-- OPEN is intentional -->

<!--

# PLEASE FOLLOW THE GUIDE BELOW

- You will be asked some questions, please read them **carefully** and answer honestly
- Put an `x` into all the boxes `[ ]` relevant to your *pull request* (like [x])
- Use *Preview* tab to see how your *pull request* will actually look like

-->

### Before submitting a *pull request* make sure you have:
- [X] At least skimmed through [contributing guidelines](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#developer-instructions) including [yt-dlp coding conventions](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#yt-dlp-coding-conventions)
- [X] [Searched](https://github.com/yt-dlp/yt-dlp/search?q=is%3Apr&type=Issues) the bugtracker for similar pull requests
- [X] Checked the code with [flake8](https://pypi.python.org/pypi/flake8) and [ran relevant tests](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#developer-instructions)

### In order to be accepted and merged into yt-dlp each piece of code must be in public domain or released under [Unlicense](http://unlicense.org/). Check all of the following options that apply:
- [X] I am the original author of this code and I am willing to release it under [Unlicense](http://unlicense.org/)
- [ ] I am not the original author of this code but it is in public domain or released under [Unlicense](http://unlicense.org/) (provide reliable evidence)

### What is the purpose of your *pull request*?
- [ ] Fix or improvement to an extractor (Make sure to add/update tests)
- [ ] New extractor ([Piracy websites will not be accepted](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#is-the-website-primarily-used-for-piracy))
- [X] Core bug fix/improvement
- [ ] New feature (It is strongly [recommended to open an issue first](https://github.com/yt-dlp/yt-dlp/blob/master/CONTRIBUTING.md#adding-new-feature-or-making-overarching-changes))


<!-- Do NOT edit/remove anything below this! -->
</details><details><summary>Copilot Summary</summary>  

<!--
copilot:all
-->
### <samp>🤖 Generated by Copilot at 3b62ee5</samp>

### Summary
🗜️🧪🛠️

<!--
1.  🗜️ - This emoji represents compression and decompression, which are the main features added by this pull request. It can also be seen as a symbol of reducing the size of data or making it more efficient.
2.  🧪 - This emoji represents testing, which is another important aspect of this pull request. It shows that the code is verified and validated by adding test cases for different scenarios and encodings.
3.  🛠️ - This emoji represents fixing and refactoring, which are the secondary goals of this pull request. It shows that the code is improved and made more robust and consistent by fixing bugs and refactoring existing logic.
-->
Add support for different content-encodings in HTTP responses. Improve the `YoutubeDLHandler` class to decompress gzip files with junk data and handle multiple and unsupported encodings. Update `test_http.py` to test the new functionality.

> _Oh we are the coders of the sea, and we work on `YoutubeDL`_
> _We handle the responses that we get, whether they're compressed or not_
> _We use `zlib` and `brotli` to decompress what we fetch_
> _And we heave away on the count of three, to raise our code quality_

### Walkthrough
*  Decompress HTTP responses with different content-encodings using `zlib` and `brotli` modules ([link](https://github.com/yt-dlp/yt-dlp/pull/7142/files?diff=unified&w=0#diff-feda8f56946dc048e754111850baaef0eec4b9f8bbc2d3f04b1a785626ea5c0eR1364-R1380), [link](https://github.com/yt-dlp/yt-dlp/pull/7142/files?diff=unified&w=0#diff-feda8f56946dc048e754111850baaef0eec4b9f8bbc2d3f04b1a785626ea5c0eL1397-R1429))
* Import `zlib` and `brotli` modules from `yt_dlp.dependencies` in `test_http.py` ([link](https://github.com/yt-dlp/yt-dlp/pull/7142/files?diff=unified&w=0#diff-a02095e104f9a967ff2b3b674faaf042cb3442f8f53ade020ee9c195c374311bL20-R24))
* Add handler for `/content-encoding` path in `HTTPTestRequestHandler` class to mock web server responses with different content-encodings ([link](https://github.com/yt-dlp/yt-dlp/pull/7142/files?diff=unified&w=0#diff-a02095e104f9a967ff2b3b674faaf042cb3442f8f53ade020ee9c195c374311bR153-R177))
* Add test cases to `TestHTTP` class to test `YoutubeDL` class's ability to handle different content-encodings ([link](https://github.com/yt-dlp/yt-dlp/pull/7142/files?diff=unified&w=0#diff-a02095e104f9a967ff2b3b674faaf042cb3442f8f53ade020ee9c195c374311bL305-R384))



</details>


coletdjnz: I will note that if brotli is not available and server returns a multi content-encoding (such as `gzip,br`), then decoding will fail fatally. Extremely unlikely we'll ever see this so no need to handle it imo.

Patch:
diff --git a/yt_dlp/utils/_utils.py b/yt_dlp/utils/_utils.py
index 6f4f22bb3..7c91faff8 100644
--- a/yt_dlp/utils/_utils.py
+++ b/yt_dlp/utils/_utils.py
@@ -1361,6 +1361,23 @@ def brotli(data):
             return data
         return brotli.decompress(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
+
     def http_request(self, req):
         # According to RFC 3986, URLs can not contain non-ASCII characters, however this is not
         # always respected by websites, some tend to give out URLs with non percent-encoded
@@ -1394,35 +1411,21 @@ def http_request(self, req):
 
     def http_response(self, req, resp):
         old_resp = resp
-        # gzip
-        if resp.headers.get('Content-encoding', '') == 'gzip':
-            content = resp.read()
-            gz = gzip.GzipFile(fileobj=io.BytesIO(content), mode='rb')
-            try:
-                uncompressed = io.BytesIO(gz.read())
-            except OSError as original_ioerror:
-                # 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(content[:-i]), mode='rb')
-                        uncompressed = io.BytesIO(gz.read())
-                    except OSError:
-                        continue
-                    break
-                else:
-                    raise original_ioerror
-            resp = urllib.request.addinfourl(uncompressed, old_resp.headers, old_resp.url, old_resp.code)
-            resp.msg = old_resp.msg
-        # deflate
-        if resp.headers.get('Content-encoding', '') == 'deflate':
-            gz = io.BytesIO(self.deflate(resp.read()))
-            resp = urllib.request.addinfourl(gz, old_resp.headers, old_resp.url, old_resp.code)
-            resp.msg = old_resp.msg
-        # brotli
-        if resp.headers.get('Content-encoding', '') == 'br':
-            resp = urllib.request.addinfourl(
-                io.BytesIO(self.brotli(resp.read())), old_resp.headers, old_resp.url, old_resp.code)
+
+        # Content-Encoding header lists the encodings in order that they were applied [1].
+        # To decompress, we simply do the reverse.
+        # [1]: https://datatracker.ietf.org/doc/html/rfc9110#name-content-encoding
+        decoded_response = None
+        for encoding in (e.strip() for e in reversed(resp.headers.get('Content-encoding', '').split(','))):
+            if encoding == 'gzip':
+                decoded_response = self.gz(decoded_response or resp.read())
+            elif encoding == 'deflate':
+                decoded_response = self.deflate(decoded_response or resp.read())
+            elif encoding == 'br' and brotli:
+                decoded_response = self.brotli(decoded_response or resp.read())
+
+        if decoded_response is not None:
+            resp = urllib.request.addinfourl(io.BytesIO(decoded_response), old_resp.headers, old_resp.url, old_resp.code)
             resp.msg = old_resp.msg
         # Percent-encode redirect URL of Location HTTP header to satisfy RFC 3986 (see
         # https://github.com/ytdl-org/youtube-dl/issues/6457).


Test patch:
diff --git a/test/test_http.py b/test/test_http.py
index d684905da..3941a6e77 100644
--- a/test/test_http.py
+++ b/test/test_http.py
@@ -17,9 +17,11 @@
 import threading
 import urllib.error
 import urllib.request
+import zlib
 
 from test.helper import http_server_port
 from yt_dlp import YoutubeDL
+from yt_dlp.dependencies import brotli
 from yt_dlp.utils import sanitized_Request, urlencode_postdata
 
 from .helper import FakeYDL
@@ -148,6 +150,31 @@ def do_GET(self):
             self.send_header('Location', new_url)
             self.send_header('Content-Length', '0')
             self.end_headers()
+        elif self.path == '/content-encoding':
+            encodings = self.headers.get('ytdl-encoding', '')
+            payload = b'<html><video src="/vid.mp4" /></html>'
+            for encoding in filter(None, (e.strip() for e in encodings.split(','))):
+                if encoding == 'br' and brotli:
+                    payload = brotli.compress(payload)
+                elif encoding == 'gzip':
+                    buf = io.BytesIO()
+                    with gzip.GzipFile(fileobj=buf, mode='wb') as f:
+                        f.write(payload)
+                    payload = buf.getvalue()
+                elif encoding == 'deflate':
+                    payload = zlib.compress(payload)
+                elif encoding == 'unsupported':
+                    payload = b'raw'
+                    break
+                else:
+                    self._status(415)
+                    return
+            self.send_response(200)
+            self.send_header('Content-Encoding', encodings)
+            self.send_header('Content-Length', str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+
         else:
             self._status(404)
 
@@ -302,6 +329,55 @@ def test_gzip_trailing_garbage(self):
             data = ydl.urlopen(sanitized_Request(f'http://localhost:{self.http_port}/trailing_garbage')).read().decode('utf-8')
             self.assertEqual(data, '<html><video src="/vid.mp4" /></html>')
 
+    @unittest.skipUnless(brotli, 'brotli support is not installed')
+    def test_brotli(self):
+        with FakeYDL() as ydl:
+            res = ydl.urlopen(
+                sanitized_Request(
+                    f'http://127.0.0.1:{self.http_port}/content-encoding',
+                    headers={'ytdl-encoding': 'br'}))
+            self.assertEqual(res.headers.get('Content-Encoding'), 'br')
+            self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+    def test_deflate(self):
+        with FakeYDL() as ydl:
+            res = ydl.urlopen(
+                sanitized_Request(
+                    f'http://127.0.0.1:{self.http_port}/content-encoding',
+                    headers={'ytdl-encoding': 'deflate'}))
+            self.assertEqual(res.headers.get('Content-Encoding'), 'deflate')
+            self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+    def test_gzip(self):
+        with FakeYDL() as ydl:
+            res = ydl.urlopen(
+                sanitized_Request(
+                    f'http://127.0.0.1:{self.http_port}/content-encoding',
+                    headers={'ytdl-encoding': 'gzip'}))
+            self.assertEqual(res.headers.get('Content-Encoding'), 'gzip')
+            self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+    def test_multiple_encodings(self):
+        # https://www.rfc-editor.org/rfc/rfc9110.html#section-8.4
+        with FakeYDL() as ydl:
+            for pair in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
+                res = ydl.urlopen(
+                    sanitized_Request(
+                        f'http://127.0.0.1:{self.http_port}/content-encoding',
+                        headers={'ytdl-encoding': pair}))
+                self.assertEqual(res.headers.get('Content-Encoding'), pair)
+                self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+    def test_unsupported_encoding(self):
+        # it should return the raw content
+        with FakeYDL() as ydl:
+            res = ydl.urlopen(
+                sanitized_Request(
+                    f'http://127.0.0.1:{self.http_port}/content-encoding',
+                    headers={'ytdl-encoding': 'unsupported'}))
+            self.assertEqual(res.headers.get('Content-Encoding'), 'unsupported')
+            self.assertEqual(res.read(), b'raw')
+
 
 class TestClientCert(unittest.TestCase):
     def setUp(self):