Repository: streamlink/streamlink
Pull request #3770 (2021-06-02 22:08:08)

stream.hls: replace custom PKCS#7 unpad function
- replace stream.hls.pkcs7_decode with Crypto.Util.Padding.unpad
- add tests for invalid encryption data

----

This is what I had originally suggested in #3446 but it keeps the garbage data removal.


Patch:
diff --git a/src/streamlink/plugin/api/http_session.py b/src/streamlink/plugin/api/http_session.py
index d56739be..6cd3764c 100644
--- a/src/streamlink/plugin/api/http_session.py
+++ b/src/streamlink/plugin/api/http_session.py
@@ -1,5 +1,7 @@
 import time
 
+import requests.adapters
+import urllib3
 from requests import Session
 
 from streamlink.exceptions import PluginError
@@ -7,16 +9,39 @@ from streamlink.packages.requests_file import FileAdapter
 from streamlink.plugin.api import useragents
 from streamlink.utils import parse_json, parse_xml
 
-try:
-    from requests.packages import urllib3
 
+try:
     # We tell urllib3 to disable warnings about unverified HTTPS requests,
     # because in some plugins we have to do unverified requests intentionally.
     urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-except (ImportError, AttributeError):
+except AttributeError:
     pass
 
-__all__ = ["HTTPSession"]
+
+class _HTTPResponse(urllib3.response.HTTPResponse):
+    def __init__(self, *args, **kwargs):
+        # Always enforce content length validation!
+        # This fixes a bug in requests which doesn't raise errors on HTTP responses where
+        # the "Content-Length" header doesn't match the response's body length.
+        # https://github.com/psf/requests/issues/4956#issuecomment-573325001
+        #
+        # Summary:
+        # This bug is related to urllib3.response.HTTPResponse.stream() which calls urllib3.response.HTTPResponse.read() as
+        # a wrapper for http.client.HTTPResponse.read(amt=...), where no http.client.IncompleteRead exception gets raised
+        # due to "backwards compatiblity" of an old bug if a specific amount is attempted to be read on an incomplete response.
+        #
+        # urllib3.response.HTTPResponse.read() however has an additional check implemented via the enforce_content_length
+        # parameter, but it doesn't check by default and requests doesn't set the parameter for enabling it either.
+        #
+        # Fix this by overriding urllib3.response.HTTPResponse's constructor and always setting enforce_content_length to True,
+        # as there is no way to make requests set this parameter on its own.
+        kwargs.update({"enforce_content_length": True})
+        super().__init__(*args, **kwargs)
+
+
+# override all urllib3.response.HTTPResponse references in requests.adapters.HTTPAdapter.send
+urllib3.connectionpool.HTTPConnectionPool.ResponseCls = _HTTPResponse
+requests.adapters.HTTPResponse = _HTTPResponse
 
 
 def _parse_keyvalue_list(val):
@@ -29,12 +54,10 @@ def _parse_keyvalue_list(val):
 
 
 class HTTPSession(Session):
-    def __init__(self, *args, **kwargs):
-        Session.__init__(self, *args, **kwargs)
-
-        if self.headers['User-Agent'].startswith('python-requests'):
-            self.headers['User-Agent'] = useragents.FIREFOX
+    def __init__(self):
+        super().__init__()
 
+        self.headers['User-Agent'] = useragents.FIREFOX
         self.timeout = 20.0
 
         self.mount('file://', FileAdapter())
@@ -123,12 +146,16 @@ class HTTPSession(Session):
 
         while True:
             try:
-                res = Session.request(self, method, url,
-                                      headers=headers,
-                                      params=params,
-                                      timeout=timeout,
-                                      proxies=proxies,
-                                      *args, **kwargs)
+                res = super().request(
+                    method,
+                    url,
+                    headers=headers,
+                    params=params,
+                    timeout=timeout,
+                    proxies=proxies,
+                    *args,
+                    **kwargs
+                )
                 if raise_for_status and res.status_code not in acceptable_status:
                     res.raise_for_status()
                 break
@@ -136,8 +163,7 @@ class HTTPSession(Session):
                 raise
             except Exception as rerr:
                 if retries >= total_retries:
-                    err = exception("Unable to open URL: {url} ({err})".format(url=url,
-                                                                               err=rerr))
+                    err = exception(f"Unable to open URL: {url} ({rerr})")
                     err.err = rerr
                     raise err
                 retries += 1
@@ -150,3 +176,6 @@ class HTTPSession(Session):
             res = schema.validate(res.text, name="response text", exception=PluginError)
 
         return res
+
+
+__all__ = ["HTTPSession"]
diff --git a/src/streamlink/plugins/twitch.py b/src/streamlink/plugins/twitch.py
index 4ef7f1d1..47a8ebe5 100644
--- a/src/streamlink/plugins/twitch.py
+++ b/src/streamlink/plugins/twitch.py
@@ -513,7 +513,7 @@ class Twitch(Plugin):
             (?:
                 /video/(?P<video_id>\d+)
                 |
-                /clip/(?P<clip_name>[\w]+)
+                /clip/(?P<clip_name>[\w-]+)
             )?
         )
     """, re.VERBOSE)
diff --git a/src/streamlink/plugins/zattoo.py b/src/streamlink/plugins/zattoo.py
index b0fbf4ac..bf527f7b 100644
--- a/src/streamlink/plugins/zattoo.py
+++ b/src/streamlink/plugins/zattoo.py
@@ -17,7 +17,6 @@ log = logging.getLogger(__name__)
 class Zattoo(Plugin):
     API_CHANNELS = '{0}/zapi/v2/cached/channels/{1}?details=False'
     API_HELLO = '{0}/zapi/session/hello'
-    API_HELLO_V2 = '{0}/zapi/v2/session/hello'
     API_HELLO_V3 = '{0}/zapi/v3/session/hello'
     API_LOGIN = '{0}/zapi/v2/account/login'
     API_LOGIN_V3 = '{0}/zapi/v3/account/login'
@@ -157,19 +156,16 @@ class Zattoo(Plugin):
         # a new session is required for the app_token
         self.session.http.cookies = cookiejar_from_dict({})
         if self.base_url == 'https://zattoo.com':
-            app_token_url = 'https://zattoo.com/client/token-2fb69f883fea03d06c68c6e5f21ddaea.json'
+            app_token_url = 'https://zattoo.com/token-46a1dfccbd4c3bdaf6182fea8f8aea3f.json'
         elif self.base_url == 'https://www.quantum-tv.com':
             app_token_url = 'https://www.quantum-tv.com/token-4d0d61d4ce0bf8d9982171f349d19f34.json'
         else:
             app_token_url = self.base_url
 
         res = self.session.http.get(app_token_url)
-        if self.base_url == 'https://www.quantum-tv.com':
+        if self.base_url == 'https://www.quantum-tv.com' or self.base_url == 'https://zattoo.com':
             app_token = self.session.http.json(res)["session_token"]
             hello_url = self.API_HELLO_V3.format(self.base_url)
-        elif self.base_url == 'https://zattoo.com':
-            app_token = self.session.http.json(res)['app_tid']
-            hello_url = self.API_HELLO_V2.format(self.base_url)
         else:
             match = self._app_token_re.search(res.text)
             app_token = match.group(1)
@@ -182,19 +178,12 @@ class Zattoo(Plugin):
             self._session_attributes.set(
                 'uuid', __uuid, expires=self.TIME_SESSION)
 
-        if self.base_url == 'https://zattoo.com':
-            params = {
-                'uuid': __uuid,
-                'app_tid': app_token,
-                'app_version': '1.0.0'
-            }
-        else:
-            params = {
-                'client_app_token': app_token,
-                'uuid': __uuid,
-            }
+        params = {
+            'client_app_token': app_token,
+            'uuid': __uuid,
+        }
 
-        if self.base_url == 'https://www.quantum-tv.com':
+        if self.base_url == 'https://www.quantum-tv.com' or self.base_url == 'https://zattoo.com':
             params['app_version'] = '3.2028.3'
         else:
             params['lang'] = 'en'
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index bfea5295..ddc5615a 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -6,6 +6,7 @@ from threading import Event
 from urllib.parse import urlparse
 
 from Crypto.Cipher import AES
+from Crypto.Util.Padding import unpad
 from requests.exceptions import ChunkedEncodingError
 
 from streamlink.exceptions import StreamError
@@ -23,18 +24,6 @@ def num_to_iv(n):
     return struct.pack(">8xq", n)
 
 
-def pkcs7_decode(paddedData, keySize=16):
-    '''
-    Remove the PKCS#7 padding
-    '''
-    # Use ord + [-1:] to support both python 2 and 3
-    val = ord(paddedData[-1:])
-    if val > keySize:
-        raise StreamError("Input is not padded or padding is corrupt, got padding size of {0}".format(val))
-
-    return paddedData[:-val]
-
-
 class HLSStreamWriter(SegmentedStreamWriter):
     def __init__(self, reader, *args, **kwargs):
         options = reader.stream.session.options
@@ -156,21 +145,21 @@ class HLSStreamWriter(SegmentedStreamWriter):
 
             data = res.content
             # If the input data is not a multiple of 16, cut off any garbage
-            garbage_len = len(data) % 16
+            garbage_len = len(data) % AES.block_size
             if garbage_len:
                 log.debug(f"Cutting off {garbage_len} bytes of garbage before decrypting")
                 decrypted_chunk = decryptor.decrypt(data[:-garbage_len])
             else:
                 decrypted_chunk = decryptor.decrypt(data)
 
-            self.reader.buffer.write(pkcs7_decode(decrypted_chunk))
+            chunk = unpad(decrypted_chunk, AES.block_size, style="pkcs7")
+            self.reader.buffer.write(chunk)
         else:
             try:
                 for chunk in res.iter_content(chunk_size):
                     self.reader.buffer.write(chunk)
             except ChunkedEncodingError:
                 log.error(f"Download of segment {sequence.num} failed")
-
                 return
 
         log.debug(f"Download of segment {sequence.num} complete")


Test patch:
diff --git a/tests/mixins/stream_hls.py b/tests/mixins/stream_hls.py
index 88d3b5f7..165b8e8c 100644
--- a/tests/mixins/stream_hls.py
+++ b/tests/mixins/stream_hls.py
@@ -87,6 +87,7 @@ class EventedHLSStreamWriter(_HLSStreamWriter):
         super().__init__(*args, **kwargs)
         self.write_wait = Event()
         self.write_done = Event()
+        self.write_error = None
 
     def write(self, *args, **kwargs):
         # only write once per step
@@ -97,6 +98,9 @@ class EventedHLSStreamWriter(_HLSStreamWriter):
             # don't write again during teardown
             if not self.closed:
                 super().write(*args, **kwargs)
+        except Exception as err:
+            self.write_error = err
+            self.reader.close()
         finally:
             # notify main thread that writing has finished
             self.write_done.set()
@@ -223,6 +227,8 @@ class TestMixinStreamHLS(unittest.TestCase):
             writer.write_wait.set()
             writer.write_done.wait(timeout)
             writer.write_done.clear()
+            if writer.write_error:
+                raise writer.write_error
 
     # make one read call on the read thread and wait until it has finished
     def await_read(self, read_all=False, timeout=5):
diff --git a/tests/plugins/test_twitch.py b/tests/plugins/test_twitch.py
index 50abd993..7d26639b 100644
--- a/tests/plugins/test_twitch.py
+++ b/tests/plugins/test_twitch.py
@@ -18,6 +18,7 @@ class TestPluginCanHandleUrlTwitch(PluginCanHandleUrl):
         'https://www.twitch.tv/twitch',
         'https://www.twitch.tv/videos/150942279',
         'https://clips.twitch.tv/ObservantBenevolentCarabeefPhilosoraptor',
+        'https://www.twitch.tv/weplaydota/clip/FurryIntelligentDonutAMPEnergyCherry-akPRxv7Y3w58WmFq'
         'https://www.twitch.tv/twitch/video/292713971',
         'https://www.twitch.tv/twitch/v/292713971',
     ]
diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index a974c2e1..21e765b7 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -1,28 +1,18 @@
 import os
 import unittest
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, call, patch
 
 import pytest
 import requests_mock
 from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad
 
 from streamlink.session import Streamlink
 from streamlink.stream import hls
-from tests.mixins.stream_hls import Playlist, Segment, Tag, TestMixinStreamHLS
+from tests.mixins.stream_hls import EventedHLSStreamWriter, Playlist, Segment, Tag, TestMixinStreamHLS
 from tests.resources import text
 
 
-def pkcs7_encode(data, keySize):
-    val = keySize - (len(data) % keySize)
-    return b''.join([data, bytes(bytearray(val * [val]))])
-
-
-def encrypt(data, key, iv):
-    aesCipher = AES.new(key, AES.MODE_CBC, iv)
-    encrypted_data = aesCipher.encrypt(pkcs7_encode(data, len(key)))
-    return encrypted_data
-
-
 class TagKey(Tag):
     path = "encryption.key"
 
@@ -44,10 +34,12 @@ class TagKey(Tag):
 
 
 class SegmentEnc(Segment):
-    def __init__(self, num, key, iv, *args, **kwargs):
+    def __init__(self, num, key, iv, *args, padding=b"", append=b"", **kwargs):
         super().__init__(num, *args, **kwargs)
+        aesCipher = AES.new(key, AES.MODE_CBC, iv)
+        padded = self.content + padding if padding else pad(self.content, AES.block_size, style="pkcs7")
         self.content_plain = self.content
-        self.content = encrypt(self.content, key, iv)
+        self.content = aesCipher.encrypt(padded) + append
 
 
 class TestHLSStreamRepr(unittest.TestCase):
@@ -90,6 +82,14 @@ class TestHLSVariantPlaylist(unittest.TestCase):
         )
 
 
+class EventedHLSReader(hls.HLSStreamReader):
+    __writer__ = EventedHLSStreamWriter
+
+
+class EventedHLSStream(hls.HLSStream):
+    __reader__ = EventedHLSReader
+
+
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
 class TestHLSStream(TestMixinStreamHLS, unittest.TestCase):
     def get_session(self, options=None, *args, **kwargs):
@@ -111,6 +111,8 @@ class TestHLSStream(TestMixinStreamHLS, unittest.TestCase):
 
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
 class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
+    __stream__ = EventedHLSStream
+
     def get_session(self, options=None, *args, **kwargs):
         session = super().get_session(options)
         session.set_option("hls-live-edge", 3)
@@ -136,6 +138,7 @@ class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
             Playlist(4, [key] + [SegmentEnc(num, aesKey, aesIv) for num in range(4, 8)], end=True)
         ])
 
+        self.await_write(3 + 4)
         data = self.await_read(read_all=True)
         expected = self.content(segments, prop="content_plain", cond=lambda s: s.num >= 1)
         self.assertEqual(data, expected, "Decrypts the AES-128 identity stream")
@@ -156,6 +159,7 @@ class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
             Playlist(4, [key_invalid] + [SegmentEnc(num, aesKey, aesIv) for num in range(4, 8)], end=True)
         ], options={"hls-segment-key-uri": "{scheme}://real-{netloc}{path}?{query}"})
 
+        self.await_write(3 + 4)
         data = self.await_read(read_all=True)
         expected = self.content(segments, prop="content_plain", cond=lambda s: s.num >= 1)
         self.assertEqual(data, expected, "Decrypts stream from custom key")
@@ -163,6 +167,49 @@ class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
         self.assertTrue(self.called(key), "Downloads custom encryption key")
         self.assertEqual(self.get_mock(key).last_request._request.headers.get("X-FOO"), "BAR")
 
+    @patch("streamlink.stream.hls.log")
+    def test_hls_encrypted_aes128_incorrect_block_length(self, mock_log):
+        aesKey, aesIv, key = self.gen_key()
+
+        # noinspection PyTypeChecker
+        thread, segments = self.subject([
+            Playlist(0, [key] + [
+                SegmentEnc(0, aesKey, aesIv, append=b"?" * 1),
+                SegmentEnc(1, aesKey, aesIv, append=b"?" * (AES.block_size - 1))
+            ], end=True)
+        ])
+
+        self.await_write(2)
+        data = self.await_read(read_all=True)
+        expected = self.content(segments, prop="content_plain")
+        self.assertEqual(data, expected, "Removes garbage data from segments")
+        self.assertIn(call("Cutting off 1 bytes of garbage before decrypting"), mock_log.debug.mock_calls)
+        self.assertIn(call("Cutting off 15 bytes of garbage before decrypting"), mock_log.debug.mock_calls)
+
+    def test_hls_encrypted_aes128_incorrect_padding_length(self):
+        aesKey, aesIv, key = self.gen_key()
+
+        padding = b"\x00" * (AES.block_size - len(b"[0]"))
+        self.subject([
+            Playlist(0, [key, SegmentEnc(0, aesKey, aesIv, padding=padding)], end=True)
+        ])
+
+        with self.assertRaises(ValueError) as cm:
+            self.await_write()
+        self.assertEqual(str(cm.exception), "Padding is incorrect.", "Crypto.Util.Padding.unpad exception")
+
+    def test_hls_encrypted_aes128_incorrect_padding_content(self):
+        aesKey, aesIv, key = self.gen_key()
+
+        padding = (b"\x00" * (AES.block_size - len(b"[0]") - 1)) + bytes([AES.block_size])
+        self.subject([
+            Playlist(0, [key, SegmentEnc(0, aesKey, aesIv, padding=padding)], end=True)
+        ])
+
+        with self.assertRaises(ValueError) as cm:
+            self.await_write()
+        self.assertEqual(str(cm.exception), "PKCS#7 padding is incorrect.", "Crypto.Util.Padding.unpad exception")
+
 
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
 @patch("streamlink.stream.hls.HLSStreamWriter.run", Mock(return_value=True))
diff --git a/tests/test_api_http_session.py b/tests/test_api_http_session.py
index 694e435f..3b692a1a 100644
--- a/tests/test_api_http_session.py
+++ b/tests/test_api_http_session.py
@@ -5,23 +5,29 @@ import requests
 
 from streamlink.exceptions import PluginError
 from streamlink.plugin.api.http_session import HTTPSession
+from streamlink.plugin.api.useragents import FIREFOX
 
 
 class TestPluginAPIHTTPSession(unittest.TestCase):
+    def test_session_init(self):
+        session = HTTPSession()
+        self.assertEqual(session.headers.get("User-Agent"), FIREFOX)
+        self.assertEqual(session.timeout, 20.0)
+        self.assertIn("file://", session.adapters.keys())
+
     @patch("streamlink.plugin.api.http_session.time.sleep")
-    @patch("streamlink.plugin.api.http_session.Session.request")
+    @patch("streamlink.plugin.api.http_session.Session.request", side_effect=requests.Timeout)
     def test_read_timeout(self, mock_request, mock_sleep):
-        mock_request.side_effect = requests.Timeout
         session = HTTPSession()
 
         with self.assertRaises(PluginError) as cm:
             session.get("http://localhost/", timeout=123, retries=3, retry_backoff=2, retry_max_backoff=5)
         self.assertTrue(str(cm.exception).startswith("Unable to open URL: http://localhost/"))
         self.assertEqual(mock_request.mock_calls, [
-            call(session, "GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
-            call(session, "GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
-            call(session, "GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
-            call(session, "GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
+            call("GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
+            call("GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
+            call("GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
+            call("GET", "http://localhost/", headers={}, params={}, timeout=123, proxies={}, allow_redirects=True),
         ])
         self.assertEqual(mock_sleep.mock_calls, [
             call(2),