224 lines
9.5 KiB
Python
224 lines
9.5 KiB
Python
# Natural Language Toolkit: ReDoS backstop for caller-supplied patterns
|
|
#
|
|
# Copyright (C) 2001-2026 NLTK Project
|
|
# URL: <https://www.nltk.org>
|
|
# For license information, see LICENSE.TXT
|
|
|
|
r"""
|
|
A wall-clock backstop for NLTK code paths that compile *caller-supplied*
|
|
regular expressions or tag patterns.
|
|
|
|
Several public APIs accept a pattern from the caller and run it over
|
|
(possibly attacker-controlled) text:
|
|
|
|
* :class:`nltk.tokenize.RegexpTokenizer` / ``regexp_tokenize`` -- ``pattern``
|
|
* :class:`nltk.tag.RegexpTagger` -- the ``regexp`` of each ``(regexp, tag)`` pair
|
|
* :func:`nltk.tgrep.tgrep_compile` -- the ``/regex/`` node literal in a query
|
|
* :class:`nltk.chunk.regexp.RegexpChunkRule` and its ``ChunkRule`` / ``StripRule``
|
|
/ ``UnChunkRule`` / ``MergeRule`` / ``SplitRule`` subclasses -- the tag pattern
|
|
|
|
A crafted pattern (e.g. ``(a+)+$`` or ``(a|a)*$``) can trigger catastrophic
|
|
backtracking and pin a CPU core indefinitely -- a denial of service
|
|
(CWE-1333). This module centralises a single, tested defence used by all of
|
|
those sinks, in two layers:
|
|
|
|
1. **Compile with the third-party ``regex`` engine** instead of the stdlib
|
|
``re``. Its optimiser collapses a large class of catastrophic patterns
|
|
-- notably nested quantifiers over the *same* sub-expression such as
|
|
``(a+)+$`` -- to linear time outright, so those "attacks" finish in
|
|
microseconds and never reach the timeout.
|
|
|
|
2. **Run every match under a wall-clock ``timeout``**. The optimiser does *not*
|
|
cover every shape -- an alternation of identical branches such as
|
|
``(a|a)*`` (and the equivalent ``<a|a>*`` chunk tag pattern) still
|
|
backtracks exponentially in *both* ``re`` and ``regex``. The timeout is
|
|
therefore a *mandatory* backstop, not a nicety: when it trips, an
|
|
actionable :class:`TimeoutError` is raised instead of hanging the process.
|
|
|
|
Rewriting the pattern to be linear (atomic groups / possessive quantifiers)
|
|
was evaluated and rejected: it does *not* defuse the identical-branch
|
|
alternation, and it silently changes match semantics for legitimate greedy
|
|
patterns such as ``<.*>*<NN>``. A bounded timeout is the only defence that is
|
|
both complete and behaviour-preserving for arbitrary caller patterns.
|
|
|
|
Usage::
|
|
|
|
from nltk import redos
|
|
rx = redos.compile(pattern, flags) # -> a TimedPattern
|
|
rx.findall(text) # runs under DEFAULT_TIMEOUT
|
|
rx.search(text, timeout=1.0) # per-call override
|
|
|
|
``TimedPattern`` quacks like the subset of ``re.Pattern`` these sinks use
|
|
(``findall`` / ``finditer`` / ``split`` / ``search`` / ``match`` / ``fullmatch``
|
|
/ ``sub`` / ``subn`` plus attribute access such as ``.pattern`` and ``.flags``),
|
|
so it is a drop-in replacement for a compiled pattern at those call sites.
|
|
"""
|
|
|
|
import regex
|
|
|
|
__all__ = ["DEFAULT_TIMEOUT", "TimedPattern", "compile"]
|
|
|
|
#: Wall-clock seconds any single caller-supplied-pattern match may run before it
|
|
#: is abandoned with :class:`TimeoutError`. Legitimate tokenizing / tagging /
|
|
#: chunking of even large inputs finishes in well under a second, so this is
|
|
#: roughly a thousand-fold head-room for real use while still capping a crafted
|
|
#: pattern's CPU burn. It is deliberately far below a naive "just make it large"
|
|
#: value (e.g. 60s): in a denial-of-service the busy-wait window *is* the damage,
|
|
#: so a shorter cap is strictly safer. Set a call's ``timeout`` to ``None`` to
|
|
#: disable the limit for a trusted pattern.
|
|
DEFAULT_TIMEOUT = 5.0
|
|
|
|
#: Sentinel so ``timeout=None`` (disable) is distinguishable from "use the
|
|
#: module default", and so a later ``nltk.redos.DEFAULT_TIMEOUT = ...`` still
|
|
#: takes effect for calls that did not pass an explicit timeout.
|
|
_UNSET = object()
|
|
|
|
|
|
class TimedPattern:
|
|
"""A compiled :mod:`regex` pattern whose match operations are bounded by a
|
|
wall-clock timeout.
|
|
|
|
Every exposed match method runs the underlying engine with ``timeout=`` and
|
|
re-raises a bare :class:`TimeoutError` as an actionable one naming the
|
|
offending pattern. Any other attribute (``pattern``, ``flags``, ``groups``,
|
|
``groupindex``, ...) is delegated to the wrapped pattern, so instances are a
|
|
drop-in for a compiled pattern at NLTK's regex sinks.
|
|
"""
|
|
|
|
__slots__ = ("_rx", "_timeout")
|
|
|
|
def __init__(self, compiled, timeout=_UNSET):
|
|
self._rx = compiled
|
|
self._timeout = timeout
|
|
|
|
# -- internals ---------------------------------------------------------
|
|
|
|
def _resolve(self, timeout):
|
|
if timeout is _UNSET:
|
|
timeout = self._timeout
|
|
if timeout is _UNSET:
|
|
timeout = DEFAULT_TIMEOUT
|
|
return timeout
|
|
|
|
def _fail(self, timeout):
|
|
return TimeoutError(
|
|
f"regular-expression match exceeded its {timeout}s time limit; the "
|
|
f"pattern {self._rx.pattern!r} may be pathological for this input "
|
|
f"(catastrophic backtracking / ReDoS). Pass timeout=None to disable "
|
|
f"the limit for a trusted pattern."
|
|
)
|
|
|
|
# -- match API (timeout-guarded) --------------------------------------
|
|
|
|
def search(self, string, *args, timeout=_UNSET, **kwargs):
|
|
t = self._resolve(timeout)
|
|
try:
|
|
return self._rx.search(string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
def match(self, string, *args, timeout=_UNSET, **kwargs):
|
|
t = self._resolve(timeout)
|
|
try:
|
|
return self._rx.match(string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
def fullmatch(self, string, *args, timeout=_UNSET, **kwargs):
|
|
t = self._resolve(timeout)
|
|
try:
|
|
return self._rx.fullmatch(string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
def findall(self, string, *args, timeout=_UNSET, **kwargs):
|
|
t = self._resolve(timeout)
|
|
try:
|
|
return self._rx.findall(string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
def split(self, string, *args, timeout=_UNSET, **kwargs):
|
|
t = self._resolve(timeout)
|
|
try:
|
|
return self._rx.split(string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
def sub(self, repl, string, *args, timeout=_UNSET, **kwargs):
|
|
t = self._resolve(timeout)
|
|
try:
|
|
return self._rx.sub(repl, string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
def subn(self, repl, string, *args, timeout=_UNSET, **kwargs):
|
|
t = self._resolve(timeout)
|
|
try:
|
|
return self._rx.subn(repl, string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
def finditer(self, string, *args, timeout=_UNSET, **kwargs):
|
|
# ``regex.finditer`` is lazy: the timeout is enforced as items are
|
|
# pulled, so the guard must wrap the *iteration*, not just the call.
|
|
t = self._resolve(timeout)
|
|
try:
|
|
yield from self._rx.finditer(string, *args, timeout=t, **kwargs)
|
|
except TimeoutError:
|
|
raise self._fail(t) from None
|
|
|
|
# -- transparent delegation for everything else -----------------------
|
|
|
|
def __getattr__(self, name):
|
|
# Only reached for attributes not defined above. Do NOT delegate dunder
|
|
# lookups (pickle/copy/etc.) -- delegating those to the wrapped pattern
|
|
# would make the wrapper masquerade as the pattern in confusing ways.
|
|
if name.startswith("__") and name.endswith("__"):
|
|
raise AttributeError(name)
|
|
return getattr(self._rx, name)
|
|
|
|
def __eq__(self, other):
|
|
# Two wrappers are equal when they guard the same source pattern under
|
|
# the same flags and timeout. Without this, instances fall back to
|
|
# identity, so an object that has been round-tripped (e.g. a
|
|
# ``RegexpTagger`` serialised to and from JSON, which re-compiles its
|
|
# patterns) never compares equal to the original.
|
|
if not isinstance(other, TimedPattern):
|
|
return NotImplemented
|
|
return (
|
|
self._rx.pattern == other._rx.pattern
|
|
and self._rx.flags == other._rx.flags
|
|
and self._timeout == other._timeout
|
|
)
|
|
|
|
def __hash__(self):
|
|
return hash((self._rx.pattern, self._rx.flags, self._timeout))
|
|
|
|
def __repr__(self):
|
|
return f"TimedPattern({self._rx.pattern!r}, timeout={self._timeout!r})"
|
|
|
|
|
|
def compile(pattern, flags=0, timeout=_UNSET):
|
|
r"""Compile ``pattern`` with the :mod:`regex` engine and wrap it so every
|
|
match runs under a wall-clock timeout.
|
|
|
|
:param pattern: a regular-expression string, or an already-compiled
|
|
``re``/``regex`` pattern (its ``.pattern`` is re-compiled with
|
|
``regex``), or an existing :class:`TimedPattern` (returned as-is with
|
|
the requested timeout).
|
|
:param flags: regex flags (stdlib ``re`` flag constants are accepted).
|
|
:param timeout: default wall-clock limit for matches on the returned
|
|
pattern; ``None`` disables it. Omit to use :data:`DEFAULT_TIMEOUT` at
|
|
match time (so a later change to that module attribute still applies).
|
|
:rtype: TimedPattern
|
|
"""
|
|
if isinstance(pattern, TimedPattern):
|
|
if timeout is _UNSET:
|
|
return pattern
|
|
return TimedPattern(pattern._rx, timeout)
|
|
# Accept a pre-compiled ``re``/``regex`` pattern by re-compiling its source.
|
|
src = getattr(pattern, "pattern", pattern)
|
|
compiled = regex.compile(src, flags)
|
|
return TimedPattern(compiled, timeout)
|