VTT Reader: handle WebVTT files with bad signatures

src/main/python/ttconv/tt.py

@@ -172,6 +172,9 @@ def progress_callback_write(percent_progress: float):
     }
   )
 
+def die(msg: str):
+  LOGGER.error(msg)
+  sys.exit(msg)
 
 class FileTypes(Enum):
   '''Enumerates the types of supported'''
@@ -361,8 +364,13 @@ def convert(args):
     else:
       exit_str = f'Input file {args.input} is not supported'
 
-    LOGGER.error(exit_str)
-    sys.exit(exit_str)
+    die(exit_str)
+
+  #
+  # handle the case where the input file could not be read into the model
+  #
+  if model is None:
+    die("Aborting due to invalid input file contents.")
 
   #
   # apply document language
@@ -466,8 +474,7 @@ def convert(args):
     else:
       exit_str = f'Output file is {args.output} is not supported'
 
-    LOGGER.error(exit_str)
-    sys.exit(exit_str)
+    die(exit_str)
 
 
 # Ensure that the handler is added only once/globally

src/main/python/ttconv/vtt/reader.py

@@ -505,7 +505,7 @@ def _get_or_make_region(
   return found_region
 
 _VTT_TS_RE = re.compile(r"(?:(?P<hh>[0-9]{2,}):)?(?P<mm>[0-9]{2}):(?P<ss>[0-9]{2})\.(?P<ms>[0-9]{3})")
-_VTT_TS_TAG_RE = re.compile(r"<((?:[0-9]{2,3}:)?[0-9]{2}:[0-9]{2}\.[0-9]{3})>")
+_VTT_FIRST_LINE_RE = re.compile(r"WEBVTT([\n\t ].*)?")
 
 def vtt_timestamp_to_secs(vtt_ts: str):
   m = _VTT_TS_RE.fullmatch(vtt_ts)
@@ -518,8 +518,9 @@ def vtt_timestamp_to_secs(vtt_ts: str):
 
   return None
 
-def to_model(data_file: typing.IO, _config = None, progress_callback=lambda _: None):
-  """Converts a WebVTT document to the data model"""
+def to_model(data_file: typing.IO, _config = None, progress_callback=lambda _: None) -> typing.Optional[model.ContentDocument]:
+  """Converts a WebVTT document to the data model. Returns `None` if the
+  document does not start with the correct WebVTT file signature."""
 
   class _State(Enum):
     LOOKING = 1
@@ -530,25 +531,25 @@ class _State(Enum):
     NOTE = 6
     STYLE = 7
 
-
-  doc = model.ContentDocument()
-
-  body = model.Body(doc)
-  doc.set_body(body)
-
-  div = model.Div(doc)
-  body.push_child(div)
-
-  lines : str = data_file.readlines()
+  # see https://www.w3.org/TR/webvtt1/#file-parsing
+  lines = data_file.read().replace("\u0000", "\uFFFD") \
+                          .replace("\u000D\u000A", "\u000A") \
+                          .replace("\u000D", "\u000A") \
+                          .split("\u000A")
 
   state = _State.START
-  current_p = None
+  doc = None              # output document
+  div = None              # div into which cues will be inserted as p's
+  current_p = None        # current p
 
   for line_index, line in enumerate(_none_terminated(lines)):
 
     if state is _State.START:
-      if not line.startswith("WEBVTT"):
-        LOGGER.warning("The first line of the file does not start with WEBVTT")
+      if not _VTT_FIRST_LINE_RE.fullmatch(line):
+        LOGGER.error("The first line of the file does not start with WEBVTT")
+        break
+      doc = model.ContentDocument()
+      doc.set_body(model.Body(doc))
       state = _State.LOOKING
       continue
 
@@ -610,15 +611,15 @@ class _State(Enum):
       current_p.set_region(_get_or_make_region(doc, cue_params[3:]))
 
       state = _State.TEXT
-      subtitle_text = None
+      subtitle_lines = []
       continue
 
     if state in (_State.TEXT, _State.TEXT_MORE):
 
-      if line is None or _EMPTY_RE.fullmatch(line):
-        if subtitle_text is not None:
+      if line is None or len(line) == 0:
+        if len(subtitle_lines) > 0:
           _parse_cue_text(
-            subtitle_text.strip('\r\n').replace(r"\n\r", "\n"),
+            "\n".join(subtitle_lines),
             current_p,
             line_index
           )
@@ -629,10 +630,12 @@ class _State(Enum):
         continue
 
       if state is _State.TEXT:
+        if div is None:
+          div = model.Div(doc)
+          doc.get_body().push_child(div)
         div.push_child(current_p)
-        subtitle_text = ""
 
-      subtitle_text += line
+      subtitle_lines.append(line)
 
       state = _State.TEXT_MORE
 

src/test/python/test_tt.py

@@ -111,6 +111,13 @@ def test_convert_bad_output_file_arg(self):
         "--otype", "not_ttml", 
         "--config_file", "src/test/resources/config_files/unit_test_cfg.json"])
 
+  def test_convert_bad_vtt_signature(self):
+    with self.assertRaises(SystemExit):
+      tt.main(["convert",
+        "-i", "src/test/resources/vtt/invalid/bad-signature.vtt",
+        "-o", "build/bad-signature.out.ttml",
+        "--config_file", "src/test/resources/config_files/unit_test_cfg.json"])
+
   def test_convert_mismtach_file_type_and_file_name(self):
     tt.main(["convert", 
       "-i", "src/test/resources/ttml/body_only.ttml", 

src/test/python/test_vtt_reader.py

@@ -50,13 +50,15 @@ def test_sample(self):
     self.assertIsNotNone(to_model(f))
 
 
-  def test_samples(self):
-    for root, _subdirs, files in os.walk("src/test/resources/vtt/"):
+  def test_valid_samples(self):
+    for root, subdirs, files in os.walk("src/test/resources/vtt/"):
+      # exclude invalid samples
+      subdirs[:] = [d for d in subdirs if d != "invalid"]
       for filename in files:
         (name, ext) = os.path.splitext(filename)
         if ext == ".vtt":
           with self.subTest(name):
-            with open(os.path.join(root, filename), encoding="utf-8") as f:
+            with open(os.path.join(root, filename), "r", encoding="utf-8-sig") as f:
               self.assertIsNotNone(to_model(f))
 
   def test_bold(self):
@@ -438,5 +440,81 @@ def test_default_positioning(self):
     self.assertEqual(o.x.value, 100*1/40)
     self.assertEqual(e.width.value, 100 - 2*100*1/40)
 
+  def test_bad_signature_1(self):
+    SAMPLE = """webvtt
+
+1
+00:00:00.000 --> 00:00:02.000
+Line 0 starting from top
+
+"""
+    f = io.StringIO(SAMPLE)
+    self.assertIsNone(to_model(f))
+
+  def test_bad_signature_2(self):
+    SAMPLE = """WEB
+
+1
+00:00:00.000 --> 00:00:02.000
+Line 0 starting from top
+
+"""
+    f = io.StringIO(SAMPLE)
+    self.assertIsNone(to_model(f))
+
+  def test_bad_signature_3(self):
+    SAMPLE = """WEBVTTS
+
+1
+00:00:00.000 --> 00:00:02.000
+Line 0 starting from top
+
+"""
+    f = io.StringIO(SAMPLE)
+    self.assertIsNone(to_model(f))
+
+  def test_bad_signature_4(self):
+    SAMPLE = """1
+00:00:00.000 --> 00:00:02.000
+Line 0 starting from top
+
+"""
+    f = io.StringIO(SAMPLE)
+    self.assertIsNone(to_model(f))
+
+  def test_bad_signature_5(self):
+    SAMPLE = ""
+    f = io.StringIO(SAMPLE)
+    self.assertIsNone(to_model(f))
+
+  def test_null_char(self):
+    SAMPLE = """WEBVTT
+1
+00:00:00.000 --> 00:00:02.000
+Line \u0000
+
+"""
+    f = io.StringIO(SAMPLE)
+    doc = to_model(f)
+
+    text_node = doc.get_body().first_child().first_child().first_child().first_child()
+    self.assertIsInstance(text_node, model.Text)
+    self.assertEqual(text_node.get_text(), "Line �")
+
+  def test_empty_file_1(self):
+    SAMPLE = """WEBVTT"""
+    f = io.StringIO(SAMPLE)
+    m = to_model(f)
+    self.assertIsNotNone(m)
+    self.assertEqual(len(m.get_body()), 0)
+
+  def test_empty_file_2(self):
+    SAMPLE = """WEBVTT ABD
+"""
+    f = io.StringIO(SAMPLE)
+    m = to_model(f)
+    self.assertIsNotNone(m)
+    self.assertEqual(len(m.get_body()), 0)
+
 if __name__ == '__main__':
   unittest.main()

src/test/resources/vtt/invalid/bad-signature.vtt

@@ -0,0 +1,4 @@
+vtt
+
+00:00:01.002 --> 00:00:03.004
+Hello