[ie] Detect invalid m3u8 playlist data (#13563)

Authored by: Grub4K

test/test_InfoExtractor.py

@@ -36,6 +36,18 @@ class InfoExtractorTestRequestHandler(http.server.BaseHTTPRequestHandler):
             self.send_header('Content-Type', 'text/html; charset=utf-8')
             self.end_headers()
             self.wfile.write(TEAPOT_RESPONSE_BODY.encode())
+        elif self.path == '/fake.m3u8':
+            self.send_response(200)
+            self.send_header('Content-Length', '1024')
+            self.end_headers()
+            self.wfile.write(1024 * b'\x00')
+        elif self.path == '/bipbop.m3u8':
+            with open('test/testdata/m3u8/bipbop_16x9.m3u8', 'rb') as f:
+                data = f.read()
+            self.send_response(200)
+            self.send_header('Content-Length', str(len(data)))
+            self.end_headers()
+            self.wfile.write(data)
         else:
             assert False
 
@@ -2079,5 +2091,45 @@ jwplayer("mediaplayer").setup({"abouttext":"Visit Indie DB","aboutlink":"http:\/
                 self.ie._search_nuxt_json(HTML_TMPL.format(data), None, default=DEFAULT), DEFAULT)
 
 
+class TestInfoExtractorNetwork(unittest.TestCase):
+    def setUp(self, /):
+        self.httpd = http.server.HTTPServer(
+            ('127.0.0.1', 0), InfoExtractorTestRequestHandler)
+        self.port = http_server_port(self.httpd)
+
+        self.server_thread = threading.Thread(target=self.httpd.serve_forever)
+        self.server_thread.daemon = True
+        self.server_thread.start()
+
+        self.called = False
+
+        def require_warning(*args, **kwargs):
+            self.called = True
+
+        self.ydl = FakeYDL()
+        self.ydl.report_warning = require_warning
+        self.ie = DummyIE(self.ydl)
+
+    def tearDown(self, /):
+        self.ydl.close()
+        self.httpd.shutdown()
+        self.httpd.server_close()
+        self.server_thread.join(1)
+
+    def test_extract_m3u8_formats(self):
+        formats, subtitles = self.ie._extract_m3u8_formats_and_subtitles(
+            f'http://127.0.0.1:{self.port}/bipbop.m3u8', None, fatal=False)
+        self.assertFalse(self.called)
+        self.assertTrue(formats)
+        self.assertTrue(subtitles)
+
+    def test_extract_m3u8_formats_warning(self):
+        formats, subtitles = self.ie._extract_m3u8_formats_and_subtitles(
+            f'http://127.0.0.1:{self.port}/fake.m3u8', None, fatal=False)
+        self.assertTrue(self.called, 'Warning was not issued for binary m3u8 file')
+        self.assertFalse(formats)
+        self.assertFalse(subtitles)
+
+
 if __name__ == '__main__':
     unittest.main()