Python: fix files being silently dropped over Rust/Python escape mismatch

python/extractor/semmle/logging.py

@@ -65,6 +65,14 @@ def write_message_with_proc(level, proc_id, text):
 
 _logging_process = None
 
+def format_message(fmt, args):
+    '''Applies `%`-formatting to `fmt`, but only when there are arguments to interpolate.
+
+    This mirrors the standard library's `logging` behaviour, and means that a message that has
+    already been formatted -- and may therefore contain arbitrary `%` directives coming from the
+    code being analysed -- is passed through unharmed.'''
+    return fmt % args if args else fmt
+
 def stop():
     _logging_process.join()
 
@@ -105,7 +113,7 @@ def log(self, level, fmt, *args):
         '''Log a message in a process safe fashion.
         Message will be of the form [level] fmt%args.'''
         if level <= self.level:
-            txt = fmt % args
+            txt = format_message(fmt, args)
             try:
                 self.queue.put((self.color | level, self.proc_id, txt), False)
             except Exception:

python/extractor/semmle/python/imports.py

@@ -252,6 +252,6 @@ def importer_from_options(options, finder, logger):
         importer = CachingModuleImporter(options.trap_cache, finder, logger)
     except Exception as ex:
         if options.trap_cache is not None:
-            logger.warn("Failed to create caching importer: %s", ex)
+            logger.warning("Failed to create caching importer: %s", ex)
         importer = ModuleImporter(finder, logger)
     return importer

python/extractor/semmle/python/parser/dump_ast.py

@@ -99,12 +99,12 @@ def visit(self, node, level=0, visited=None):
 class StdoutLogger(logging.Logger):
     error_count = 0
     def log(self, level, fmt, *args):
-        sys.stdout.write(fmt % args + "\n")
+        sys.stdout.write(logging.format_message(fmt, args) + "\n")
 
     def info(self, fmt, *args):
         self.log(logging.INFO, fmt, *args)
 
-    def warn(self, fmt, *args):
+    def warning(self, fmt, *args):
         self.log(logging.WARN, fmt, *args)
         self.error_count += 1
 

python/extractor/semmle/python/parser/tsg_parser.py

@@ -3,6 +3,7 @@
 # Functions and classes used for parsing Python files using `tree-sitter-graph`
 
 from ast import literal_eval
+import re
 import sys
 import os
 import semmle.python.parser
@@ -144,6 +145,7 @@ def read_tsg_python_output(path, logger):
             elif value == "#null": # e.g. `exc: #null`
                 value = None
             else: # literal values, e.g. `name: "k1.k2"` or `level: 5`
+                value = rust_to_python_escapes(value)
                 try:
                     if key =="s" and value[0] == '"': # e.g. `s: "k1.k2"`
                         value = evaluate_string(value)
@@ -171,6 +173,36 @@ def read_tsg_python_output(path, logger):
     logger.debug("Read {} nodes and {} edges from TSG output".format(len(node_attr), len(edge_attr)))
     return node_attr, edge_attr
 
+# `tsg-python` serialises string values using Rust's `Debug` formatting, which diverges from what
+# Python's `literal_eval` accepts in two ways:
+#  - characters Rust considers non-printable -- including grapheme-extending ones such as the U+FE0F
+#    variation selector, U+200D zero width joiner and combining accents -- are rendered as `\u{...}`,
+#    a syntax Python does not know at all;
+#  - NUL is rendered as `\0`, which Python reads as the start of an *octal* escape, silently
+#    swallowing up to two more digits (NUL followed by `1` is emitted as `"\01"`, which decodes
+#    to `\x01`).
+# Everything else Rust emits (`\t`, `\r`, `\n`, `\\`, `\"`, and unescaped characters) is read back
+# identically by `literal_eval`, as verified exhaustively over every Unicode scalar value.
+_RUST_ESCAPE = re.compile(r"\\(?:u\{([0-9a-fA-F]{1,6})\}|.)", re.DOTALL)
+
+def rust_to_python_escapes(text):
+    """Rewrites Rust escapes in `text` that Python would reject or misread into their equivalents.
+
+    Matching every escape sequence (rather than only the offending ones) keeps the scan in step with
+    the backslashes, so an escaped backslash -- how a literal `\\u{fe0f}` in the source is
+    serialised -- is left alone."""
+    if "\\u{" not in text and "\\0" not in text:
+        return text
+    def replace(match):
+        code_point = match.group(1)
+        if code_point is None:
+            return "\\x00" if match.group(0) == "\\0" else match.group(0)
+        code_point = int(code_point, 16)
+        if code_point > 0xFFFF:
+            return "\\U{:08x}".format(code_point)
+        return "\\u{:04x}".format(code_point)
+    return _RUST_ESCAPE.sub(replace, text)
+
 def evaluate_string(s):
     s = literal_eval(s)
     prefix, quotes, content = split_string(s, None)

python/extractor/semmle/python/passes/flow.py

@@ -12,7 +12,7 @@
 from semmle.python.passes import unroller
 from semmle.python import modules
 import semmle.graph as graph
-from semmle.logging import Logger
+from semmle.logging import Logger, format_message
 
 __all__ = [ 'FlowPass' ]
 
@@ -1924,7 +1924,7 @@ def write_ssa_phi(out, phi, arg):
 class FakeLogger(object):
 
     def debug(self, fmt, *args):
-        print(fmt % args)
+        print(format_message(fmt, args))
 
     def traceback(self):
         print(traceback.format_exc())

python/extractor/semmle/worker.py

@@ -39,7 +39,7 @@ def __init__(self, max_depth, logger: Logger):
                     changed_paths = data.get('changes', [])
                     self.overlay_changes = { os.path.abspath(p) for p in changed_paths }
             except (IOError, ValueError) as e:
-                logger.warn("Failed to read overlay changes from '%s' (falling back to full extraction): %s", overlay_changes_file, e)
+                logger.warning("Failed to read overlay changes from '%s' (falling back to full extraction): %s", overlay_changes_file, e)
                 self.overlay_changes = None
 
     def add_root(self, mod):

python/extractor/tests/parser/unicode_escapes_new.expected

@@ -0,0 +1,150 @@
+Module: [6, 0] - [29, 0]
+  body: [
+    TypeAlias: [6, 0] - [6, 12]
+      name:
+        Name: [6, 5] - [6, 6]
+          variable: Variable('X', None)
+          ctx: Store
+      type_parameters: []
+      value:
+        Name: [6, 9] - [6, 12]
+          variable: Variable('int', None)
+          ctx: Load
+    Assign: [9, 0] - [9, 37]
+      targets: [
+        Name: [9, 0] - [9, 4]
+          variable: Variable('warn', None)
+          ctx: Store
+      ]
+      value:
+        Str: [9, 7] - [9, 37]
+          s: '⚠️  problem %s: %s'
+          prefix: '"'
+          implicitly_concatenated_parts: None
+    Assign: [10, 0] - [10, 35]
+      targets: [
+        Name: [10, 0] - [10, 8]
+          variable: Variable('warn_raw', None)
+          ctx: Store
+      ]
+      value:
+        Str: [10, 11] - [10, 35]
+          s: '⚠️  problem %s: %s'
+          prefix: '"'
+          implicitly_concatenated_parts: None
+    Assign: [13, 0] - [13, 19]
+      targets: [
+        Name: [13, 0] - [13, 3]
+          variable: Variable('zwj', None)
+          ctx: Store
+      ]
+      value:
+        Str: [13, 6] - [13, 19]
+          s: '👨\u200d💻'
+          prefix: '"'
+          implicitly_concatenated_parts: None
+    Assign: [16, 0] - [16, 14]
+      targets: [
+        Name: [16, 0] - [16, 3]
+          variable: Variable('nfd', None)
+          ctx: Store
+      ]
+      value:
+        Str: [16, 6] - [16, 14]
+          s: 'café'
+          prefix: '"'
+          implicitly_concatenated_parts: None
+    Assign: [17, 0] - [17, 28]
+      targets: [
+        Name: [17, 0] - [17, 11]
+          variable: Variable('soft_hyphen', None)
+          ctx: Store
+      ]
+      value:
+        Str: [17, 14] - [17, 28]
+          s: 'soft\xadhyphen'
+          prefix: '"'
+          implicitly_concatenated_parts: None
+    Assign: [20, 0] - [20, 12]
+      targets: [
+        Name: [20, 0] - [20, 6]
+          variable: Variable('café', None)
+          ctx: Store
+      ]
+      value:
+        Name: [20, 9] - [20, 12]
+          variable: Variable('nfd', None)
+          ctx: Load
+    Assign: [23, 0] - [23, 23]
+      targets: [
+        Name: [23, 0] - [23, 3]
+          variable: Variable('raw', None)
+          ctx: Store
+      ]
+      value:
+        Str: [23, 6] - [23, 23]
+          s: '⚠️\\u{fe0f}'
+          prefix: 'r"'
+          implicitly_concatenated_parts: None
+    Assign: [24, 0] - [24, 23]
+      targets: [
+        Name: [24, 0] - [24, 6]
+          variable: Variable('joined', None)
+          ctx: Store
+      ]
+      value:
+        JoinedStr: [24, 9] - [24, 23]
+          values: [
+            Str: [24, 9] - [24, 12]
+              s: ''
+              prefix: 'f"'
+              implicitly_concatenated_parts: None
+            Name: [24, 12] - [24, 15]
+              variable: Variable('zwj', None)
+              ctx: Load
+            Str: [24, 15] - [24, 23]
+              s: '⚠️'
+              prefix: 'f"'
+              implicitly_concatenated_parts: None
+          ]
+    Assign: [25, 0] - [25, 37]
+      targets: [
+        Name: [25, 0] - [25, 12]
+          variable: Variable('concatenated', None)
+          ctx: Store
+      ]
+      value:
+        Str: [25, 15] - [25, 37]
+          s: '⚠️👨\u200d💻'
+          prefix: '"'
+          implicitly_concatenated_parts: [
+            StringPart: [25, 15] - [25, 23]
+              prefix: '"'
+              text: '"⚠️"'
+              s: '⚠️'
+            StringPart: [25, 24] - [25, 37]
+              prefix: '"'
+              text: '"👨\u200d💻"'
+              s: '👨\u200d💻'
+          ]
+    Assign: [28, 0] - [28, 17]
+      targets: [
+        Name: [28, 0] - [28, 1]
+          variable: Variable('d', None)
+          ctx: Store
+      ]
+      value:
+        Dict: [28, 4] - [28, 17]
+          items: [
+            KeyValuePair: [28, 5] - [28, 16]
+              key:
+                Str: [28, 5] - [28, 13]
+                  s: '⚠️'
+                  prefix: '"'
+                  implicitly_concatenated_parts: None
+              value:
+                Num: [28, 15] - [28, 16]
+                  n: 1
+                  text: '1'
+          ]
+  ]

python/extractor/tests/parser/unicode_escapes_new.py

@@ -0,0 +1,28 @@
+# Characters that Rust's `Debug` formatting escapes as `\u{...}` when `tsg-python` serialises the
+# source text. See https://github.com/github/codeql/issues/22435.
+
+# PEP 695 syntax is what makes the old parser bail out and hand the file to `tsg-python` in the
+# first place, so keep the reported reproducer intact.
+type X = int
+
+# U+FE0F variation selector, next to a `%` directive.
+warn = "\u26a0\ufe0f  problem %s: %s"
+warn_raw = "⚠️  problem %s: %s"
+
+# U+200D zero width joiner.
+zwj = "👨‍💻"
+
+# Combining acute accent (NFD), and a soft hyphen.
+nfd = "café"
+soft_hyphen = "soft­hyphen"
+
+# Combining marks are valid in identifiers too.
+café = nfd
+
+# In f-strings, raw strings and implicit concatenations too.
+raw = r"⚠️\u{fe0f}"
+joined = f"{zwj}⚠️"
+concatenated = "⚠️" "👨‍💻"
+
+# ... and outside of string literals.
+d = {"⚠️": 1}  # comment with ⚠️ and 👨‍💻

python/extractor/tests/test_tsg_parser.py

@@ -0,0 +1,85 @@
+import unittest
+
+from ast import literal_eval
+
+from semmle.logging import format_message
+from semmle.python.parser.tsg_parser import evaluate_string, rust_to_python_escapes
+
+
+class RustEscapeTest(unittest.TestCase):
+    """`tsg-python` serialises strings with Rust's `Debug` formatting, which escapes characters such
+    as U+FE0F as `\\u{...}` -- a syntax Python's `literal_eval` does not accept -- and NUL as `\\0`,
+    which Python reads as an octal escape."""
+
+    def test_untouched_without_escapes(self):
+        text = '"caf\u00e9 \u2713 \U0001f4be"'
+        self.assertEqual(rust_to_python_escapes(text), text)
+
+    def test_basic_multilingual_plane(self):
+        self.assertEqual(rust_to_python_escapes(r'"\u{fe0f}"'), r'"\ufe0f"')
+        self.assertEqual(rust_to_python_escapes(r'"\u{200d}"'), r'"\u200d"')
+
+    def test_short_and_astral_code_points(self):
+        self.assertEqual(rust_to_python_escapes(r'"\u{0}"'), r'"\u0000"')
+        self.assertEqual(rust_to_python_escapes(r'"\u{1f4a9}"'), r'"\U0001f4a9"')
+
+    def test_other_escapes_are_preserved(self):
+        self.assertEqual(rust_to_python_escapes(r'"a\nb\"c\u{ad}"'), r'"a\nb\"c\u00ad"')
+
+    def test_escaped_backslash_is_not_an_escape_introducer(self):
+        # How a raw string `r"\u{fe0f}"` in the analysed source gets serialised: the `\u{fe0f}` is
+        # literal text, not an escape, and must survive unchanged.
+        self.assertEqual(rust_to_python_escapes(r'"\\u{fe0f}"'), r'"\\u{fe0f}"')
+
+    def test_nul_is_not_left_as_an_octal_escape(self):
+        # Rust renders NUL as `\0`; Python would read that as the start of an octal escape and
+        # swallow the digits that follow, decoding `"\01"` to U+0001 instead of NUL then `1`.
+        self.assertEqual(rust_to_python_escapes(r'"\01"'), r'"\x001"')
+
+    def test_every_escape_shape_round_trips(self):
+        # Rust's `Debug for str` only ever emits these escape shapes. Check that each round-trips
+        # with every printable ASCII neighbour before and after it.
+        for escape_shape, expected in [
+            (r'\0', "\x00"),
+            (r'\t', "\t"),
+            (r'\n', "\n"),
+            (r'\r', "\r"),
+            (r'\\', "\\"),
+            (r'\"', '"'),
+            (r'\u{1}', "\u0001"),
+            (r'\u{1f}', "\u001f"),
+            (r'\u{300}', "\u0300"),
+            (r'\u{fe0f}', "\ufe0f"),
+            (r'\u{e0100}', "\U000e0100"),
+            (r'\u{10fffe}', "\U0010fffe"),
+        ]:
+            for neighbour in map(chr, range(0x20, 0x7F)):
+                rendered_neighbour = {"\\": r"\\", '"': r'\"'}.get(neighbour, neighbour)
+                for position, text, expected_value in [
+                    ("before", '"' + rendered_neighbour + escape_shape + '"', neighbour + expected),
+                    ("after", '"' + escape_shape + rendered_neighbour + '"', expected + neighbour),
+                ]:
+                    with self.subTest(
+                        escape_shape=escape_shape,
+                        neighbour=neighbour,
+                        position=position,
+                    ):
+                        self.assertEqual(literal_eval(rust_to_python_escapes(text)), expected_value)
+
+    def test_evaluate_string_on_reported_value(self):
+        # The exact value from https://github.com/github/codeql/issues/22435 that used to raise
+        # `truncated \uXXXX escape`.
+        value = rust_to_python_escapes('"\\"\u26a0\\u{fe0f}  problem %s: %s\\""')
+        self.assertEqual(evaluate_string(value), "\u26a0\ufe0f  problem %s: %s")
+
+
+class FormatMessageTest(unittest.TestCase):
+    """A pre-formatted log message may contain `%` directives coming from the analysed source, and
+    must not be `%`-formatted again."""
+
+    def test_no_arguments(self):
+        message = "Error while parsing value '%s: %s'"
+        self.assertEqual(format_message(message, ()), message)
+
+    def test_with_arguments(self):
+        self.assertEqual(format_message("%s and %s", ("a", "b")), "a and b")