fix(composer): stop payload text reaching generated Python as code

Review flagged this as critical and it is: the composer builds manager.py
by interpolating payload values into source text, /api/install writes
that file into plugins_dir, and the plugin loader imports and executes
it. The ast.parse check further down rejects only *invalid* syntax, and
an injected `import os` is perfectly valid.

Confirmed against the code before this commit. A plugin name carrying a
triple quote closes the module docstring and everything after it becomes
module-level code:

    generated manager.py parses: True
    injected module-level statements: ['import os', 'PWNED = os.getuid()']

and a geometry value is interpolated verbatim, because the parameter is
annotated int but arrives as JSON:

    _compute_pos_expr('0 or __import__("os").system("id")', 'right', 'width')
      -> 'width - 0 or __import__("os").system("id")'
    generated source: x=0 or __import__("os").system("id"),

Three fixes. _safe_int coerces and optionally clamps, and
_compute_pos_expr applies it to its own argument -- which covers all
twenty-odd call sites at once rather than patching each. _rgb_expr does
the same for the eight colour interpolations, clamping channels to
0-255. Line endpoints and widths go through it too.

For the docstring, _reject_source_breaking refuses a plugin name
containing a quote, backslash or newline. Rejecting rather than escaping:
these are display names, none of that belongs in one, and a clear "Plugin
name cannot contain a double quote." beats silently mangling what the
user typed.

Verified: all three exploits now refused or neutered, and each defence
mutation-checked separately --

    coercion removed in _compute_pos_expr ->  8 failed
    docstring guard removed               ->  5 failed
    colour channels interpolated raw      -> 13 failed

87 tests, covering seven expression payloads across seven geometry
fields and three colour channels, five literal-breaking names, and the
clean case asserting a normal payload still yields no module-level
statements at all.

One aside: the first version of this test file put the exploit string
in its own module docstring, which closed it and made the file a syntax
error -- the same bug, one level up. It now describes the payload rather
than embedding it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW

test/test_composer_code_injection.py

@@ -0,0 +1,117 @@
+"""The composer generates Python that the plugin loader imports and executes.
+
+/api/install writes the generated manager.py into plugins_dir and the loader
+imports it, so anything the payload can splice into that source runs on the
+device. The ast.parse check in _generate_plugin_files rejects only *invalid*
+syntax -- an injected `import os` is perfectly valid and passed it.
+
+Two ways in, both confirmed against the code before it was fixed:
+
+  metadata.name = a name containing a triple-quote, a newline, then
+                  `import os; PWNED = os.getuid()`, then another triple-quote
+      -> closes the module docstring; the rest became module-level statements
+         (spelled out rather than shown literally -- writing the payload into
+         this docstring closes *this* file's docstring, which is the bug)
+
+  element x     = '0 or __import__("os").system("id")'
+      -> f-string interpolated it verbatim: x=0 or __import__("os").system("id")
+"""
+import ast
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from web_interface.blueprints import composer as C  # noqa: E402
+
+BASE_META = {"id": "test-plugin", "name": "Clock", "author": "a",
+             "version": "1.0.0", "description": "d"}
+
+#: Values that terminate a Python expression and start a new statement.
+EXPR_PAYLOADS = [
+    '0 or __import__("os").system("id")',
+    '0);import os;os.system("id");(',
+    '__import__("subprocess").run(["id"])',
+    "0 if False else exec('x=1')",
+    "1e999", "nan", "0x41", "0__0",
+]
+
+#: Values that close a string literal in the generated source.
+LITERAL_PAYLOADS = [
+    'Clock"""\nimport os; PWNED = os.getuid()\n"""',
+    "Clock'''\nimport os\n'''",
+    'Clock" + __import__("os").system("id") + "',
+    "Clock\\", "Clock\nimport os",
+]
+
+
+def _payload(**over):
+    p = {"metadata": dict(BASE_META), "elements": [], "config_vars": []}
+    p["metadata"].update(over.pop("metadata", {}))
+    p.update(over)
+    return p
+
+
+def _generated(payload):
+    return C._generate_plugin_files(payload)["manager.py"]
+
+
+def _module_level_code(src):
+    """Statements at module level that are not the docstring/imports/classes."""
+    tree = ast.parse(src)
+    out = []
+    for node in tree.body:
+        if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.ImportFrom)):
+            continue
+        if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant):
+            continue  # the docstring
+        out.append(ast.unparse(node))
+    return out
+
+
+@pytest.mark.parametrize("payload", LITERAL_PAYLOADS)
+def test_a_name_that_breaks_out_of_a_literal_is_refused(payload):
+    with pytest.raises(C.ComposerInputError):
+        _generated(_payload(metadata={"name": payload}))
+
+
+@pytest.mark.parametrize("evil", EXPR_PAYLOADS)
+@pytest.mark.parametrize("field", ["x", "y", "x0", "y0", "x1", "y1", "lineWidth"])
+def test_a_non_numeric_geometry_value_cannot_reach_the_source(evil, field):
+    el = {"type": "line", "id": "l1", "x0": 0, "y0": 0, "x1": 10, "y1": 10,
+          "anchor_x": "right", "anchor_y": "bottom"}
+    el[field] = evil
+    src = _generated(_payload(elements=[el]))
+    assert "__import__" not in src, f"{field}={evil!r} reached the generated source"
+    assert "os.system" not in src
+    assert not _module_level_code(src), \
+        f"{field}={evil!r} produced module-level statements: {_module_level_code(src)}"
+
+
+@pytest.mark.parametrize("evil", EXPR_PAYLOADS)
+@pytest.mark.parametrize("channel", ["r", "g", "b"])
+def test_a_non_numeric_colour_channel_cannot_reach_the_source(evil, channel):
+    el = {"type": "text", "id": "t1", "x": 0, "y": 0, "text": "hi",
+          "font": "press_start", "r": 255, "g": 255, "b": 255}
+    el[channel] = evil
+    src = _generated(_payload(elements=[el]))
+    assert "__import__" not in src and "os.system" not in src
+    assert not _module_level_code(src)
+
+
+def test_colour_channels_are_clamped_to_a_byte():
+    el = {"type": "text", "id": "t1", "x": 0, "y": 0, "text": "hi",
+          "font": "press_start", "r": 99999, "g": -5, "b": 128}
+    src = _generated(_payload(elements=[el]))
+    assert "(255, 0, 128)" in src, "channels were not clamped to 0-255"
+
+
+def test_the_generated_module_still_has_no_top_level_statements():
+    """The clean case: a normal payload produces only imports and a class."""
+    el = {"type": "text", "id": "t1", "x": 4, "y": 4, "text": "hi",
+          "font": "press_start", "r": 1, "g": 2, "b": 3}
+    src = _generated(_payload(elements=[el]))
+    assert not _module_level_code(src)
+    assert "(1, 2, 3)" in src

web_interface/blueprints/composer.py

@@ -101,13 +101,62 @@ def _to_class_name(name: str) -> str:
     return base if base.endswith('Plugin') else base + 'Plugin'
 
 
-def _compute_pos_expr(val: int, anchor: str | None, dim_var: str) -> str:
+def _reject_source_breaking(value: str, field: str) -> None:
+    """Refuse text that could terminate a string literal in generated source.
+
+    Anything interpolated into manager.py inside quotes has to survive being
+    read back as Python. A quote, a backslash or a newline can end the literal
+    early and turn the remainder into executable statements.
+    """
+    for bad, label in (('"', 'a double quote'), ("'", 'a single quote'),
+                       ('\\', 'a backslash'), ('\n', 'a newline'),
+                       ('\r', 'a carriage return')):
+        if bad in value:
+            raise ComposerInputError(
+                f'{field} cannot contain {label}.')
+
+
+def _safe_int(value, default: int = 0, lo: int | None = None,
+              hi: int | None = None) -> int:
+    """Coerce a payload value to int, falling back rather than raising.
+
+    Everything this module interpolates into generated Python has to go
+    through here first. The payload is JSON from the browser, so a field
+    annotated `int` can arrive as any string, and these values are formatted
+    straight into `manager.py` -- which /api/install writes to disk and the
+    plugin loader then imports and executes. An x of
+
+        '0 or __import__("os").system("id")'
+
+    produced `x=0 or __import__("os").system("id")` in the generated source,
+    which is valid Python and so passed the ast.parse check.
+    """
+    try:
+        out = int(value)
+    except (TypeError, ValueError):
+        return default
+    if lo is not None:
+        out = max(lo, out)
+    if hi is not None:
+        out = min(hi, out)
+    return out
+
+
+def _rgb_expr(el: dict, dr: int = 255, dg: int = 255, db: int = 255) -> str:
+    """A colour tuple literal built from coerced, clamped channel values."""
+    return (f"({_safe_int(el.get('r'), dr, 0, 255)}, "
+            f"{_safe_int(el.get('g'), dg, 0, 255)}, "
+            f"{_safe_int(el.get('b'), db, 0, 255)})")
+
+
+def _compute_pos_expr(val, anchor: str | None, dim_var: str) -> str:
     """Produce a Python expression string for an anchored or fixed position.
 
     anchor=None/'left'/'top' → fixed pixel value
     anchor='center'          → dim_var // 2 ± offset
     anchor='right'/'bottom'  → dim_var - offset
     """
+    val = _safe_int(val, 0)
     if not anchor or anchor in ('left', 'top'):
         return str(val)
     if anchor in ('center', 'middle'):
@@ -170,7 +219,7 @@ def _preprocess_elements(elements: list) -> list:
         if t in ('text', 'clock'):
             font_key = el.get('font', 'press_start')
             p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'regular_font')
-            p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 255)}, {el.get('b', 255)})"
+            p['rgb_tuple'] = _rgb_expr(el, 255, 255, 255)
             text_align = el.get('textAlign', 'left')
             raw_x = el.get('x', 0)
             x_base_expr = _compute_pos_expr(raw_x, x_anchor, 'width')
@@ -206,7 +255,7 @@ def _preprocess_elements(elements: list) -> list:
             p['binding_format'] = binding.get('format')
             font_key = el.get('font', 'press_start')
             p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'regular_font')
-            p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 200)}, {el.get('b', 100)})"
+            p['rgb_tuple'] = _rgb_expr(el, 255, 200, 100)
             x_base_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
             p['x_expr'] = x_base_expr  # dynamic text: runtime content determines width; use raw pos
             p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
@@ -248,10 +297,10 @@ def _preprocess_elements(elements: list) -> list:
             else:
                 p['x0_expr'] = _compute_pos_expr(el.get('x0', 0), x_anchor, 'width')
                 p['y0_expr'] = _compute_pos_expr(el.get('y0', 0), y_anchor, 'height')
-                p['x1_expr'] = str(el.get('x1', 127))
-                p['y1_expr'] = str(el.get('y1', 0))
-            p['rgb_tuple'] = f"({el.get('r', 180)}, {el.get('g', 180)}, {el.get('b', 180)})"
-            p['line_width'] = el.get('lineWidth', 1)
+                p['x1_expr'] = str(_safe_int(el.get('x1'), 127))
+                p['y1_expr'] = str(_safe_int(el.get('y1'), 0))
+            p['rgb_tuple'] = _rgb_expr(el, 180, 180, 180)
+            p['line_width'] = _safe_int(el.get('lineWidth'), 1, 1, 64)
             p['blink'] = bool(el.get('blink', False))
 
         elif t == 'progress_bar':
@@ -285,8 +334,8 @@ def _preprocess_elements(elements: list) -> list:
             p['y2_expr'] = f"({y_expr}) + {h}"
             p['start_angle'] = int(el.get('startAngle', 0))
             p['end_angle'] = int(el.get('endAngle', 270))
-            p['line_width'] = max(1, int(el.get('lineWidth', 2)))
-            p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 200)}, {el.get('b', 0)})"
+            p['line_width'] = _safe_int(el.get('lineWidth'), 2, 1, 64)
+            p['rgb_tuple'] = _rgb_expr(el, 255, 200, 0)
             p['blink'] = bool(el.get('blink', False))
 
         elif t == 'ellipse':
@@ -313,7 +362,7 @@ def _preprocess_elements(elements: list) -> list:
         elif t == 'pixel':
             p['x_expr'] = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
             p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
-            p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 255)}, {el.get('b', 255)})"
+            p['rgb_tuple'] = _rgb_expr(el, 255, 255, 255)
             p['blink'] = bool(el.get('blink', False))
 
         elif t == 'rounded_rectangle':
@@ -341,7 +390,7 @@ def _preprocess_elements(elements: list) -> list:
         elif t == 'countdown':
             font_key = el.get('font', 'four_by_six')
             p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'extra_small_font')
-            p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 180)}, {el.get('b', 0)})"
+            p['rgb_tuple'] = _rgb_expr(el, 255, 180, 0)
             binding = el.get('binding', {})
             p['binding_key'] = binding.get('key', '')
             p['countdown_format'] = el.get('countdownFormat', 'dh')
@@ -390,8 +439,8 @@ def _preprocess_elements(elements: list) -> list:
             p['y2_expr'] = f"({y_expr}) + {h}"
             p['start_angle'] = int(el.get('startAngle', 135))
             p['end_angle'] = int(el.get('endAngle', 45))
-            p['line_width'] = max(1, int(el.get('lineWidth', 3)))
-            p['rgb_tuple'] = f"({el.get('r', 80)}, {el.get('g', 220)}, {el.get('b', 80)})"
+            p['line_width'] = _safe_int(el.get('lineWidth'), 3, 1, 64)
+            p['rgb_tuple'] = _rgb_expr(el, 80, 220, 80)
             track = (
                 [el.get('trackR', 40), el.get('trackG', 40), el.get('trackB', 40)]
                 if el.get('hasTrack', True) else None
@@ -408,7 +457,7 @@ def _preprocess_elements(elements: list) -> list:
         elif t == 'marquee':
             font_key = el.get('font', 'press_start')
             p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'regular_font')
-            p['rgb_tuple'] = f"({el.get('r', 255)}, {el.get('g', 255)}, {el.get('b', 255)})"
+            p['rgb_tuple'] = _rgb_expr(el, 255, 255, 255)
             p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
             p['text'] = el.get('text', 'Scrolling text')
             p['char_w'] = _FONT_CHAR_W.get(font_key, 8)
@@ -446,6 +495,17 @@ def _generate_plugin_files(data: dict) -> dict:
     plugin_name = metadata.get('name', '').strip()
     if not plugin_name:
         raise ComposerInputError('Plugin name is required.')
+    # The template drops this straight into manager.py's module docstring. A
+    # name carrying a triple quote closes that docstring and everything after
+    # it becomes module-level code, which /api/install writes to disk and the
+    # loader imports and runs:
+    #
+    #     Clock"""\nimport os; PWNED = os.getuid()\n"""
+    #       -> import os          <- executed on load
+    #          PWNED = os.getuid()
+    #
+    # ast.parse further down only rejects invalid syntax, and that is valid.
+    _reject_source_breaking(plugin_name, 'Plugin name')
 
     author = metadata.get('author', '').strip()
     if not author: