mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-23 19:38:13 +00:00
ChuckBuildsandClaude Opus 5
1a0864e5d4
Code injection, found by chasing why a security test could not have caught it.
_preprocess_elements built the far corner of five shapes by interpolating the
payload's width/height straight into generated Python:
w = el.get('width', 10)
p['x2_expr'] = f"({x_expr}) + {w}"
so a rectangle with width='0 or __import__("os").system("id")' generated
[0, 0, (0) + 0 or __import__("os").system("id"), (0) + 8],
inside a manager.py that /api/install writes to disk and the plugin loader
imports and executes. rectangle, arc, ellipse, rounded_rectangle and gauge all
share the pattern. Both fields now go through _safe_int, like every other
geometry value.
Unreachable today only because composer_bp is still unregistered -- the same
caveat as the docstring injection fixed earlier in this PR.
Why the existing test missed it
-------------------------------
test_a_non_numeric_geometry_value_cannot_reach_the_source drove its payloads
through a "line" element. manager.py.j2 has never had a `line` branch, so
_preprocess_elements produced nothing for it and no value it set could reach
the generated source. Every assertion passed trivially, against code that was
in fact vulnerable. The test has been vacuous since it was written; the
_RENDERABLE_ELEMENT_TYPES constant added in the previous commit only made the
cause legible.
It now runs across the five types that actually render, over x/y/width/height:
40 of those cases fail with the clamping reverted, where the old version
passed 100%.
A second test asserts every type used by the injection suite is in
_RENDERABLE_ELEMENT_TYPES, so the suite cannot quietly go vacuous again.
Also: _payload set "config_vars", but _generate_plugin_files reads
data['dataModel']['configVars']. Nothing passed through that key was ever
read. Fixed so config-var tests exercise the real path.
Full suite: 3967 passed, the one failure being test_install_lowmem
(pre-existing, awaiting #492).
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW
1060 lines
45 KiB
Python
1060 lines
45 KiB
Python
"""
|
|
Plugin Composer blueprint — drag-and-drop plugin builder for LEDMatrix.
|
|
|
|
Routes:
|
|
GET /composer/ — Composer page
|
|
POST /composer/api/generate — Generate and return plugin ZIP
|
|
POST /composer/api/install — Write plugin directly to plugins_dir
|
|
GET /composer/api/fonts/<name> — Serve TTF font files for canvas rendering
|
|
GET /composer/api/validate-id/<id> — Check if a plugin ID is already taken
|
|
"""
|
|
import ast
|
|
import io
|
|
import json
|
|
import keyword
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Optional
|
|
import zipfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import jinja2
|
|
import jsonschema
|
|
from flask import Blueprint, jsonify, render_template, request, send_file
|
|
from werkzeug.utils import secure_filename
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
composer_bp = Blueprint('composer', __name__)
|
|
|
|
# Module-level attributes injected by app.py at registration time
|
|
composer_bp.config_manager = None
|
|
composer_bp.plugin_manager = None
|
|
composer_bp.plugins_dir = None
|
|
composer_bp.project_root = None
|
|
|
|
# Fonts safe to serve to the browser for canvas rendering
|
|
_ALLOWED_FONTS = frozenset({'PressStart2P-Regular.ttf', '4x6-font.ttf', '5by7.regular.ttf'})
|
|
|
|
# Map composer font keys → DisplayManager attribute names
|
|
_FONT_ATTR_MAP = {
|
|
'press_start': 'regular_font',
|
|
'four_by_six': 'extra_small_font',
|
|
'five_by_seven': 'bdf_5x7_font',
|
|
}
|
|
|
|
# Font sizes in LED pixels (used to compute second-line Y offsets)
|
|
_FONT_SIZE_MAP = {
|
|
'press_start': 8,
|
|
'four_by_six': 6,
|
|
'five_by_seven': 7,
|
|
}
|
|
|
|
_PLUGIN_ID_RE = re.compile(r'\A[a-z][a-z0-9-]{0,62}\Z')
|
|
#: \Z, not $. Python's $ also matches just before a trailing newline,
|
|
#: so '$' would accept "myplugin\\n" and create a directory whose name
|
|
#: ends in one. Not traversal, but not a name anything should have to
|
|
#: handle either.
|
|
_PYTHON_IDENT_RE = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
|
|
|
|
# ── Jinja2 environment (separate from Flask's; autoescape=False for code gen) ──
|
|
|
|
_jinja_env: jinja2.Environment | None = None
|
|
|
|
|
|
def _get_jinja_env() -> jinja2.Environment:
|
|
global _jinja_env
|
|
if _jinja_env is None:
|
|
template_dir = Path(__file__).parent.parent / 'templates' / 'v3' / 'composer'
|
|
_jinja_env = jinja2.Environment( # nosec B701 - see below
|
|
loader=jinja2.FileSystemLoader(str(template_dir)),
|
|
# These templates emit Python source, not HTML. Autoescaping would
|
|
# turn a quote in a plugin name into " inside generated code
|
|
# and break it, so it stays off deliberately -- and the safety has
|
|
# to come from the values instead. It does: every numeric value is
|
|
# coerced by _safe_int/_rgb_expr, and text that could terminate a
|
|
# string literal is rejected by _reject_source_breaking. Both are
|
|
# covered by test/test_composer_code_injection.py, which is where
|
|
# to look before relaxing any of it.
|
|
autoescape=False,
|
|
trim_blocks=True,
|
|
lstrip_blocks=True,
|
|
)
|
|
_jinja_env.filters['as_rgb'] = _as_rgb_filter
|
|
_jinja_env.filters['as_fill'] = _as_fill_filter
|
|
return _jinja_env
|
|
|
|
|
|
def _as_rgb_filter(val) -> str:
|
|
"""[r, g, b] → '(r, g, b)'"""
|
|
if val is None:
|
|
return 'None'
|
|
return f'({int(val[0])}, {int(val[1])}, {int(val[2])})'
|
|
|
|
|
|
def _as_fill_filter(val) -> str:
|
|
"""[r, g, b] or None → '(r, g, b)' or 'None'"""
|
|
if val is None:
|
|
return 'None'
|
|
return _as_rgb_filter(val)
|
|
|
|
|
|
# ── Helper functions ──────────────────────────────────────────────────────────
|
|
|
|
def _to_class_name(name: str) -> str:
|
|
"""'My Clock' → 'MyClockPlugin' (avoids double-suffix if name already ends with Plugin)"""
|
|
words = re.sub(r'[^a-zA-Z0-9]', ' ', name).split()
|
|
base = ''.join(w.capitalize() for w in words)
|
|
return base if base.endswith('Plugin') else base + 'Plugin'
|
|
|
|
|
|
#: Attribute names BasePlugin (or the generated __init__) already owns. A
|
|
#: config var using one of these produces valid Python that quietly clobbers
|
|
#: the plugin's own state instead of failing loudly.
|
|
_RESERVED_ATTRS = frozenset({
|
|
'config', 'logger', 'display_manager', 'cache_manager', 'plugin_manager',
|
|
'plugin_id', 'enabled', 'global_config', 'self', 'update', 'display',
|
|
'validate_config', 'get_info', 'cleanup',
|
|
})
|
|
|
|
|
|
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'):
|
|
if val == 0:
|
|
return f"{dim_var} // 2"
|
|
return f"{dim_var} // 2 + {val}" if val > 0 else f"{dim_var} // 2 - {abs(val)}"
|
|
if anchor in ('right', 'bottom'):
|
|
return dim_var if val == 0 else f"{dim_var} - {val}"
|
|
return str(val)
|
|
|
|
|
|
# Character widths in LED pixels per font (for text-alignment x offset math)
|
|
_FONT_CHAR_W = {
|
|
'press_start': 8,
|
|
'four_by_six': 4,
|
|
'five_by_seven': 5,
|
|
}
|
|
|
|
|
|
def _aligned_x_expr(x_base_expr: str, text_align: str, char_count: int, char_w: int) -> str:
|
|
"""Return Python x expression for text alignment.
|
|
|
|
left → x_base_expr (no change)
|
|
center → x_base_expr - half_text_width
|
|
right → x_base_expr - text_width
|
|
"""
|
|
if text_align == 'left' or not text_align:
|
|
return x_base_expr
|
|
text_px = char_count * char_w
|
|
if text_align == 'center':
|
|
offset = text_px // 2
|
|
return f"({x_base_expr}) - {offset}" if offset else x_base_expr
|
|
if text_align == 'right':
|
|
return f"({x_base_expr}) - {text_px}" if text_px else x_base_expr
|
|
return x_base_expr
|
|
|
|
|
|
#: Element types manager.py.j2 has a drawing branch for. Kept next to the
|
|
#: preprocessor because the two must agree: a type here with no branch emits an
|
|
#: empty block, and a type with a branch but missing here is silently dropped.
|
|
_RENDERABLE_ELEMENT_TYPES = frozenset({
|
|
'text', 'dynamic_text', 'clock', 'countdown', 'rectangle', 'arc',
|
|
'ellipse', 'pixel', 'rounded_rectangle', 'pips', 'sparkline', 'gauge',
|
|
'marquee', 'progress_bar',
|
|
})
|
|
|
|
|
|
def _preprocess_elements(elements: list) -> list:
|
|
"""Expand raw element dicts into template-ready dicts with anchor expressions.
|
|
|
|
Invisible elements (visible=False) are excluded from generated code entirely.
|
|
"""
|
|
result = []
|
|
for el in elements:
|
|
# Skip hidden elements — they exist only in the preview
|
|
if el.get('visible') is False:
|
|
continue
|
|
|
|
p = dict(el)
|
|
t = el.get('type', '')
|
|
|
|
# Section elements are layer-list annotations only — no canvas output
|
|
if t == 'section':
|
|
continue
|
|
|
|
# A type the template has no branch for still gets its breakpoint and
|
|
# blink wrappers emitted, and those would open an `if` with nothing in
|
|
# it -- ast.parse then fails and the caller is told only "Generated
|
|
# code has a syntax error". Drop it here instead. The template also
|
|
# emits a `pass` fallback, so a type added to the canvas before its
|
|
# branch exists degrades to a no-op rather than a broken plugin.
|
|
if t not in _RENDERABLE_ELEMENT_TYPES:
|
|
logger.info("composer: skipping element type %r with no template branch", t)
|
|
continue
|
|
|
|
x_anchor = el.get('xAnchor') or None
|
|
y_anchor = el.get('yAnchor') or None
|
|
p['min_width'] = int(el.get('minWidth', 0) or 0)
|
|
|
|
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'] = _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')
|
|
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
font_size = _FONT_SIZE_MAP.get(font_key, 8)
|
|
char_w = _FONT_CHAR_W.get(font_key, 8)
|
|
line_spacing = int(el.get('lineSpacing', 2))
|
|
y_expr = p['y_expr']
|
|
p['y2_expr'] = f"({y_expr}) + {font_size + line_spacing}"
|
|
if t == 'text':
|
|
t1 = el.get('text', '') or ''
|
|
t2 = el.get('text2', '') or ''
|
|
# p is a copy of the raw element, so a payload omitting these
|
|
# leaves the key absent and the template renders
|
|
# {{ el.text | tojson }} over a jinja2.Undefined. tojson then
|
|
# raises TypeError, which no handler catches -- so a missing
|
|
# key came back as a 500 rather than a validation error.
|
|
p['text'] = t1
|
|
p['text2'] = t2
|
|
# Detect {variable} tokens — generate format_map() call instead of literal
|
|
_var_re = re.compile(r'\{([a-zA-Z_]\w*)\}')
|
|
p['text_is_template'] = bool(_var_re.search(t1) or _var_re.search(t2))
|
|
ref_len = max(len(t1), len(t2)) if t2 else len(t1)
|
|
p['x_expr'] = _aligned_x_expr(x_base_expr, text_align, ref_len, char_w)
|
|
p['x2_expr'] = p['x_expr'] # second line uses same x
|
|
else: # clock
|
|
fmt1 = el.get('format', '%H:%M') or '%H:%M'
|
|
p['format'] = fmt1
|
|
fmt2 = el.get('format2', '') or ''
|
|
p['format2'] = fmt2
|
|
ref_len = max(len(fmt1), len(fmt2)) if fmt2 else len(fmt1)
|
|
p['x_expr'] = _aligned_x_expr(x_base_expr, text_align, ref_len, char_w)
|
|
p['x2_expr'] = p['x_expr']
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'dynamic_text':
|
|
binding = el.get('binding', {})
|
|
p['binding_source'] = binding.get('source', 'config')
|
|
p['binding_key'] = binding.get('key', '')
|
|
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'] = _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')
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'rectangle':
|
|
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
|
|
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
w = _safe_int(el.get('width'), 10, 0, 4096)
|
|
h = _safe_int(el.get('height'), 8, 0, 4096)
|
|
p['x_expr'] = x_expr
|
|
p['y_expr'] = y_expr
|
|
# x2/y2 as runtime expressions to support anchored positions
|
|
p['x2_expr'] = f"({x_expr}) + {w}"
|
|
p['y2_expr'] = f"({y_expr}) + {h}"
|
|
fill = (
|
|
[el.get('fillR', 0), el.get('fillG', 0), el.get('fillB', 128)]
|
|
if el.get('hasFill', True) else None
|
|
)
|
|
outline = (
|
|
[el.get('outR', 255), el.get('outG', 255), el.get('outB', 255)]
|
|
if el.get('hasOutline', True) else None
|
|
)
|
|
p['fill_tuple'] = _as_fill_filter(fill)
|
|
p['outline_tuple'] = _as_fill_filter(outline)
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t in ('line', 'divider'):
|
|
if t == 'divider':
|
|
orient = el.get('orientation', 'horizontal')
|
|
if orient == 'horizontal':
|
|
y_val = el.get('y', 16)
|
|
y_expr = _compute_pos_expr(y_val, y_anchor, 'height')
|
|
p.update(x0_expr='0', y0_expr=y_expr, x1_expr='width - 1', y1_expr=y_expr)
|
|
else:
|
|
x_val = el.get('x', 64)
|
|
x_expr = _compute_pos_expr(x_val, x_anchor, 'width')
|
|
p.update(x0_expr=x_expr, y0_expr='0', x1_expr=x_expr, y1_expr='height - 1')
|
|
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(_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':
|
|
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['bar_width'] = int(el.get('barWidth', 40))
|
|
p['bar_height'] = int(el.get('barHeight', 6))
|
|
binding = el.get('binding', {})
|
|
p['binding_key'] = binding.get('key', '')
|
|
p['fill_tuple'] = f"({el.get('r', 100)}, {el.get('g', 200)}, {el.get('b', 100)})"
|
|
bg = (
|
|
[el.get('bgR', 30), el.get('bgG', 30), el.get('bgB', 30)]
|
|
if el.get('hasBg', True) else None
|
|
)
|
|
outline = (
|
|
[el.get('outR', 100), el.get('outG', 100), el.get('outB', 100)]
|
|
if el.get('hasOutline', True) else None
|
|
)
|
|
p['bg_tuple'] = _as_fill_filter(bg)
|
|
p['outline_tuple'] = _as_fill_filter(outline)
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'arc':
|
|
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
|
|
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
w = _safe_int(el.get('width'), 24, 0, 4096)
|
|
h = _safe_int(el.get('height'), 24, 0, 4096)
|
|
p['x_expr'] = x_expr
|
|
p['y_expr'] = y_expr
|
|
p['x2_expr'] = f"({x_expr}) + {w}"
|
|
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'] = _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':
|
|
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
|
|
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
w = _safe_int(el.get('width'), 24, 0, 4096)
|
|
h = _safe_int(el.get('height'), 12, 0, 4096)
|
|
p['x_expr'] = x_expr
|
|
p['y_expr'] = y_expr
|
|
p['x2_expr'] = f"({x_expr}) + {w}"
|
|
p['y2_expr'] = f"({y_expr}) + {h}"
|
|
fill = (
|
|
[el.get('fillR', 0), el.get('fillG', 100), el.get('fillB', 200)]
|
|
if el.get('hasFill', True) else None
|
|
)
|
|
outline = (
|
|
[el.get('outR', 100), el.get('outG', 180), el.get('outB', 255)]
|
|
if el.get('hasOutline', True) else None
|
|
)
|
|
p['fill_tuple'] = _as_fill_filter(fill)
|
|
p['outline_tuple'] = _as_fill_filter(outline)
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
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'] = _rgb_expr(el, 255, 255, 255)
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'rounded_rectangle':
|
|
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
|
|
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
w = _safe_int(el.get('width'), 24, 0, 4096)
|
|
h = _safe_int(el.get('height'), 10, 0, 4096)
|
|
p['x_expr'] = x_expr
|
|
p['y_expr'] = y_expr
|
|
p['x2_expr'] = f"({x_expr}) + {w}"
|
|
p['y2_expr'] = f"({y_expr}) + {h}"
|
|
p['border_radius'] = int(el.get('borderRadius', 3))
|
|
fill = (
|
|
[el.get('fillR', 0), el.get('fillG', 80), el.get('fillB', 180)]
|
|
if el.get('hasFill', True) else None
|
|
)
|
|
outline = (
|
|
[el.get('outR', 120), el.get('outG', 180), el.get('outB', 255)]
|
|
if el.get('hasOutline', True) else None
|
|
)
|
|
p['fill_tuple'] = _as_fill_filter(fill)
|
|
p['outline_tuple'] = _as_fill_filter(outline)
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
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'] = _rgb_expr(el, 255, 180, 0)
|
|
binding = el.get('binding', {})
|
|
p['binding_key'] = binding.get('key', '')
|
|
p['countdown_format'] = el.get('countdownFormat', 'dh')
|
|
x_base_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
|
|
p['x_expr'] = x_base_expr
|
|
p['y_expr'] = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'pips':
|
|
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['pip_count'] = max(1, int(el.get('count', 5)))
|
|
p['pip_size'] = max(1, int(el.get('pipSize', 4)))
|
|
p['pip_spacing'] = max(0, int(el.get('pipSpacing', 2)))
|
|
p['show_empty'] = bool(el.get('showEmpty', True))
|
|
binding = el.get('binding', {})
|
|
p['binding_key'] = binding.get('key', '')
|
|
p['fill_tuple'] = f"({el.get('r', 255)}, {el.get('g', 200)}, {el.get('b', 0)})"
|
|
p['empty_tuple'] = f"({el.get('emptyR', 50)}, {el.get('emptyG', 50)}, {el.get('emptyB', 50)})"
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'sparkline':
|
|
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
|
|
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
p['x_expr'] = x_expr
|
|
p['y_expr'] = y_expr
|
|
p['bar_width_px'] = int(el.get('width', 40))
|
|
p['bar_height_px'] = int(el.get('height', 12))
|
|
p['bar_count'] = max(1, int(el.get('barCount', 8)))
|
|
p['bar_spacing'] = max(0, int(el.get('barSpacing', 1)))
|
|
binding = el.get('binding', {})
|
|
p['binding_key'] = binding.get('key', '')
|
|
p['fill_tuple'] = f"({el.get('r', 80)}, {el.get('g', 200)}, {el.get('b', 120)})"
|
|
bg = [el.get('bgR', 30), el.get('bgG', 30), el.get('bgB', 30)] if el.get('hasBg', False) else None
|
|
p['bg_tuple'] = _as_fill_filter(bg)
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'gauge':
|
|
x_expr = _compute_pos_expr(el.get('x', 0), x_anchor, 'width')
|
|
y_expr = _compute_pos_expr(el.get('y', 0), y_anchor, 'height')
|
|
w = _safe_int(el.get('width'), 32, 0, 4096)
|
|
h = _safe_int(el.get('height'), 32, 0, 4096)
|
|
p['x_expr'] = x_expr
|
|
p['y_expr'] = y_expr
|
|
p['x2_expr'] = f"({x_expr}) + {w}"
|
|
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'] = _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
|
|
)
|
|
p['track_tuple'] = _as_fill_filter(track)
|
|
binding = el.get('binding', {})
|
|
p['binding_key'] = binding.get('key', '')
|
|
font_key = el.get('font', 'four_by_six')
|
|
p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'extra_small_font')
|
|
p['show_label'] = bool(el.get('showLabel', True))
|
|
p['label_tuple'] = f"({el.get('labelR', 200)}, {el.get('labelG', 200)}, {el.get('labelB', 200)})"
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
elif t == 'marquee':
|
|
font_key = el.get('font', 'press_start')
|
|
p['font_attr'] = _FONT_ATTR_MAP.get(font_key, 'regular_font')
|
|
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)
|
|
p['gap'] = int(el.get('gap', 16))
|
|
p['scroll_speed'] = max(1, int(el.get('scrollSpeed', 1)))
|
|
p['direction'] = el.get('direction', 'left')
|
|
# Data key stored in self._data for stateful scrolling across display() calls
|
|
raw_id = str(el.get('id', 0)).replace('-', '_')
|
|
p['data_key'] = f"mq_{raw_id}"
|
|
p['blink'] = bool(el.get('blink', False))
|
|
|
|
result.append(p)
|
|
return result
|
|
|
|
|
|
def _generate_plugin_files(data: dict) -> dict:
|
|
"""
|
|
Generate all plugin file contents as strings.
|
|
|
|
Returns dict: {'manager.py', 'manifest.json', 'config_schema.json', 'requirements.txt'}
|
|
Raises ValueError with a human-readable message on any validation failure.
|
|
"""
|
|
metadata = data.get('metadata', {})
|
|
elements = data.get('elements', [])
|
|
data_model = data.get('dataModel', {})
|
|
config_vars = data_model.get('configVars', [])
|
|
|
|
plugin_id = metadata.get('id', '').strip()
|
|
if not _PLUGIN_ID_RE.match(plugin_id):
|
|
raise ComposerInputError(
|
|
'Plugin ID must start with a lowercase letter and contain only '
|
|
'lowercase letters, numbers, and hyphens (max 63 chars).'
|
|
)
|
|
|
|
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:
|
|
raise ComposerInputError('Author is required.')
|
|
|
|
version = metadata.get('version', '1.0.0').strip()
|
|
|
|
# Validate config var keys are valid Python identifiers
|
|
for cv in config_vars:
|
|
key = cv.get('key', '')
|
|
if not _PYTHON_IDENT_RE.match(key):
|
|
raise ComposerInputError(f'Config variable key "{key}" is not a valid Python identifier.')
|
|
# A keyword produces `self.class = ...`, which the ast.parse check
|
|
# below does catch -- but as "Generated code has a syntax error:
|
|
# invalid syntax (line 17)", which tells the user nothing about which
|
|
# field to fix.
|
|
if keyword.iskeyword(key) or keyword.issoftkeyword(key):
|
|
raise ComposerInputError(
|
|
f'Config variable key "{key}" is a Python keyword.')
|
|
# These generate *valid* code that silently shadows the plugin's own
|
|
# state. "config" is the worst: the assignment runs immediately after
|
|
# super().__init__(), so
|
|
# self.config = config.get("config", "x")
|
|
# replaces the plugin's config dict with a string and every later
|
|
# self.config.get(...) fails at runtime.
|
|
if key in _RESERVED_ATTRS:
|
|
raise ComposerInputError(
|
|
f'Config variable key "{key}" is reserved by BasePlugin.')
|
|
|
|
class_name = _to_class_name(plugin_name)
|
|
# Only consider visible elements for code generation flags
|
|
visible_elements = [e for e in elements if e.get('visible') is not False]
|
|
processed = _preprocess_elements(elements)
|
|
has_clock = any(e.get('type') == 'clock' for e in visible_elements)
|
|
has_blink = any(e.get('blink') for e in visible_elements)
|
|
has_countdown = any(e.get('type') == 'countdown' for e in visible_elements)
|
|
_var_re = re.compile(r'\{[a-zA-Z_]\w*\}')
|
|
has_text_template = any(
|
|
e.get('type') == 'text' and (
|
|
_var_re.search(e.get('text', '') or '') or
|
|
_var_re.search(e.get('text2', '') or '')
|
|
)
|
|
for e in visible_elements
|
|
)
|
|
|
|
# Background fill color (None → don't render, use LED panel's native black)
|
|
bg_color: str | None = None
|
|
bg_raw = metadata.get('bgColor')
|
|
if isinstance(bg_raw, dict):
|
|
r, g, b = int(bg_raw.get('r', 0)), int(bg_raw.get('g', 0)), int(bg_raw.get('b', 0))
|
|
if r or g or b:
|
|
bg_color = f'({r}, {g}, {b})'
|
|
|
|
# Render manager.py
|
|
env = _get_jinja_env()
|
|
try:
|
|
tmpl = env.get_template('manager.py.j2')
|
|
except jinja2.TemplateNotFound:
|
|
raise ComposerInputError('Code generation template not found. This is a server configuration issue.')
|
|
|
|
manager_py = tmpl.render(
|
|
plugin_name=plugin_name,
|
|
class_name=class_name,
|
|
plugin_id=plugin_id,
|
|
generated_date=datetime.now().strftime('%Y-%m-%d'),
|
|
config_vars=config_vars,
|
|
elements=processed,
|
|
has_clock=has_clock,
|
|
has_blink=has_blink,
|
|
has_countdown=has_countdown,
|
|
has_text_template=has_text_template,
|
|
bg_color=bg_color,
|
|
)
|
|
|
|
# Syntax-check the generated Python
|
|
try:
|
|
ast.parse(manager_py)
|
|
except SyntaxError as exc:
|
|
raise ComposerInputError(f'Generated code has a syntax error: {exc}') from exc
|
|
|
|
# Build manifest
|
|
manifest = {
|
|
'id': plugin_id,
|
|
'name': plugin_name,
|
|
'version': version,
|
|
'author': author,
|
|
'description': metadata.get('description', 'Custom plugin created with LEDMatrix Plugin Composer'),
|
|
'category': metadata.get('category', 'custom'),
|
|
'tags': ['composer', 'custom'],
|
|
'entry_point': 'manager.py',
|
|
'class_name': class_name,
|
|
'display_modes': [plugin_id],
|
|
'compatible_versions': ['>=2.0.0'],
|
|
'last_updated': datetime.now().strftime('%Y-%m-%d'),
|
|
'update_interval': int(metadata.get('update_interval', 60)),
|
|
'default_duration': float(metadata.get('display_duration', 15)),
|
|
'versions': [
|
|
{'released': datetime.now().strftime('%Y-%m-%d'), 'version': version}
|
|
],
|
|
}
|
|
|
|
# Validate manifest against the project's schema
|
|
if composer_bp.project_root:
|
|
schema_path = Path(composer_bp.project_root) / 'schema' / 'manifest_schema.json'
|
|
if schema_path.exists():
|
|
schema = json.loads(schema_path.read_text())
|
|
validator = jsonschema.Draft7Validator(schema)
|
|
errors = list(validator.iter_errors(manifest))
|
|
if errors:
|
|
msgs = '; '.join(e.message for e in errors[:3])
|
|
raise ComposerInputError(f'Manifest validation failed: {msgs}')
|
|
|
|
# Build config_schema
|
|
type_map = {
|
|
'string': {'type': 'string'},
|
|
'number': {'type': 'number', 'minimum': 0},
|
|
'boolean': {'type': 'boolean'},
|
|
'color': {
|
|
'type': 'array',
|
|
'items': {'type': 'integer', 'minimum': 0, 'maximum': 255},
|
|
'minItems': 3,
|
|
'maxItems': 3,
|
|
},
|
|
}
|
|
|
|
config_properties = {
|
|
'enabled': {'type': 'boolean', 'default': True},
|
|
'display_duration': {'type': 'number', 'minimum': 1, 'default': float(metadata.get('display_duration', 15))},
|
|
}
|
|
for cv in config_vars:
|
|
cv_type = cv.get('type', 'string')
|
|
prop = dict(type_map.get(cv_type, {'type': 'string'}))
|
|
if cv.get('description'):
|
|
prop['description'] = cv['description']
|
|
if cv.get('label'):
|
|
prop['title'] = cv['label']
|
|
default = cv.get('default', '')
|
|
if cv_type == 'number':
|
|
try:
|
|
prop['default'] = float(default) if default != '' else 0
|
|
except (TypeError, ValueError):
|
|
prop['default'] = 0
|
|
elif cv_type == 'boolean':
|
|
prop['default'] = bool(default)
|
|
else:
|
|
prop['default'] = default
|
|
config_properties[cv['key']] = prop
|
|
|
|
config_schema = {
|
|
'$schema': 'http://json-schema.org/draft-07/schema#',
|
|
'type': 'object',
|
|
'properties': config_properties,
|
|
}
|
|
|
|
return {
|
|
'manager.py': manager_py,
|
|
'manifest.json': json.dumps(manifest, indent=2),
|
|
'config_schema.json': json.dumps(config_schema, indent=2),
|
|
'requirements.txt': '',
|
|
}
|
|
|
|
|
|
class ComposerInputError(ValueError):
|
|
"""A validation failure whose message is safe to show the caller.
|
|
|
|
_generate_plugin_files raises this for input the user can fix. Anything
|
|
else reaching the handlers is unexpected, and its text may name internal
|
|
paths or library internals, so it is logged and answered generically.
|
|
"""
|
|
|
|
|
|
def _plugin_dir(plugin_id: str) -> Optional[Path]:
|
|
"""Resolve a plugin directory, refusing anything outside plugins_dir.
|
|
|
|
_PLUGIN_ID_RE already rejects '/', '.' and '..', so this cannot currently
|
|
fail -- every traversal payload is blocked before it gets here. It exists
|
|
anyway for two reasons: the guarantee then lives with the path building
|
|
rather than in a regex several hundred lines away, so loosening that regex
|
|
later cannot silently open a traversal; and it is the form static analysis
|
|
recognises, which is why CodeQL reported sixteen path-injection alerts
|
|
against code that was already safe.
|
|
|
|
Returns None for a malformed id or one that escapes the base. It returns
|
|
rather than raises so the handlers answer with a fixed literal: routing a
|
|
caught exception's text into a response is what py/stack-trace-exposure
|
|
flags, and there is nothing here a caller needs beyond "that id is not ok".
|
|
"""
|
|
if not _PLUGIN_ID_RE.match(plugin_id or ''):
|
|
return None
|
|
# secure_filename strips path separators and traversal. Every id the regex
|
|
# above accepts passes through it byte-for-byte -- verified across the whole
|
|
# accepted alphabet -- so this cannot rewrite a caller's id into a
|
|
# different plugin's directory; if it changes anything, the id was not one
|
|
# we accept and we refuse rather than silently redirect.
|
|
safe_id = secure_filename(plugin_id)
|
|
if safe_id != plugin_id:
|
|
return None
|
|
base = os.path.realpath(str(composer_bp.plugins_dir))
|
|
candidate = os.path.realpath(os.path.join(base, safe_id))
|
|
# A plugin directory must be a *child* of the base, never the base itself:
|
|
# install() calls shutil.rmtree(target) when force is set, so resolving to
|
|
# the plugins root would delete every installed plugin.
|
|
#
|
|
# commonpath, not startswith: "/plugins-evil" starts with "/plugins" but is
|
|
# a different directory. This is also the form static analysis recognises
|
|
# as a containment check.
|
|
if candidate == base or os.path.commonpath([base, candidate]) != base:
|
|
return None
|
|
return Path(candidate)
|
|
|
|
|
|
def _save_composer_state(target_dir: Path, payload: dict) -> None:
|
|
"""Persist the raw composer payload alongside the generated plugin files."""
|
|
(target_dir / '_composer_state.json').write_text(
|
|
json.dumps(payload, indent=2, ensure_ascii=False), encoding='utf-8'
|
|
)
|
|
|
|
|
|
def _pack_zip(files: dict, plugin_id: str) -> io.BytesIO:
|
|
"""Pack generated plugin files into an in-memory ZIP."""
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, 'w', compression=zipfile.ZIP_DEFLATED) as zf:
|
|
for filename, content in files.items():
|
|
info = zipfile.ZipInfo(f'{plugin_id}/{filename}')
|
|
info.compress_type = zipfile.ZIP_DEFLATED
|
|
zf.writestr(info, content.encode('utf-8') if isinstance(content, str) else content)
|
|
buf.seek(0)
|
|
return buf
|
|
|
|
|
|
# ── Routes ────────────────────────────────────────────────────────────────────
|
|
|
|
@composer_bp.route('/')
|
|
def index():
|
|
return render_template('v3/composer.html')
|
|
|
|
|
|
@composer_bp.route('/api/generate', methods=['POST'])
|
|
def generate_zip():
|
|
data = request.get_json(force=True, silent=True)
|
|
if not data:
|
|
return jsonify({'status': 'error', 'message': 'No JSON body'}), 400
|
|
try:
|
|
files = _generate_plugin_files(data)
|
|
except ComposerInputError as exc:
|
|
return jsonify({'status': 'error', 'message': str(exc)}), 422
|
|
except ValueError as exc:
|
|
# Not one of ours: the text may name internal paths or library
|
|
# internals, so log it and answer generically.
|
|
logger.exception('Unexpected error generating plugin files: %s', exc)
|
|
return jsonify({'status': 'error', 'message': 'Could not generate plugin files'}), 422
|
|
|
|
# .strip() to match _generate_plugin_files, which strips before it
|
|
# validates. Without it " my-plugin " generates successfully and then
|
|
# fails the id check here, which reads as a bug in the generator.
|
|
plugin_id = data.get('metadata', {}).get('id', 'plugin').strip() or 'plugin'
|
|
files['_composer_state.json'] = json.dumps(data, indent=2, ensure_ascii=False)
|
|
zip_buf = _pack_zip(files, plugin_id)
|
|
return send_file(
|
|
zip_buf,
|
|
mimetype='application/zip',
|
|
as_attachment=True,
|
|
download_name=f'{plugin_id}.zip',
|
|
)
|
|
|
|
|
|
@composer_bp.route('/api/install', methods=['POST'])
|
|
def install_locally():
|
|
if not composer_bp.plugins_dir:
|
|
return jsonify({'status': 'error', 'message': 'Plugin directory not configured'}), 503
|
|
|
|
data = request.get_json(force=True, silent=True)
|
|
if not data:
|
|
return jsonify({'status': 'error', 'message': 'No JSON body'}), 400
|
|
|
|
try:
|
|
files = _generate_plugin_files(data)
|
|
except ComposerInputError as exc:
|
|
return jsonify({'status': 'error', 'message': str(exc)}), 422
|
|
except ValueError as exc:
|
|
# Not one of ours: the text may name internal paths or library
|
|
# internals, so log it and answer generically.
|
|
logger.exception('Unexpected error generating plugin files: %s', exc)
|
|
return jsonify({'status': 'error', 'message': 'Could not generate plugin files'}), 422
|
|
|
|
plugin_id = data.get('metadata', {}).get('id', '').strip()
|
|
# _generate_plugin_files() above already validates metadata.id via this
|
|
# same regex before it will return, but that guarantee lives in a
|
|
# different function -- re-check here, at the point the path is actually
|
|
# built, so this route stays safe on its own if that call is ever
|
|
# reordered or changed.
|
|
target = _plugin_dir(plugin_id)
|
|
if target is None:
|
|
return jsonify({'status': 'error', 'message': 'Invalid plugin ID'}), 400
|
|
force = bool(data.get('_force', False))
|
|
|
|
if target.exists() and not force:
|
|
return jsonify({
|
|
'status': 'conflict',
|
|
'message': f'Plugin "{plugin_id}" is already installed.',
|
|
}), 409
|
|
|
|
try:
|
|
if target.exists() and force:
|
|
import shutil as _shutil
|
|
_shutil.rmtree(target)
|
|
target.mkdir(parents=True, exist_ok=False)
|
|
for filename, content in files.items():
|
|
(target / filename).write_text(content, encoding='utf-8')
|
|
_save_composer_state(target, data)
|
|
except OSError as exc:
|
|
logger.error('Failed to write plugin files for %s: %s', plugin_id, exc)
|
|
return jsonify({'status': 'error', 'message': 'Failed to write plugin files'}), 500
|
|
|
|
# Trigger plugin discovery so it shows up in the Plugin Manager immediately
|
|
if composer_bp.plugin_manager:
|
|
try:
|
|
composer_bp.plugin_manager.discover_plugins()
|
|
except Exception as exc:
|
|
logger.warning('discover_plugins() failed after composer install: %s', exc)
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Plugin "{plugin_id}" installed successfully.',
|
|
'plugin_id': plugin_id,
|
|
})
|
|
|
|
|
|
@composer_bp.route('/api/fonts/<font_name>')
|
|
def serve_font(font_name):
|
|
"""Serve an allowlisted font file for canvas FontFace loading."""
|
|
# Build the path from the allowlist entry, not from the request value.
|
|
# They are equal strings, so this changes nothing at runtime -- but the
|
|
# name that reaches the filesystem now provably originates in a module
|
|
# constant, which is the difference between "guarded" and "not derived
|
|
# from user input at all".
|
|
allowed_name = next((f for f in sorted(_ALLOWED_FONTS) if f == font_name), None)
|
|
if allowed_name is None:
|
|
return '', 404
|
|
if not composer_bp.project_root:
|
|
return '', 503
|
|
font_path = Path(composer_bp.project_root) / 'assets' / 'fonts' / allowed_name
|
|
if not font_path.exists():
|
|
return '', 404
|
|
return send_file(str(font_path), mimetype='font/ttf')
|
|
|
|
|
|
@composer_bp.route('/api/validate-id/<plugin_id>')
|
|
def validate_id(plugin_id):
|
|
"""Check whether a plugin ID is valid and available."""
|
|
if not _PLUGIN_ID_RE.match(plugin_id):
|
|
return jsonify({'valid': False, 'available': False, 'reason': 'Invalid format'})
|
|
if composer_bp.plugins_dir:
|
|
resolved = _plugin_dir(plugin_id)
|
|
if resolved is None:
|
|
return jsonify({'valid': False, 'available': False, 'reason': 'Invalid format'})
|
|
taken = resolved.exists()
|
|
if taken:
|
|
return jsonify({'valid': True, 'available': False, 'reason': 'Already installed'})
|
|
return jsonify({'valid': True, 'available': True})
|
|
|
|
|
|
@composer_bp.route('/api/plugins')
|
|
def list_plugins():
|
|
"""List installed plugins, flagging which ones have a saved composer state."""
|
|
if not composer_bp.plugins_dir:
|
|
return jsonify([])
|
|
plugins_dir = Path(composer_bp.plugins_dir)
|
|
if not plugins_dir.is_dir():
|
|
# Configured but not created yet -- a fresh install, or a bad path.
|
|
# iterdir() raises FileNotFoundError/NotADirectoryError here, which
|
|
# surfaced as a 500 rather than "no plugins".
|
|
logger.warning("Plugin directory %s does not exist", plugins_dir)
|
|
return jsonify([])
|
|
results = []
|
|
for entry in sorted(plugins_dir.iterdir()):
|
|
if not entry.is_dir():
|
|
continue
|
|
manifest_path = entry / 'manifest.json'
|
|
if not manifest_path.exists():
|
|
continue
|
|
try:
|
|
manifest = json.loads(manifest_path.read_text())
|
|
except Exception as e:
|
|
logger.warning("Skipping %s: unreadable manifest.json (%s)", entry.name, e)
|
|
continue
|
|
has_state = (entry / '_composer_state.json').exists()
|
|
results.append({
|
|
'id': manifest.get('id', entry.name),
|
|
'name': manifest.get('name', entry.name),
|
|
'version': manifest.get('version', ''),
|
|
'author': manifest.get('author', ''),
|
|
'has_composer_state': has_state,
|
|
})
|
|
return jsonify(results)
|
|
|
|
|
|
@composer_bp.route('/api/preview', methods=['POST'])
|
|
def preview_code():
|
|
"""Generate plugin files and return them as JSON for the code preview modal."""
|
|
data = request.get_json(force=True, silent=True)
|
|
if not data:
|
|
return jsonify({'status': 'error', 'message': 'No JSON body'}), 400
|
|
try:
|
|
files = _generate_plugin_files(data)
|
|
except ComposerInputError as exc:
|
|
return jsonify({'status': 'error', 'message': str(exc)}), 422
|
|
except ValueError as exc:
|
|
# Not one of ours: the text may name internal paths or library
|
|
# internals, so log it and answer generically.
|
|
logger.exception('Unexpected error generating plugin files: %s', exc)
|
|
return jsonify({'status': 'error', 'message': 'Could not generate plugin files'}), 422
|
|
return jsonify({
|
|
'status': 'ok',
|
|
'files': {
|
|
'manager.py': files['manager.py'],
|
|
'manifest.json': files['manifest.json'],
|
|
'config_schema.json': files['config_schema.json'],
|
|
},
|
|
})
|
|
|
|
|
|
@composer_bp.route('/api/load/<plugin_id>')
|
|
def load_plugin(plugin_id):
|
|
"""Load a plugin's composer state for editing.
|
|
|
|
If a _composer_state.json exists, return it verbatim.
|
|
Otherwise, extract config vars from config_schema.json for a partial import.
|
|
"""
|
|
if not composer_bp.plugins_dir:
|
|
return jsonify({'status': 'error', 'message': 'Plugin directory not configured'}), 503
|
|
plugin_dir = _plugin_dir(plugin_id)
|
|
if plugin_dir is None:
|
|
return jsonify({'status': 'error', 'message': 'Invalid plugin ID'}), 400
|
|
if not plugin_dir.exists():
|
|
return jsonify({'status': 'error', 'message': 'Plugin not found'}), 404
|
|
|
|
# Full composer state
|
|
state_path = plugin_dir / '_composer_state.json'
|
|
if state_path.exists():
|
|
try:
|
|
state = json.loads(state_path.read_text())
|
|
return jsonify({'status': 'ok', 'source': 'composer', 'state': state})
|
|
except Exception as exc:
|
|
logger.error('Failed to read composer state for %s: %s', plugin_id, exc)
|
|
return jsonify({'status': 'error', 'message': 'Failed to read state'}), 500
|
|
|
|
# Partial import from config_schema.json
|
|
schema_path = plugin_dir / 'config_schema.json'
|
|
manifest_path = plugin_dir / 'manifest.json'
|
|
config_vars = []
|
|
|
|
if schema_path.exists():
|
|
try:
|
|
schema = json.loads(schema_path.read_text())
|
|
props = schema.get('properties', {})
|
|
skip = {'enabled', 'display_duration', 'update_interval'}
|
|
type_map = {'boolean': 'boolean', 'number': 'number', 'integer': 'number', 'string': 'string'}
|
|
for key, prop in props.items():
|
|
if key in skip:
|
|
continue
|
|
prop_type = prop.get('type', 'string')
|
|
if isinstance(prop_type, list):
|
|
prop_type = next((t for t in prop_type if t != 'null'), 'string')
|
|
# Detect color arrays
|
|
if prop_type == 'array' and prop.get('maxItems') == 3:
|
|
cv_type = 'color'
|
|
else:
|
|
cv_type = type_map.get(prop_type, 'string')
|
|
config_vars.append({
|
|
'key': key,
|
|
'label': prop.get('title', key.replace('_', ' ').title()),
|
|
'type': cv_type,
|
|
'default': prop.get('default', ''),
|
|
'description': prop.get('description', ''),
|
|
})
|
|
except Exception as e:
|
|
logger.warning("Failed to parse config_schema.json for %s: %s", plugin_id, e)
|
|
|
|
manifest = {}
|
|
if manifest_path.exists():
|
|
try:
|
|
manifest = json.loads(manifest_path.read_text())
|
|
except (OSError, ValueError) as exc:
|
|
# Swallowing this left "partial import produced nothing" with no
|
|
# way to tell a malformed manifest from an absent one.
|
|
logger.warning("Failed to parse manifest.json for %s: %s", plugin_id, exc)
|
|
|
|
partial_state = {
|
|
'composer_version': '1.0',
|
|
'metadata': {
|
|
'id': manifest.get('id', plugin_id),
|
|
'name': manifest.get('name', plugin_id),
|
|
'author': manifest.get('author', ''),
|
|
'version': manifest.get('version', '1.0.0'),
|
|
'description': manifest.get('description', ''),
|
|
'category': manifest.get('category', 'custom'),
|
|
'display_duration': manifest.get('default_duration', 15),
|
|
'update_interval': manifest.get('update_interval', 60),
|
|
'api_requirements': manifest.get('api_requirements', []),
|
|
},
|
|
'elements': [],
|
|
'dataModel': {'configVars': config_vars, 'dataSources': [], 'computedVars': []},
|
|
}
|
|
return jsonify({'status': 'ok', 'source': 'schema_import', 'state': partial_state})
|