Parser robustness: guard against Earley chart blow-up with an opt-in timeout

isabelle_parser/cli.py

@@ -14,6 +14,14 @@ def main() -> None:
     arg_parser.add_argument(
         "-f", "--file", action="store_true", help="Interpret input as a filename"
     )
+    arg_parser.add_argument(
+        "-t",
+        "--timeout",
+        type=float,
+        default=None,
+        help="Abort the parse after this many seconds (guards against "
+        "pathological inputs that grow the Earley chart super-linearly)",
+    )
 
     # Parse arguments
     args = arg_parser.parse_args()
@@ -31,7 +39,7 @@ def main() -> None:
 
     # Lex and parse
     try:
-        result = parse(data)
+        result = parse(data, timeout=args.timeout)
     except ParsingError as e:
         print("Parsing failed due to errors.")
         print(f"Error: {e.with_source_code(data)}")

isabelle_parser/error.py

@@ -26,7 +26,7 @@ def _source_hint(self) -> str:
                 print(line)
                 underline = " " * (self.column - 1) + "^" * len(self.token)
                 return f"\n\n{line}\n{underline}\n"
-        return "source_hint"
+        return ""
 
     def __str__(self) -> str:
         location_info = ""

isabelle_parser/thy_parser.py

@@ -1,13 +1,52 @@
 import logging
 import os
-from typing import Any, List
+import signal
+import threading
+from contextlib import contextmanager
+from typing import Any, Iterator, List
 
 from lark import Lark, Transformer, Tree
 from lark.tree import Meta
 
+from .error import ParsingError
+
 logger = logging.getLogger(__name__)
 
 
+class _ParseTimeout(Exception):
+    """Internal signal that a parse exceeded its time budget."""
+
+
+@contextmanager
+def _time_limit(seconds: float | None) -> Iterator[None]:
+    """Abort the wrapped block after ``seconds`` using SIGALRM.
+
+    The Earley parser can grow its chart super-linearly with input size, so a
+    large/ambiguous theory file can take minutes. SIGALRM interrupts the
+    pure-Python parse loop cleanly. It is only available on the main thread of a
+    Unix process; in any other context (worker threads, platforms without
+    SIGALRM) the limit is silently skipped and parsing proceeds without a bound.
+    """
+    if (
+        not seconds
+        or not hasattr(signal, "SIGALRM")
+        or threading.current_thread() is not threading.main_thread()
+    ):
+        yield
+        return
+
+    def _handler(signum: int, frame: Any) -> None:
+        raise _ParseTimeout()
+
+    previous = signal.signal(signal.SIGALRM, _handler)
+    signal.setitimer(signal.ITIMER_REAL, seconds)
+    try:
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous)
+
+
 def load_parser(start: str = "start") -> Lark:
     # Construct the path to the .lark file
     grammar_path = os.path.join(os.path.dirname(__file__), "thy_grammar.lark")
@@ -21,10 +60,22 @@ def load_parser(start: str = "start") -> Lark:
     return parser
 
 
-def parse(input_text: str, parser: Lark | None = None) -> Any | None:
+def parse(
+    input_text: str, parser: Lark | None = None, timeout: float | None = None
+) -> Any | None:
+    """Parse ``input_text`` into a tree.
+
+    ``timeout`` (seconds) bounds the parse: if it is exceeded a
+    :class:`ParsingError` is raised instead of letting a pathological file hang
+    the caller. See :func:`_time_limit` for its limitations.
+    """
     if parser is None:
         parser = load_parser()
-    tree = parser.parse(input_text)
+    try:
+        with _time_limit(timeout):
+            tree = parser.parse(input_text)
+    except _ParseTimeout:
+        raise ParsingError(f"parsing timed out after {timeout}s")
     transformer = PositionPrinter()
     return transformer.transform(tree)
 

tests/test_thy_parser.py

@@ -14,10 +14,14 @@
 test_thy_grammer.py.
 """
 
+import signal
+import threading
+import time
+
 import pytest
 from lark import Lark, Tree
 
-from isabelle_parser import load_parser
+from isabelle_parser import ParsingError, load_parser
 from isabelle_parser.thy_parser import parse as _raw_parse
 
 
@@ -64,3 +68,57 @@ def test_invalid_input_raises(src):
     """Invalid theory inputs must raise rather than return a result."""
     with pytest.raises(Exception):
         _raw_parse(src, None)
+
+
+class _SlowParser:
+    """Stand-in Lark parser whose parse() sleeps, to test the timeout path
+    deterministically without depending on a real pathological input."""
+
+    def __init__(self, seconds: float) -> None:
+        self.seconds = seconds
+
+    def parse(self, _text: str):
+        time.sleep(self.seconds)
+        return Tree("start", [])
+
+
+class TestParseTimeout:
+    def test_timeout_not_hit_returns_tree(self):
+        result = _raw_parse(
+            "theory T imports Main begin end", load_parser(), timeout=30
+        )
+        assert isinstance(result, Tree)
+
+    def test_timeout_none_is_default(self):
+        result = _raw_parse("theory T imports Main begin end", load_parser())
+        assert isinstance(result, Tree)
+
+    def test_timeout_exceeded_raises_parsing_error(self):
+        with pytest.raises(ParsingError):
+            _raw_parse("ignored", _SlowParser(5), timeout=0.2)
+
+    def test_timeout_exceeded_is_fast(self):
+        start = time.time()
+        with pytest.raises(ParsingError):
+            _raw_parse("ignored", _SlowParser(10), timeout=0.2)
+        assert time.time() - start < 2  # aborted promptly, not after 10s
+
+    def test_sigalrm_handler_restored_after_parse(self):
+        sentinel = signal.getsignal(signal.SIGALRM)
+        _raw_parse("theory T imports Main begin end", load_parser(), timeout=30)
+        assert signal.getsignal(signal.SIGALRM) is sentinel
+
+    def test_timeout_ignored_off_main_thread(self):
+        # SIGALRM is main-thread only; off-thread the limit is skipped and a
+        # normal (fast) parse still succeeds rather than erroring.
+        results = {}
+
+        def run():
+            results["tree"] = _raw_parse(
+                "theory T imports Main begin end", load_parser(), timeout=30
+            )
+
+        t = threading.Thread(target=run)
+        t.start()
+        t.join()
+        assert isinstance(results["tree"], Tree)