mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-23 03:18:15 +00:00
ChuckBuildsandClaude Opus 5
37fc1b56b5
Three more routes into the generated source, plus a fix to one of my own tests
that was checking the wrong branch.
Prefixed colour channels were interpolated raw
----------------------------------------------
Five tuples were built without coercion:
p['fill_tuple'] = f"({el.get('r', 100)}, {el.get('g', 200)}, ...)"
p['empty_tuple'] = f"({el.get('emptyR', 50)}, ...)"
p['label_tuple'] = f"({el.get('labelR', 200)}, ...)"
so progress_bar, pips, sparkline and gauge took arbitrary expressions the same
way width/height did. Confirmed: every one of the five put __import__ into the
generated source. They now go through a new _rgb_tuple helper, which _rgb_expr
also delegates to.
The pre-existing colour test only covered r/g/b on a text element, which is why
the prefixed channels and these four types were never exercised.
Non-finite numbers escaped as a 500
-----------------------------------
json.loads accepts Infinity/-Infinity/NaN by default and Flask's get_json
passes them straight through, so a payload can hand _safe_int a non-finite
float. int(inf) raises OverflowError, which is neither ValueError nor
ComposerInputError, so it escaped both handlers and surfaced as a 500 with a
traceback rather than a 422. Verified end to end through Flask's parser.
Marquee ids reached the source as identifiers
---------------------------------------------
data_key is spliced UNQUOTED into variable names (_{{ data_key }}_text = ...)
and only '-' was normalised. A punctuated id landed in the generated source as
code. ast.parse caught it, so this was not exploitable, but the caller got an
opaque "Generated code has a syntax error" instead of being told the id was
unusable -- the same failure mode as the empty-block bug. Now restricted to
identifier characters and bounded to 64.
The line-anchor test was testing the wrong branch
-------------------------------------------------
test_line_branch_applies_the_anchor_offset searched the whole file for
"case 'line': {". getBoundingBox has one too and comes first, so the assertion
was reading the bounding-box branch: stripping the anchor offset from
_drawElement left all 11 checks green. Both line tests are now scoped to their
own function via tree-sitter, so they cannot be satisfied by the same branch.
Tests: 35 of the injection suite's checks fail against the reverted fixes; the
scoped line test fails when _drawElement's offset is removed. Full suite 4059
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
1085 lines
47 KiB
Python
1085 lines
47 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 math
|
|
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.
|
|
"""
|
|
# json.loads accepts Infinity/-Infinity/NaN by default and Flask's
|
|
# get_json passes them straight through, so a payload can hand this a
|
|
# non-finite float. int(inf) raises OverflowError, which is neither
|
|
# ValueError nor ComposerInputError -- it escaped both handlers and became
|
|
# a 500 with a traceback instead of a 422.
|
|
if isinstance(value, float) and not math.isfinite(value):
|
|
return default
|
|
try:
|
|
out = int(value)
|
|
except (TypeError, ValueError, OverflowError):
|
|
return default
|
|
if lo is not None:
|
|
out = max(lo, out)
|
|
if hi is not None:
|
|
out = min(hi, out)
|
|
return out
|
|
|
|
|
|
def _rgb_tuple(el: dict, keys, defaults) -> str:
|
|
"""A colour tuple literal from coerced, clamped values under *keys*.
|
|
|
|
Several element types carry prefixed channels (emptyR/G/B, labelR/G/B)
|
|
rather than r/g/b. Those were interpolated straight into the generated
|
|
source, so they were an injection route exactly like an uncoerced
|
|
dimension. Every channel now goes through _safe_int.
|
|
"""
|
|
return "(" + ", ".join(
|
|
str(_safe_int(el.get(k), d, 0, 255)) for k, d in zip(keys, defaults)
|
|
) + ")"
|
|
|
|
|
|
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 _rgb_tuple(el, ('r', 'g', 'b'), (dr, dg, db))
|
|
|
|
|
|
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'] = _rgb_tuple(el, ('r', 'g', 'b'), (100, 200, 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'] = _rgb_tuple(el, ('r', 'g', 'b'), (255, 200, 0))
|
|
p['empty_tuple'] = _rgb_tuple(el, ('emptyR', 'emptyG', 'emptyB'), (50, 50, 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'] = _rgb_tuple(el, ('r', 'g', 'b'), (80, 200, 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'] = _rgb_tuple(el, ('labelR', 'labelG', 'labelB'), (200, 200, 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. It is spliced UNQUOTED into variable names
|
|
# (_{{ data_key }}_text = ...), so anything that is not an
|
|
# identifier character lands in the generated source as code.
|
|
# ast.parse catches the result, but the caller then gets an opaque
|
|
# "Generated code has a syntax error" instead of being told the id
|
|
# is unusable. Restrict it to identifier characters and bound it.
|
|
raw_id = re.sub(r'[^A-Za-z0-9_]', '_', str(el.get('id', 0)))[:64]
|
|
p['data_key'] = f"mq_{raw_id or '0'}"
|
|
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})
|