Repository: streamlink/streamlink
Pull request #3774 (2021-06-04 13:08:56)

plugin.api.validate: add nested lookups to get() and implement union_get()
There are two validation schema patterns which get re-used all the time and which can be improved.

1. **Nested/recursive `validate.get` calls:**
   Nested key lookups currently require three separate `get`-transform calls and one `all` call, which is wasteful and tedious to write. Also, since the `get` method sets its default value to `None` and returns it on any `KeyError`/`IndexError`, error messages in consecutive lookups are not particularly useful.
   ```py
   >>> validate(
   ...     all(get("first"), get("second"), get("third")),
   ...     {"first": {"second": {"third": "value"}}}
   ... )
   "value"

   >>> validate(
   ...     all(get("invalid"), get("second")),
   ...     {"first": {"second": "value"}}
   ... )
   ValueError: "'NoneType' object is not subscriptable"
   ```

   Single `get()` calls with nested lookups improve the readability, only use one single transform and allow having better error messages. The runtime performance difference for non-recursive lookups should be minimal, as it's just a simple iterator over the input that now always gets turned into a tuple.

   ```py
   >>> validate(
   ...     get(("first", "second", "third")),
   ...     {"first": {"second": {"third": "value"}}}
   ... )
   "value"

   >>> validate(
   ...     get(("invalid", "second")),
   ...     {"first": {"second": "value"}}
   ... )
   ValueError: "Object \"{'first': {'second': 'value'}}\" does not have item \"invalid\""

   >>> validate(
   ...     get(("first", "invalid")),
   ...     {"first": {"second": "value"}}
   ... )
   None

   >>> validate(
   ...     get(("first", "second"), strict=True),
   ...     {("first", "second"): "value"}
   ... )
   "value"
   ```

2. **`validate.union(tuple(...))` calls with multiple `validate.get()` calls:**
   The purpose here is to simplify tuples of items that are used for destructuring. The `get`-transforms are created during validation and not during declaration, which can improve overall memory consumption.
   ```py
   >>> validate(
   ...     union((get("first"), get("second"))),
   ...     {"first": "foo", "second": "bar"}
   ... )
   ("foo", "bar")
   ```

   ```py
   >>> validate(
   ...     union_get("first", "second"),
   ...     {"first": "foo", "second": "bar"}
   ... )
   ("foo", "bar")
   ```


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/plugin/api/validate.py b/src/streamlink/plugin/api/validate.py
index 3d2a0385..932affe7 100644
--- a/src/streamlink/plugin/api/validate.py
+++ b/src/streamlink/plugin/api/validate.py
@@ -18,6 +18,7 @@
 
 from copy import copy as copy_obj
 from functools import singledispatch
+from typing import Any, Tuple, Union
 from urllib.parse import urlparse
 from xml.etree import ElementTree as ET
 
@@ -25,7 +26,7 @@ from streamlink.exceptions import PluginError
 
 __all__ = [
     "any", "all", "filter", "get", "getattr", "hasattr", "length", "optional",
-    "transform", "text", "union", "url", "startswith", "endswith", "contains",
+    "transform", "text", "union", "union_get", "url", "startswith", "endswith", "contains",
     "xml_element", "xml_find", "xml_findall", "xml_findtext",
     "validate", "Schema", "SchemaContainer"
 ]
@@ -86,6 +87,12 @@ class attr(SchemaContainer):
     """Validates an object's attributes."""
 
 
+class union_get:
+    def __init__(self, *keys, seq=tuple):
+        self.keys = keys
+        self.seq = seq
+
+
 class xml_element:
     """A XML element."""
 
@@ -140,25 +147,36 @@ def contains(string):
     return contains_str
 
 
-def get(item, default=None):
+def get(item: Union[Any, Tuple[Any]], default: Any = None, strict: bool = False):
     """Get item from value (value[item]).
 
-    If the item is not found, return the default.
+    Unless strict is set to True, item can be a tuple of items for recursive lookups.
+
+    If the item is not found in the last object of a recursive lookup, return the default.
 
     Handles XML elements, regex matches and anything that has __getitem__.
     """
 
-    def getter(value):
-        if ET.iselement(value):
-            value = value.attrib
+    if type(item) is not tuple or strict:
+        item = (item,)
 
+    def getter(value):
+        idx = 0
         try:
-            # Use .group() if this is a regex match object
-            if _is_re_match(value):
-                return value.group(item)
-            else:
-                return value[item]
+            for key in item:
+                if ET.iselement(value):
+                    value = value.attrib
+                # Use .group() if this is a regex match object
+                elif _is_re_match(value):
+                    value = value.group(key)
+                else:
+                    value = value[key]
+                idx += 1
+            return value
         except (KeyError, IndexError):
+            # only return default value on last item in nested lookup
+            if idx < len(item) - 1:
+                raise ValueError(f"Object \"{value}\" does not have item \"{key}\"")
             return default
         except (TypeError, AttributeError) as err:
             raise ValueError(err)
@@ -417,6 +435,11 @@ def validate_attr(schema, value):
     return new
 
 
+@validate.register(union_get)
+def validate_union_from(schema, value):
+    return schema.seq(validate(get(k), value) for k in schema.keys)
+
+
 @singledispatch
 def validate_union(schema, value):
     raise ValueError("Invalid union type: {0}".format(type(schema).__name__))
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),
diff --git a/tests/test_api_validate.py b/tests/test_api_validate.py
index b3f9177e..c2a0800e 100644
--- a/tests/test_api_validate.py
+++ b/tests/test_api_validate.py
@@ -4,7 +4,7 @@ from xml.etree.ElementTree import Element
 
 from streamlink.plugin.api.validate import (
     all, any, attr, endswith, filter, get, getattr, hasattr,
-    length, map, optional, startswith, text, transform, union, url,
+    length, map, optional, startswith, text, transform, union, union_get, url,
     validate, xml_element, xml_find, xml_findall, xml_findtext
 )
 
@@ -41,6 +41,12 @@ class TestPluginAPIValidate(unittest.TestCase):
         assert validate(union((get("foo"), get("bar"))),
                         {"foo": "alpha", "bar": "beta"}) == ("alpha", "beta")
 
+    def test_union_get(self):
+        assert validate(union_get("foo", "bar"), {"foo": "alpha", "bar": "beta"}) == ("alpha", "beta")
+        assert validate(union_get("foo", "bar", seq=list), {"foo": "alpha", "bar": "beta"}) == ["alpha", "beta"]
+        assert validate(union_get(("foo", "bar"), ("baz", "qux")),
+                        {"foo": {"bar": "alpha"}, "baz": {"qux": "beta"}}) == ("alpha", "beta")
+
     def test_list(self):
         assert validate([1, 0], [1, 0, 1, 1]) == [1, 0, 1, 1]
         assert validate([1, 0], []) == []
@@ -84,7 +90,25 @@ class TestPluginAPIValidate(unittest.TestCase):
 
     def test_get(self):
         assert validate(get("key"), {"key": "value"}) == "value"
+        assert validate(get("key"), re.match(r"(?P<key>.+)", "value")) == "value"
+        assert validate(get("invalidkey"), {"key": "value"}) is None
         assert validate(get("invalidkey", "default"), {"key": "value"}) == "default"
+        assert validate(get(3, "default"), [0, 1, 2]) == "default"
+
+        with self.assertRaisesRegex(ValueError, "'NoneType' object is not subscriptable"):
+            validate(get("key"), None)
+
+        data = {"one": {"two": {"three": "value1"}},
+                ("one", "two", "three"): "value2"}
+        assert validate(get(("one", "two", "three")), data) == "value1", "Recursive lookup"
+        assert validate(get(("one", "two", "three"), strict=True), data) == "value2", "Strict tuple-key lookup"
+        assert validate(get(("one", "two", "invalidkey")), data) is None, "Default value is None"
+        assert validate(get(("one", "two", "invalidkey"), "default"), data) == "default", "Custom default value"
+
+        with self.assertRaisesRegex(ValueError, "Object \"{'two': {'three': 'value1'}}\" does not have item \"invalidkey\""):
+            validate(get(("one", "invalidkey", "three")), data)
+        with self.assertRaisesRegex(ValueError, "'NoneType' object is not subscriptable"):
+            validate(all(get("one"), get("invalidkey"), get("three")), data)
 
     def test_get_re(self):
         m = re.match(r"(\d+)p", "720p")