diff --git a/test/test_composer_code_injection.py b/test/test_composer_code_injection.py new file mode 100644 index 00000000..542ed378 --- /dev/null +++ b/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 diff --git a/web_interface/blueprints/composer.py b/web_interface/blueprints/composer.py index 88e2f2c7..544aa086 100644 --- a/web_interface/blueprints/composer.py +++ b/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: