fix(composer): coerce remaining raw int() payload values, harden JS file/download handling

CodeRabbit flagged several payload-derived int() conversions in composer.py
that could raise ValueError instead of clamping like every other coerced
value in the module (_as_rgb_filter and its callers -- fillR/G/B, outR/G/B,
bgR/G/B, trackR/G/B -- plus min_width, lineSpacing, barWidth/Height,
start/endAngle, borderRadius, pip*, sparkline bar*, marquee gap/scrollSpeed).
Route them all through _safe_int for consistency with the rest of the module
and to avoid the generic 422 a raw ValueError produces.

Also:
- FileReader.onerror was unset in importDesign(), so a failed file read
  produced no status message.
- URL.revokeObjectURL() ran synchronously right after link.click() in both
  exportDesign() and generateZip(); deferred via setTimeout(..., 0) so the
  download reliably starts before the object URL is revoked.
- _module_level_code() (test helper) filtered ast.ImportFrom but not
  ast.Import, so a bare `import os` payload wouldn't be caught by the
  helper itself, even though downstream assertions still caught it.

Added regression tests for the fillR/G/B injection + clamping path, which
had no coverage (existing tests only covered the r/g/b _rgb_expr path).

Skipped as not worth the churn (CodeRabbit nitpicks, both "Trivial/Low value"):
- test_composer_empty_block.py's branch regex not matching digit-containing
  type names -- no such type exists today.
- test_composer_path_containment.py's `C.composer_bp.name and Flask(...)`
  truthiness guard -- cosmetic, blueprint name is never falsy.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XKc832xpVEx3C3W5BVqQ5Z
This commit is contained in:
Claude
2026-09-12 17:35:16 +00:00
parent f12d11334a
commit 6c23994b2f
3 changed files with 50 additions and 23 deletions
+23 -1
View File
@@ -67,7 +67,7 @@ def _module_level_code(src):
tree = ast.parse(src) tree = ast.parse(src)
out = [] out = []
for node in tree.body: for node in tree.body:
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.ImportFrom)): if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.ImportFrom, ast.Import)):
continue continue
if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant): if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant):
continue # the docstring continue # the docstring
@@ -145,6 +145,28 @@ def test_colour_channels_are_clamped_to_a_byte():
assert "(255, 0, 128)" in src, "channels were not clamped to 0-255" assert "(255, 0, 128)" in src, "channels were not clamped to 0-255"
#: fillR/fillG/fillB (and outR/G/B, bgR/G/B, trackR/G/B) go through
#: _as_fill_filter -> _as_rgb_filter, a separate path from _rgb_expr above.
#: It used raw int() until it was found to raise ValueError on a non-numeric
#: channel instead of clamping like every other coerced value in this module.
@pytest.mark.parametrize("evil", EXPR_PAYLOADS)
@pytest.mark.parametrize("channel", ["fillR", "fillG", "fillB"])
def test_a_non_numeric_fill_channel_cannot_reach_the_source(evil, channel):
el = {"type": "rectangle", "id": "r1", "x": 0, "y": 0, "width": 10, "height": 8,
"fillR": 0, "fillG": 0, "fillB": 128}
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_fill_channels_are_clamped_to_a_byte():
el = {"type": "rectangle", "id": "r1", "x": 0, "y": 0, "width": 10, "height": 8,
"fillR": 99999, "fillG": -5, "fillB": 128}
src = _generated(_payload(elements=[el]))
assert "(255, 0, 128)" in src, "fill channels were not clamped to 0-255"
def test_the_generated_module_still_has_no_top_level_statements(): def test_the_generated_module_still_has_no_top_level_statements():
"""The clean case: a normal payload produces only imports and a class.""" """The clean case: a normal payload produces only imports and a class."""
el = {"type": "text", "id": "t1", "x": 4, "y": 4, "text": "hi", el = {"type": "text", "id": "t1", "x": 4, "y": 4, "text": "hi",
+22 -20
View File
@@ -95,8 +95,10 @@ def _as_rgb_filter(val) -> str:
if val is None: if val is None:
return 'None' return 'None'
# nosemgrep: not a Flask route -- a Jinja filter emitting a Python # nosemgrep: not a Flask route -- a Jinja filter emitting a Python
# tuple literal, with every channel coerced by int(). # tuple literal, with every channel coerced and clamped by _safe_int().
return f'({int(val[0])}, {int(val[1])}, {int(val[2])})' # nosemgrep return (f'({_safe_int(val[0], 0, 0, 255)}, ' # nosemgrep
f'{_safe_int(val[1], 0, 0, 255)}, '
f'{_safe_int(val[2], 0, 0, 255)})')
def _as_fill_filter(val) -> str: def _as_fill_filter(val) -> str:
@@ -278,7 +280,7 @@ def _preprocess_elements(elements: list) -> list:
x_anchor = el.get('xAnchor') or None x_anchor = el.get('xAnchor') or None
y_anchor = el.get('yAnchor') or None y_anchor = el.get('yAnchor') or None
p['min_width'] = int(el.get('minWidth', 0) or 0) p['min_width'] = _safe_int(el.get('minWidth', 0) or 0, 0, 0, 4096)
if t in ('text', 'clock'): if t in ('text', 'clock'):
font_key = el.get('font', 'press_start') font_key = el.get('font', 'press_start')
@@ -290,7 +292,7 @@ def _preprocess_elements(elements: list) -> list:
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height') p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
font_size = _FONT_SIZE_MAP.get(font_key, 8) font_size = _FONT_SIZE_MAP.get(font_key, 8)
char_w = _FONT_CHAR_W.get(font_key, 8) char_w = _FONT_CHAR_W.get(font_key, 8)
line_spacing = int(el.get('lineSpacing', 2)) line_spacing = _safe_int(el.get('lineSpacing', 2), 2, 0, 256)
y_expr = p['y_expr'] y_expr = p['y_expr']
p['y2_expr'] = f"({y_expr}) + {font_size + line_spacing}" p['y2_expr'] = f"({y_expr}) + {font_size + line_spacing}"
if t == 'text': if t == 'text':
@@ -377,8 +379,8 @@ def _preprocess_elements(elements: list) -> list:
elif t == 'progress_bar': elif t == 'progress_bar':
p['x_expr'] = _compute_pos_expr(el.get('x', 0), x_anchor, 'width') 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['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['bar_width'] = int(el.get('barWidth', 40)) p['bar_width'] = _safe_int(el.get('barWidth', 40), 40, 0, 4096)
p['bar_height'] = int(el.get('barHeight', 6)) p['bar_height'] = _safe_int(el.get('barHeight', 6), 6, 0, 4096)
binding = el.get('binding', {}) binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '') p['binding_key'] = binding.get('key', '')
p['fill_tuple'] = _rgb_tuple(el, ('r', 'g', 'b'), (100, 200, 100)) p['fill_tuple'] = _rgb_tuple(el, ('r', 'g', 'b'), (100, 200, 100))
@@ -403,8 +405,8 @@ def _preprocess_elements(elements: list) -> list:
p['y_expr'] = y_expr p['y_expr'] = y_expr
p['x2_expr'] = f"({x_expr}) + {w}" p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}" p['y2_expr'] = f"({y_expr}) + {h}"
p['start_angle'] = int(el.get('startAngle', 0)) p['start_angle'] = _safe_int(el.get('startAngle', 0), 0, -3600, 3600)
p['end_angle'] = int(el.get('endAngle', 270)) p['end_angle'] = _safe_int(el.get('endAngle', 270), 270, -3600, 3600)
p['line_width'] = _safe_int(el.get('lineWidth'), 2, 1, 64) p['line_width'] = _safe_int(el.get('lineWidth'), 2, 1, 64)
p['rgb_tuple'] = _rgb_expr(el, 255, 200, 0) p['rgb_tuple'] = _rgb_expr(el, 255, 200, 0)
p['blink'] = bool(el.get('blink', False)) p['blink'] = bool(el.get('blink', False))
@@ -445,7 +447,7 @@ def _preprocess_elements(elements: list) -> list:
p['y_expr'] = y_expr p['y_expr'] = y_expr
p['x2_expr'] = f"({x_expr}) + {w}" p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}" p['y2_expr'] = f"({y_expr}) + {h}"
p['border_radius'] = int(el.get('borderRadius', 3)) p['border_radius'] = _safe_int(el.get('borderRadius', 3), 3, 0, 128)
fill = ( fill = (
[el.get('fillR', 0), el.get('fillG', 80), el.get('fillB', 180)] [el.get('fillR', 0), el.get('fillG', 80), el.get('fillB', 180)]
if el.get('hasFill', True) else None if el.get('hasFill', True) else None
@@ -473,9 +475,9 @@ def _preprocess_elements(elements: list) -> list:
elif t == 'pips': elif t == 'pips':
p['x_expr'] = _compute_pos_expr(el.get('x', 0), x_anchor, 'width') 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['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['pip_count'] = max(1, int(el.get('count', 5))) p['pip_count'] = _safe_int(el.get('count', 5), 5, 1, 256)
p['pip_size'] = max(1, int(el.get('pipSize', 4))) p['pip_size'] = _safe_int(el.get('pipSize', 4), 4, 1, 256)
p['pip_spacing'] = max(0, int(el.get('pipSpacing', 2))) p['pip_spacing'] = _safe_int(el.get('pipSpacing', 2), 2, 0, 256)
p['show_empty'] = bool(el.get('showEmpty', True)) p['show_empty'] = bool(el.get('showEmpty', True))
binding = el.get('binding', {}) binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '') p['binding_key'] = binding.get('key', '')
@@ -488,10 +490,10 @@ def _preprocess_elements(elements: list) -> list:
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height') y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['x_expr'] = x_expr p['x_expr'] = x_expr
p['y_expr'] = y_expr p['y_expr'] = y_expr
p['bar_width_px'] = int(el.get('width', 40)) p['bar_width_px'] = _safe_int(el.get('width', 40), 40, 0, 4096)
p['bar_height_px'] = int(el.get('height', 12)) p['bar_height_px'] = _safe_int(el.get('height', 12), 12, 0, 4096)
p['bar_count'] = max(1, int(el.get('barCount', 8))) p['bar_count'] = _safe_int(el.get('barCount', 8), 8, 1, 256)
p['bar_spacing'] = max(0, int(el.get('barSpacing', 1))) p['bar_spacing'] = _safe_int(el.get('barSpacing', 1), 1, 0, 256)
binding = el.get('binding', {}) binding = el.get('binding', {})
p['binding_key'] = binding.get('key', '') p['binding_key'] = binding.get('key', '')
p['fill_tuple'] = _rgb_tuple(el, ('r', 'g', 'b'), (80, 200, 120)) p['fill_tuple'] = _rgb_tuple(el, ('r', 'g', 'b'), (80, 200, 120))
@@ -508,8 +510,8 @@ def _preprocess_elements(elements: list) -> list:
p['y_expr'] = y_expr p['y_expr'] = y_expr
p['x2_expr'] = f"({x_expr}) + {w}" p['x2_expr'] = f"({x_expr}) + {w}"
p['y2_expr'] = f"({y_expr}) + {h}" p['y2_expr'] = f"({y_expr}) + {h}"
p['start_angle'] = int(el.get('startAngle', 135)) p['start_angle'] = _safe_int(el.get('startAngle', 135), 135, -3600, 3600)
p['end_angle'] = int(el.get('endAngle', 45)) p['end_angle'] = _safe_int(el.get('endAngle', 45), 45, -3600, 3600)
p['line_width'] = _safe_int(el.get('lineWidth'), 3, 1, 64) p['line_width'] = _safe_int(el.get('lineWidth'), 3, 1, 64)
p['rgb_tuple'] = _rgb_expr(el, 80, 220, 80) p['rgb_tuple'] = _rgb_expr(el, 80, 220, 80)
track = ( track = (
@@ -532,8 +534,8 @@ def _preprocess_elements(elements: list) -> list:
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height') p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
p['text'] = el.get('text', 'Scrolling text') p['text'] = el.get('text', 'Scrolling text')
p['char_w'] = _FONT_CHAR_W.get(font_key, 8) p['char_w'] = _FONT_CHAR_W.get(font_key, 8)
p['gap'] = int(el.get('gap', 16)) p['gap'] = _safe_int(el.get('gap', 16), 16, 0, 4096)
p['scroll_speed'] = max(1, int(el.get('scrollSpeed', 1))) p['scroll_speed'] = _safe_int(el.get('scrollSpeed', 1), 1, 1, 256)
p['direction'] = el.get('direction', 'left') p['direction'] = el.get('direction', 'left')
# Data key stored in self._data for stateful scrolling across # Data key stored in self._data for stateful scrolling across
# display() calls. It is spliced UNQUOTED into variable names # display() calls. It is spliced UNQUOTED into variable names
@@ -572,7 +572,7 @@ function composerApp() {
link.download = (this.metadata.id || 'composer-design') + '.composer.json'; link.download = (this.metadata.id || 'composer-design') + '.composer.json';
link.href = url; link.href = url;
link.click(); link.click();
URL.revokeObjectURL(url); setTimeout(() => URL.revokeObjectURL(url), 0);
}, },
importDesign() { importDesign() {
@@ -583,6 +583,9 @@ function composerApp() {
const file = e.target.files[0]; const file = e.target.files[0];
if (!file) return; if (!file) return;
const reader = new FileReader(); const reader = new FileReader();
reader.onerror = () => {
this._setStatus('Failed to read file', 'error');
};
reader.onload = (ev) => { reader.onload = (ev) => {
try { try {
const data = JSON.parse(ev.target.result); const data = JSON.parse(ev.target.result);
@@ -1326,7 +1329,7 @@ function composerApp() {
const url = URL.createObjectURL(blob); const url = URL.createObjectURL(blob);
const a = document.createElement('a'); const a = document.createElement('a');
a.href = url; a.download = `${this.metadata.id}.zip`; a.click(); a.href = url; a.download = `${this.metadata.id}.zip`; a.click();
URL.revokeObjectURL(url); setTimeout(() => URL.revokeObjectURL(url), 0);
this.generateStatus = 'done'; this.generateStatus = 'done';
this._setStatus('Plugin ZIP downloaded', 'success'); this._setStatus('Plugin ZIP downloaded', 'success');
setTimeout(() => { if (this.generateStatus === 'done') this.generateStatus = 'idle'; }, 4000); setTimeout(() => { if (this.generateStatus === 'done') this.generateStatus = 'idle'; }, 4000);