add templating for hls-segment-key-uri option #2821
Conversation
|
satisfies #2806 |
8420417 to
9c7ade8
Compare
Codecov Report
@@ Coverage Diff @@
## master #2821 +/- ##
==========================================
+ Coverage 52.93% 52.95% +0.01%
==========================================
Files 248 248
Lines 15537 15542 +5
==========================================
+ Hits 8225 8230 +5
Misses 7312 7312 |
9c7ade8 to
93caf72
Compare
|
You should import urlparse from streamlink/src/streamlink/compat.py Lines 35 to 43 in 8ef3c73 Regarding the implementation, why do we need two parameters for the replacement of segment keys? I think the already existing one could be improved, so that it can accept a string with Tests also have to be added before changes like this can be merged. |
|
Thanks for the feedback! I saw the 2.7 test fail for that same reason and it has been updated in latest push. Edit: I see, I will import from compat The new option doesn't specify what the exact URI will be, since the segment URI can change throughout a stream. We won't know ahead of time what the full I'll definitely add a test for this once we land on the optimal solution. |
93caf72 to
3ab11f1
Compare
|
use only one command, don't add a 2nd here is some unfinished code, might help you with it from base64 import b64encode
from streamlink.compat import urlparse
from streamlink.utils import LazyFormatter
if self.key_uri_override:
p = urlparse(key.uri)
key_uri = LazyFormatter.format(
self.key_uri_override,
url=key.uri,
url_base64=b64encode(key.uri.encode('utf-8')).decode('utf-8'),
scheme=p.scheme,
netloc=p.netloc,
path=p.path,
query=p.query,
)
else:
key_uri = key.urithis would allow |
|
Ahh, very nifty! I'll leverage LazyFormatter, thanks! |
3ab11f1 to
a78a04e
Compare
|
@back-to That worked perfectly. I still need to incorporate a test for this. |
|
How would you guys write a test for this? The HLS Stream is already well tested, this change just impacts the key URI that is requested, and an error is logged when that decryption fails. Should the functionality of I've tested |
|
You could add a test like this. Not sure if it's a bit too much though. Also not sure about the diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index 61db631e..26b35733 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -59,13 +59,14 @@ class TestHLS(unittest.TestCase):
return playlist + playlistEnd
- def start_streamlink(self, masterPlaylist, kwargs=None):
+ def start_streamlink(self, masterPlaylist, hls_segment_key_uri=None, kwargs=None):
kwargs = kwargs or {}
log.info("Executing streamlink")
streamlink = Streamlink()
# Set to default value to avoid a test fail if the default change
streamlink.set_option("hls-live-edge", 3)
+ streamlink.set_option("hls-segment-key-uri", hls_segment_key_uri)
masterStream = hls.HLSStream.parse_variant_playlist(streamlink, masterPlaylist, **kwargs)
stream = masterStream["1080p (source)"].open()
@@ -86,13 +87,13 @@ class TestHLS(unittest.TestCase):
# Start streamlink on the generated stream
streamlinkResult = self.start_streamlink("http://mocked/path/master.m3u8",
- {'start_offset': 1, 'duration': 1})
+ kwargs={'start_offset': 1, 'duration': 1})
# Check result, each segment is 1 second, with duration=1 only one segment should be returned
expectedResult = b''.join(streams[1:2])
self.assertEqual(streamlinkResult, expectedResult)
- def test_hls_encryted_aes128(self):
+ def test_hls_encrypted_aes128(self):
# Encryption parameters
aesKey = os.urandom(16)
aesIv = os.urandom(16)
@@ -122,6 +123,38 @@ class TestHLS(unittest.TestCase):
expectedResult = b''.join(clearStreams[1:] + clearStreams)
self.assertEqual(streamlinkResult, expectedResult)
+ def test_hls_encrypted_aes128_key_uri_override(self):
+ aesKey = os.urandom(16)
+ aesIv = os.urandom(16)
+ aesKeyInvalid = bytes([ord(aesKey[i:i + 1]) ^ 0xFF for i in range(16)])
+ clearStreams = [os.urandom(1024) for i in range(4)]
+ encryptedStreams = [encrypt(clearStream, aesKey, aesIv) for clearStream in clearStreams]
+
+ masterPlaylist = self.getMasterPlaylist()
+ playlist1 = self.getPlaylist(aesIv, "stream{0}.ts.enc")
+ playlist2 = self.getPlaylist(aesIv, "stream2_{0}.ts.enc") + "#EXT-X-ENDLIST\n"
+
+ mocked_key_uri = None
+ streamlinkResult = None
+ with requests_mock.Mocker() as mock:
+ mock.get("http://mocked/path/master.m3u8", text=masterPlaylist)
+ mock.get("http://mocked/path/playlist.m3u8", [{'text': playlist1}, {'text': playlist2}])
+ mock.get("http://mocked/path/encryption_key.key", content=aesKeyInvalid)
+ mocked_key_uri = mock.get("http://real-mocked/path/encryption_key.key", content=aesKey)
+ for i, encryptedStream in enumerate(encryptedStreams):
+ mock.get("http://mocked/path/stream{0}.ts.enc".format(i), content=encryptedStream)
+ for i, encryptedStream in enumerate(encryptedStreams):
+ mock.get("http://mocked/path/stream2_{0}.ts.enc".format(i), content=encryptedStream)
+
+ streamlinkResult = self.start_streamlink(
+ "http://mocked/path/master.m3u8",
+ hls_segment_key_uri="{scheme}://real-{netloc}{path}{query}"
+ )
+
+ self.assertTrue(mocked_key_uri.called)
+ expectedResult = b''.join(clearStreams[1:] + clearStreams)
+ self.assertEqual(streamlinkResult, expectedResult)
+
@patch('streamlink.stream.hls.FFMPEGMuxer.is_usable', Mock(return_value=True))
class TestHlsExtAudio(unittest.TestCase): |
|
That's what I was thinking of last night, but was in the same boat about thinking it was probably too much. Though, it does the job. I removed the |
bastimeyer
left a comment
There was a problem hiding this comment.
I removed the
aesKeyInvalidline and mock
Why? This is part of the validity of the test, in case the create_decryptor logic gets changed.
I suggested the aesKeyInvalid key as inverted real key, because an invalid key needs to be of the same size, otherwise the test fails at the wrong place (important for debugging reasons). What I meant with unsure was whether there's a better way to invert this, as it requires an unnecessary "workaround" for Py3 to also make it work in Py2. I guess using urandom here is not ideal because of this, but changing that would make reading the tests a bit more difficult, as this test is bascially an extended copy of the one above.
Not mocking the default URI is also problematic, because in case the logic changes, an actual network request could be made, and we don't want that. What I forgot to add was an assertion that the default URI does not get called, which should definitely be done here.
diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index 9660de30..0fc1cd1a 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -126,6 +126,7 @@ class TestHLS(unittest.TestCase):
def test_hls_encrypted_aes128_key_uri_override(self):
aesKey = os.urandom(16)
aesIv = os.urandom(16)
+ aesKeyInvalid = bytes([ord(aesKey[i:i + 1]) ^ 0xFF for i in range(16)])
clearStreams = [os.urandom(1024) for i in range(4)]
encryptedStreams = [encrypt(clearStream, aesKey, aesIv) for clearStream in clearStreams]
@@ -133,12 +134,14 @@ class TestHLS(unittest.TestCase):
playlist1 = self.getPlaylist(aesIv, "stream{0}.ts.enc")
playlist2 = self.getPlaylist(aesIv, "stream2_{0}.ts.enc") + "#EXT-X-ENDLIST\n"
- mocked_key_uri = None
+ mocked_key_uri_default = None
+ mocked_key_uri_override = None
streamlinkResult = None
with requests_mock.Mocker() as mock:
mock.get("http://mocked/path/master.m3u8", text=masterPlaylist)
mock.get("http://mocked/path/playlist.m3u8", [{'text': playlist1}, {'text': playlist2}])
- mocked_key_uri = mock.get("http://real-mocked/path/encryption_key.key", content=aesKey)
+ mocked_key_uri_default = mock.get("http://mocked/path/encryption_key.key", content=aesKeyInvalid)
+ mocked_key_uri_override = mock.get("http://real-mocked/path/encryption_key.key", content=aesKey)
for i, encryptedStream in enumerate(encryptedStreams):
mock.get("http://mocked/path/stream{0}.ts.enc".format(i), content=encryptedStream)
for i, encryptedStream in enumerate(encryptedStreams):
@@ -149,7 +152,8 @@ class TestHLS(unittest.TestCase):
hls_segment_key_uri="{scheme}://real-{netloc}{path}{query}"
)
- self.assertTrue(mocked_key_uri.called)
+ self.assertFalse(mocked_key_uri_default.called)
+ self.assertTrue(mocked_key_uri_override.called)
expectedResult = b''.join(clearStreams[1:] + clearStreams)
self.assertEqual(streamlinkResult, expectedResult)
|
Damn, I didn't think through those scenarios. Sorry, I don't have much experience with writing tests, so I appreciate your helpful explanations of everything! This feels like a lot to properly isolate and test "using the right key URI" and "does our template string work". Would it be better to create a new function in the module, like the following: def key_uri(key, key_uri_override=None):
if key_uri_override:
p = urlparse(key.uri)
key_uri = LazyFormatter.format(
self.key_uri_override,
url=key.uri,
scheme=p.scheme,
netloc=p.netloc,
path=p.path,
query=p.query,
)
else:
key_uri = key.uri
return key_uriAnd just test this against our hls-segment-key-uri templated string? Edit: I guess that doesn't stop someone from still removing that from the |
a1e5574 to
781b444
Compare
bastimeyer
left a comment
There was a problem hiding this comment.
Should be fine now. Unless @back-to has something to say regarding the test, we can merge this.
781b444 to
f038aea
Compare
|
@back-to I've removed the unnecessary bracket escaping / format |
This will replace the host in the segment URI with the supplied HOST