286 lines
8.6 KiB
Python
286 lines
8.6 KiB
Python
"""Fixed, bounded CPython ``re`` bridge used by the Python engine worker.
|
|
|
|
This file is bundled as source by Vite and evaluated once while the trusted
|
|
Pyodide runtime starts. User-controlled values are passed as function
|
|
arguments; they are never evaluated as Python source.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import re
|
|
import sys
|
|
|
|
|
|
_BRIDGE_ABI_VERSION = 1
|
|
_MAX_CAPTURE_GROUPS = 1_000
|
|
_MAX_NATIVE_EXPANSION_CODE_POINTS = 16 * 1024 * 1024
|
|
|
|
|
|
def _json_result(value):
|
|
return json.dumps(value, ensure_ascii=True, separators=(",", ":"))
|
|
|
|
|
|
def _error(phase, code, error, position=None):
|
|
message = str(error)
|
|
if len(message) > 1_024:
|
|
message = message[:1_023] + "\N{HORIZONTAL ELLIPSIS}"
|
|
result = {
|
|
"abi": _BRIDGE_ABI_VERSION,
|
|
"ok": False,
|
|
"phase": phase,
|
|
"code": code,
|
|
"message": message,
|
|
}
|
|
if isinstance(position, int) and position >= 0:
|
|
result["position"] = position
|
|
return _json_result(result)
|
|
|
|
|
|
def _flag_bits(flags):
|
|
bits = 0
|
|
if "a" in flags:
|
|
bits |= re.ASCII
|
|
if "i" in flags:
|
|
bits |= re.IGNORECASE
|
|
if "m" in flags:
|
|
bits |= re.MULTILINE
|
|
if "s" in flags:
|
|
bits |= re.DOTALL
|
|
if "x" in flags:
|
|
bits |= re.VERBOSE
|
|
return bits
|
|
|
|
|
|
def _match_record(match, group_count):
|
|
captures = []
|
|
for group_number in range(1, group_count + 1):
|
|
start, end = match.span(group_number)
|
|
captures.append(None if start < 0 else [start, end])
|
|
start, end = match.span(0)
|
|
return {"span": [start, end], "captures": captures}
|
|
|
|
|
|
def _utf8_width(character):
|
|
code_point = ord(character)
|
|
if code_point <= 0x7F:
|
|
return 1
|
|
if code_point <= 0x7FF:
|
|
return 2
|
|
if code_point <= 0xFFFF:
|
|
return 3
|
|
return 4
|
|
|
|
|
|
def _append_utf8_bounded(chunks, value, byte_count, maximum_bytes):
|
|
remaining = maximum_bytes - byte_count
|
|
if not value:
|
|
return byte_count, True
|
|
end = 0
|
|
for character in value:
|
|
width = _utf8_width(character)
|
|
if width > remaining:
|
|
break
|
|
remaining -= width
|
|
byte_count += width
|
|
end += 1
|
|
if end:
|
|
chunks.append(value[:end])
|
|
return byte_count, end == len(value)
|
|
|
|
|
|
def _expansion_code_points(parsed_template, match):
|
|
total = 0
|
|
for part in parsed_template:
|
|
if isinstance(part, int):
|
|
start, end = match.span(part)
|
|
if start >= 0:
|
|
total += end - start
|
|
else:
|
|
total += len(part)
|
|
if total > _MAX_NATIVE_EXPANSION_CODE_POINTS:
|
|
return total
|
|
return total
|
|
|
|
|
|
def _identity():
|
|
return _json_result(
|
|
{
|
|
"abi": _BRIDGE_ABI_VERSION,
|
|
"ok": True,
|
|
"identity": {
|
|
"implementation": sys.implementation.name,
|
|
"pythonVersion": ".".join(
|
|
str(part) for part in sys.version_info[:3]
|
|
),
|
|
"reModule": re.__name__,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
def _regex_tools_run(
|
|
operation,
|
|
pattern,
|
|
flags,
|
|
subject,
|
|
maximum_matches,
|
|
maximum_capture_rows,
|
|
replacement,
|
|
maximum_output_bytes,
|
|
):
|
|
if operation == "identity":
|
|
return _identity()
|
|
if operation not in ("execute", "replace"):
|
|
return _error("bridge", "invalid-operation", "Unsupported bridge operation.")
|
|
|
|
try:
|
|
compiled = re.compile(pattern, _flag_bits(flags))
|
|
except re.PatternError as error:
|
|
return _error(
|
|
"compile",
|
|
"compile-error",
|
|
error,
|
|
getattr(error, "pos", None),
|
|
)
|
|
except Exception as error:
|
|
return _error("compile", "compile-error", error)
|
|
|
|
group_count = compiled.groups
|
|
if group_count > _MAX_CAPTURE_GROUPS:
|
|
return _error(
|
|
"compile",
|
|
"capture-group-limit",
|
|
(
|
|
f"The pattern declares {group_count} capture groups; "
|
|
f"the bridge limit is {_MAX_CAPTURE_GROUPS}."
|
|
),
|
|
)
|
|
|
|
group_names = sorted(
|
|
([group_number, name] for name, group_number in compiled.groupindex.items()),
|
|
key=lambda item: item[0],
|
|
)
|
|
|
|
parsed_template = None
|
|
if operation == "replace":
|
|
try:
|
|
parsed_template = re._parser.parse_template(replacement, compiled)
|
|
except (re.PatternError, IndexError) as error:
|
|
return _error(
|
|
"replacement",
|
|
"replacement-error",
|
|
error,
|
|
getattr(error, "pos", None),
|
|
)
|
|
except Exception as error:
|
|
return _error("replacement", "replacement-error", error)
|
|
|
|
iterate = "g" in flags
|
|
try:
|
|
matches = compiled.finditer(subject) if iterate else (
|
|
candidate for candidate in (compiled.search(subject),) if candidate
|
|
)
|
|
records = []
|
|
capture_rows = 0
|
|
results_truncated = False
|
|
output_chunks = []
|
|
output_bytes = 0
|
|
output_truncated = False
|
|
output_stopped = False
|
|
next_source_position = 0
|
|
|
|
for match in matches:
|
|
rows_for_match = max(1, group_count)
|
|
if (
|
|
len(records) >= maximum_matches
|
|
or capture_rows + rows_for_match > maximum_capture_rows
|
|
):
|
|
results_truncated = True
|
|
break
|
|
|
|
records.append(_match_record(match, group_count))
|
|
capture_rows += rows_for_match
|
|
|
|
if operation == "replace" and not output_stopped:
|
|
start, end = match.span(0)
|
|
output_bytes, complete = _append_utf8_bounded(
|
|
output_chunks,
|
|
subject[next_source_position:start],
|
|
output_bytes,
|
|
maximum_output_bytes,
|
|
)
|
|
if not complete:
|
|
output_truncated = True
|
|
output_stopped = True
|
|
else:
|
|
expansion_size = _expansion_code_points(
|
|
parsed_template, match
|
|
)
|
|
if expansion_size > _MAX_NATIVE_EXPANSION_CODE_POINTS:
|
|
return _error(
|
|
"replacement",
|
|
"replacement-expansion-limit",
|
|
(
|
|
"One native match expansion exceeds the "
|
|
f"{_MAX_NATIVE_EXPANSION_CODE_POINTS} "
|
|
"code-point safety limit."
|
|
),
|
|
)
|
|
expanded = match.expand(replacement)
|
|
output_bytes, complete = _append_utf8_bounded(
|
|
output_chunks,
|
|
expanded,
|
|
output_bytes,
|
|
maximum_output_bytes,
|
|
)
|
|
if not complete:
|
|
output_truncated = True
|
|
output_stopped = True
|
|
next_source_position = end
|
|
|
|
if operation == "replace" and not output_stopped:
|
|
output_bytes, complete = _append_utf8_bounded(
|
|
output_chunks,
|
|
subject[next_source_position:],
|
|
output_bytes,
|
|
maximum_output_bytes,
|
|
)
|
|
if not complete:
|
|
output_truncated = True
|
|
|
|
result = {
|
|
"abi": _BRIDGE_ABI_VERSION,
|
|
"ok": True,
|
|
"groupCount": group_count,
|
|
"groupNames": group_names,
|
|
"matches": records,
|
|
"resultsTruncated": results_truncated,
|
|
}
|
|
if operation == "replace":
|
|
output = "".join(output_chunks).encode("utf-8")
|
|
if len(output) != output_bytes:
|
|
return _error(
|
|
"bridge",
|
|
"output-byte-count-mismatch",
|
|
"The bounded Python output byte count is inconsistent.",
|
|
)
|
|
result.update(
|
|
{
|
|
"outputBase64": base64.b64encode(output).decode("ascii"),
|
|
"outputBytes": output_bytes,
|
|
"outputTruncated": output_truncated,
|
|
}
|
|
)
|
|
return _json_result(result)
|
|
except re.PatternError as error:
|
|
phase = "replacement" if operation == "replace" else "match"
|
|
code = "replacement-error" if operation == "replace" else "match-error"
|
|
return _error(phase, code, error, getattr(error, "pos", None))
|
|
except Exception as error:
|
|
phase = "replacement" if operation == "replace" else "match"
|
|
code = "replacement-error" if operation == "replace" else "match-error"
|
|
return _error(phase, code, error)
|
|
|
|
|
|
_regex_tools_run
|