mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 22:35:08 +00:00
refactor(web): delete dead routes, JS files and duplicate definitions (#609)
* refactor(web): drop validators nothing calls escape_html, validate_image_url, validate_font_awesome_class, validate_mime_type, validate_numeric_range, validate_string_length and sanitize_plugin_config had no callers outside their own tests. Only validate_file_upload (fonts upload) is imported by the web interface. dedup_unique_arrays is kept: its one caller in save_plugin_config was removed by the unrelated sync PR (#330), which looks accidental. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(api): remove the music-auth and of-the-day JSON routes POST /plugins/authenticate/spotify and /plugins/authenticate/ytm had no caller but their tests: the music plugin authenticates through its web_ui_actions (authenticate_spotify.py / authenticate_ytm.py) via /plugins/action. POST /plugins/of-the-day/json/upload and /json/delete looked the plugin up by the id ledmatrix-of-the-day (its manifest id is of-the-day), were reachable only from a file_type "json" upload field that no schema declares, and put the plugin directory on sys.path per request to import scripts.update_config. of-the-day manages its files through plugin-file-manager and its own web_ui_actions. The of-the-day branch of GET /plugins/config stays: it matches the real manifest id and still merges the on-disk category files into the form. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(api): read managers only from the blueprints api_v3/__init__.py and pages_v3.py declared module globals (plugin_store_manager, saved_repositories_manager, schema_manager, operation_queue, plugin_state_manager, operation_history, sync_manager, config_manager, plugin_manager) that nothing assigns: app.py sets the managers as attributes on the Blueprint objects, and every route reads them there. The one reader, backup restore's fallback to the module plugin_store_manager, could only ever fall back to None. _ensure_cache_manager() built a second CacheManager in the web process instead of using the one app.py puts on api_v3. The display routes now read api_v3.cache_manager, creating it on the blueprint only when nothing set it (the same None handling as the /cache routes). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore(web): drop run.sh and the unused log_config_change web_interface/run.sh was referenced only by web_interface/README.md; the service starts the UI through scripts/utils/start_web_conditionally.py and the README already documents `python3 web_interface/start.py`. log_config_change() in web_interface/logging_config.py was never called. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(web): delete unreferenced store_manager.js, diff_viewer.js, htmx-sse.js - js/plugins/store_manager.js (window.PluginStoreManager) and js/config/diff_viewer.js (window.ConfigDiffViewer) were loaded on every page but nothing reads either global. - htmx-sse.js (plus its CDN fallback) was loaded after HTMX, but no template or plugin page uses sse-connect / hx-ext="sse": the live streams run through LEDStreams in app-shell.js. js/plugins/state_manager.js stays: install_manager.js's updateAll() reads and refreshes window.PluginStateManager. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(web): remove app.js helpers nothing calls - hexToRgb, rgbToHex, validateForm, uploadFont and switchTab (whose 'switch-tab' event had no listener) have no caller in the templates, static JS or the plugin monorepo. - installPlugin: plugins_manager.js (loaded last) assigns window.installPlugin, and its own store cards are the only callers. - The showNotification fallback could never install: app-shell.js is deferred ahead of app.js and defines the same fallback at top level. - performanceMonitor only logged with ?debug=perf and read an unset this.measures; the marks it took on every load had no reader. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(web): drop app-shell.js refreshPlugin A top-level function in app-shell.js, so a window global, but nothing calls it (no inline handler, no window lookup, no string-built name). The other plugin actions in that block stay. updatePlugin is the live window.updatePlugin: plugins_manager.js only installs its own copy when none exists. uninstallPlugin/pollUninstallOperation, updateAllPlugins, executePluginAction and toggleNestedSection are replaced by later deferred scripts, but a click that lands while those scripts are still downloading reaches the app-shell copies, so removing them is not a pure no-op. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(web): remove definitions plugins_manager.js always overrides All of these are replaced before anything can call them, checked against the load order in base.html and the live window.* values: - openOnDemandModal/requestOnDemandStop stubs: the IIFE later in the same script assigns the real functions synchronously. - updatePlugin and uninstallPlugin stubs (`window.X || stub`): app-shell.js already defined both, so the fallback never installed. Same for the later updatePlugin override, gated on the live function containing '[UPDATE]', which app-shell.js's never does. - The first addArrayObjectItem/removeArrayObjectItem: reassigned by the top-level copies after the IIFE. - The first `function formatDate` in the IIFE: a later declaration of the same name in the same scope wins. - deleteUploadedImage, getCurrentImages, showUploadProgress, formatFileSize and getScheduleSummary: character-for-character copies of js/widgets/file-upload.js, which stays the owner. - `typeof X === 'undefined'` fallbacks and `typeof X !== 'undefined'` re-exports after the IIFE: always false, or a self-assignment. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(web): render the shell directly and delete index.html index.html extended base.html with {% block content %}, but base.html defines no blocks, so none of index.html ever rendered: rendering both with jinja2 gives byte-identical output. index() still loaded the config, read config.json and config_secrets.json raw and json.dumps'd them on every page load for variables base.html never reads, and flashed errors that base.html never shows. It now renders base.html with no context. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): stop htmx-config.js replacing console.error and console.warn It swapped both globals for filters that dropped any error mentioning insertBefore / "Cannot read properties of null" when "htmx" appeared in the message or stack, and a list of Permissions-Policy warnings. That hid real errors from every script on the page, and made every logged error and warning report htmx-config.js as its source. The beforeSwap target validation above it, which prevents the insertBefore errors in the first place, stays. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore(web): quiet the widget load announcements and debug logs About 30 lines hit the console on every page load: one "... widget registered" per widget file, one "[WidgetRegistry] Registered widget: X" per registration, plus the registry, base widget and plugin loader announcing themselves. The load-time announcements are removed; the per-call ones (registry register, plugin widget loads, "Render called") now go through the page's debugLog switch (localStorage.pluginDebug), guarded because the widgets also load in node tests without it. fonts.html and wifi.html debug logging goes through debugLog as well. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(api): drop the removed music-auth and of-the-day JSON routes Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
Vendored
-32
@@ -358,22 +358,6 @@
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/plugins/authenticate/spotify",
|
||||
"api_v3.authenticate_spotify",
|
||||
[
|
||||
"OPTIONS",
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/plugins/authenticate/ytm",
|
||||
"api_v3.authenticate_ytm",
|
||||
[
|
||||
"OPTIONS",
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/plugins/calendar/authenticate",
|
||||
"api_v3.authenticate_calendar",
|
||||
@@ -511,22 +495,6 @@
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/plugins/of-the-day/json/delete",
|
||||
"api_v3.delete_of_the_day_json",
|
||||
[
|
||||
"OPTIONS",
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/plugins/of-the-day/json/upload",
|
||||
"api_v3.upload_of_the_day_json",
|
||||
[
|
||||
"OPTIONS",
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/plugins/operation/<operation_id>",
|
||||
"api_v3.get_operation_status",
|
||||
|
||||
@@ -76,13 +76,12 @@ def fresh_web_process(api_v3_module, plugins_dir):
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def display_service():
|
||||
def display_service(api_v3_module):
|
||||
"""Keep on-demand start away from systemctl and the real cache."""
|
||||
with patch('web_interface.blueprints.api_v3.display._ensure_cache_manager') as cache, \
|
||||
patch('web_interface.blueprints.api_v3.display._get_display_service_status') as status:
|
||||
cache.return_value = MagicMock()
|
||||
cache = api_v3_module.api_v3.cache_manager = MagicMock()
|
||||
with patch('web_interface.blueprints.api_v3.display._get_display_service_status') as status:
|
||||
status.return_value = {'active': True}
|
||||
yield cache.return_value
|
||||
yield cache
|
||||
|
||||
|
||||
def _start(client, **body):
|
||||
|
||||
@@ -1,302 +0,0 @@
|
||||
"""
|
||||
Endpoint tests for /plugins/authenticate/spotify and .../ytm.
|
||||
|
||||
The Spotify step-2 handler writes a Python wrapper script to a temp file
|
||||
with the user's redirect URL embedded in it, then runs that file through
|
||||
subprocess. That is the most dangerous shape in the blueprint and had no
|
||||
tests: the URL is user input reaching generated source code.
|
||||
|
||||
The two endpoints are NOT symmetrical, despite the matching names. Only
|
||||
Spotify has a two-step flow, a wrapper script, and a redirect_url; YTM
|
||||
just runs its script directly.
|
||||
|
||||
Regression coverage for one fixed bug: the wrapper file was unlinked in
|
||||
the success/failure branch and again in the TimeoutExpired handler, so
|
||||
any other failure from subprocess.run — the interpreter missing, a fork
|
||||
failure, an interrupted call — left a temp file containing the user's
|
||||
redirect URL behind.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def plugin_dir(tmp_path, api_v3_module):
|
||||
"""A plugin directory containing both auth scripts."""
|
||||
directory = tmp_path / "plugins" / "ledmatrix-music"
|
||||
directory.mkdir(parents=True)
|
||||
(directory / "authenticate_spotify.py").write_text("print('spotify')\n")
|
||||
(directory / "authenticate_ytm.py").write_text("print('ytm')\n")
|
||||
api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
|
||||
return directory
|
||||
|
||||
|
||||
def completed(returncode=0, stdout="ok", stderr=""):
|
||||
return subprocess.CompletedProcess(
|
||||
args=["python3"], returncode=returncode, stdout=stdout, stderr=stderr)
|
||||
|
||||
|
||||
class TestSpotifyPreconditions:
|
||||
URL = "/api/v3/plugins/authenticate/spotify"
|
||||
|
||||
def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
|
||||
api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
|
||||
tmp_path / "not-installed")
|
||||
response = api_v3_client.post(self.URL, json={})
|
||||
assert response.status_code == 404
|
||||
assert response.get_json()["message"] == "Plugin not found"
|
||||
|
||||
def test_none_plugin_directory_is_404(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = None
|
||||
assert api_v3_client.post(self.URL, json={}).status_code == 404
|
||||
|
||||
def test_missing_auth_script_is_404(self, api_v3_client, plugin_dir):
|
||||
(plugin_dir / "authenticate_spotify.py").unlink()
|
||||
response = api_v3_client.post(self.URL, json={})
|
||||
assert response.status_code == 404
|
||||
assert "script not found" in response.get_json()["message"]
|
||||
|
||||
|
||||
class TestSpotifyStepTwo:
|
||||
"""redirect_url present — the wrapper-script path."""
|
||||
|
||||
URL = "/api/v3/plugins/authenticate/spotify"
|
||||
|
||||
def test_success(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed(0, "done")):
|
||||
response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
|
||||
assert response.status_code == 200
|
||||
body = response.get_json()
|
||||
assert body["status"] == "success"
|
||||
assert body["output"] == "done"
|
||||
|
||||
def test_script_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
|
||||
response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
|
||||
assert response.status_code == 400
|
||||
assert response.get_json()["output"] == "outerr"
|
||||
|
||||
def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run",
|
||||
side_effect=subprocess.TimeoutExpired("python3", 120)):
|
||||
response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
|
||||
assert response.status_code == 408
|
||||
assert "timed out" in response.get_json()["message"]
|
||||
|
||||
def test_runs_a_list_argv_never_a_shell(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed()) as run:
|
||||
api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
|
||||
args, kwargs = run.call_args
|
||||
assert isinstance(args[0], list)
|
||||
assert args[0][0] == "python3"
|
||||
assert kwargs.get("shell") in (None, False)
|
||||
|
||||
def test_timeout_is_bounded(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed()) as run:
|
||||
api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
|
||||
assert run.call_args.kwargs["timeout"] == 120
|
||||
|
||||
|
||||
class TestSpotifyWrapperCleanup:
|
||||
URL = "/api/v3/plugins/authenticate/spotify"
|
||||
|
||||
def _wrapper_paths_after(self, api_v3_client, run_mock):
|
||||
"""Run the endpoint and return the wrapper path subprocess saw."""
|
||||
seen = {}
|
||||
|
||||
def capture(args, **kwargs):
|
||||
seen["path"] = args[1]
|
||||
return run_mock(args, **kwargs)
|
||||
|
||||
with patch.object(subprocess, "run", side_effect=capture):
|
||||
api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
|
||||
return seen["path"]
|
||||
|
||||
def test_removed_after_success(self, api_v3_client, plugin_dir):
|
||||
path = self._wrapper_paths_after(api_v3_client, lambda *a, **kw: completed())
|
||||
assert not os.path.exists(path)
|
||||
|
||||
def test_removed_after_script_failure(self, api_v3_client, plugin_dir):
|
||||
path = self._wrapper_paths_after(
|
||||
api_v3_client, lambda *a, **kw: completed(1, "out", "err"))
|
||||
assert not os.path.exists(path)
|
||||
|
||||
def test_removed_after_timeout(self, api_v3_client, plugin_dir):
|
||||
def raise_timeout(*a, **kw):
|
||||
raise subprocess.TimeoutExpired("python3", 120)
|
||||
path = self._wrapper_paths_after(api_v3_client, raise_timeout)
|
||||
assert not os.path.exists(path)
|
||||
|
||||
def test_removed_when_subprocess_cannot_start(self, api_v3_client, plugin_dir):
|
||||
# Regression: cleanup lived in the success/failure branch and in the
|
||||
# TimeoutExpired handler only. An OSError from subprocess.run itself
|
||||
# — no interpreter, fork failure — skipped both and left the wrapper,
|
||||
# which contains the user's redirect URL, on disk.
|
||||
def raise_oserror(*a, **kw):
|
||||
raise OSError("[Errno 12] Cannot allocate memory")
|
||||
path = self._wrapper_paths_after(api_v3_client, raise_oserror)
|
||||
assert not os.path.exists(path)
|
||||
|
||||
|
||||
class TestSpotifyRedirectUrlIsNotInjectable:
|
||||
"""The wrapper embeds redirect_url into generated Python source."""
|
||||
|
||||
URL = "/api/v3/plugins/authenticate/spotify"
|
||||
|
||||
ADVERSARIAL = [
|
||||
'''http://cb/?code=x"''',
|
||||
"""http://cb/?code=x'""",
|
||||
'http://cb/?code=x\\',
|
||||
'http://cb/?code=x\nimport os; os.system("id")',
|
||||
'http://cb/?code=x"""\nimport os\n"""',
|
||||
"http://cb/?code=x'''",
|
||||
'http://cb/?code=x\\"\\n',
|
||||
'"; import os; os.system("id"); "',
|
||||
]
|
||||
|
||||
def _wrapper_source(self, api_v3_client, redirect_url):
|
||||
captured = {}
|
||||
|
||||
def capture(args, **kwargs):
|
||||
captured["source"] = Path(args[1]).read_text()
|
||||
return completed()
|
||||
|
||||
with patch.object(subprocess, "run", side_effect=capture):
|
||||
api_v3_client.post(self.URL, json={"redirect_url": redirect_url})
|
||||
return captured["source"]
|
||||
|
||||
@pytest.mark.parametrize("redirect_url", ADVERSARIAL)
|
||||
def test_wrapper_is_still_valid_python(self, api_v3_client, plugin_dir, redirect_url):
|
||||
# If escaping failed, the generated file would not parse at all.
|
||||
source = self._wrapper_source(api_v3_client, redirect_url)
|
||||
ast.parse(source)
|
||||
|
||||
@pytest.mark.parametrize("redirect_url", ADVERSARIAL)
|
||||
def test_url_survives_as_one_string_literal(
|
||||
self, api_v3_client, plugin_dir, redirect_url):
|
||||
# Stronger than "it parses": the URL must still be a single string
|
||||
# assigned to redirect_url, not code that escaped into statements.
|
||||
source = self._wrapper_source(api_v3_client, redirect_url)
|
||||
tree = ast.parse(source)
|
||||
assigned = [
|
||||
node.value.value for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Assign)
|
||||
and isinstance(node.value, ast.Constant)
|
||||
and any(getattr(t, "id", None) == "redirect_url" for t in node.targets)
|
||||
]
|
||||
assert assigned == [redirect_url.strip()]
|
||||
|
||||
def test_injected_call_does_not_become_a_statement(self, api_v3_client, plugin_dir):
|
||||
source = self._wrapper_source(
|
||||
api_v3_client, 'http://cb/\nimport os; os.system("id")')
|
||||
tree = ast.parse(source)
|
||||
imported = {
|
||||
alias.name for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Import) for alias in node.names
|
||||
}
|
||||
# The wrapper legitimately imports sys, subprocess and os; what it
|
||||
# must not gain is a *call* smuggled in through the URL.
|
||||
calls = [
|
||||
node for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Attribute)
|
||||
and node.func.attr == "system"
|
||||
]
|
||||
assert calls == []
|
||||
|
||||
|
||||
class TestSpotifyStepOne:
|
||||
"""No redirect_url — the OAuth-URL path, which imports the script."""
|
||||
|
||||
URL = "/api/v3/plugins/authenticate/spotify"
|
||||
|
||||
def test_script_without_credentials_helper_is_an_error(
|
||||
self, api_v3_client, plugin_dir):
|
||||
# The stub script defines neither get_auth_url nor
|
||||
# load_spotify_credentials, so no URL can be produced.
|
||||
response = api_v3_client.post(self.URL, json={})
|
||||
assert response.status_code in (400, 500)
|
||||
assert response.get_json()["status"] == "error"
|
||||
|
||||
def test_unusable_credentials_do_not_leak_into_the_response(
|
||||
self, api_v3_client, plugin_dir):
|
||||
(plugin_dir / "authenticate_spotify.py").write_text(
|
||||
"def load_spotify_credentials():\n"
|
||||
" return ('id-abc', 'super-secret-value', None)\n"
|
||||
)
|
||||
response = api_v3_client.post(self.URL, json={})
|
||||
assert "super-secret-value" not in response.get_data(as_text=True)
|
||||
|
||||
def test_script_raising_on_import_is_handled(self, api_v3_client, plugin_dir):
|
||||
(plugin_dir / "authenticate_spotify.py").write_text("raise RuntimeError('boom')\n")
|
||||
response = api_v3_client.post(self.URL, json={})
|
||||
assert response.status_code == 500
|
||||
assert response.get_json()["status"] == "error"
|
||||
|
||||
def test_bodyless_post_reaches_step_one(self, api_v3_client, plugin_dir):
|
||||
# Covered by the silent=True fix: previously a 500 from body parsing.
|
||||
response = api_v3_client.post(self.URL)
|
||||
assert response.status_code in (400, 500)
|
||||
assert response.get_json()["status"] == "error"
|
||||
|
||||
def test_whitespace_redirect_url_is_treated_as_absent(
|
||||
self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed()) as run:
|
||||
api_v3_client.post(self.URL, json={"redirect_url": " "})
|
||||
# Step 2 never runs, so no wrapper is executed.
|
||||
run.assert_not_called()
|
||||
|
||||
|
||||
class TestYouTubeMusic:
|
||||
"""No wrapper script and no redirect_url — deliberately not symmetric."""
|
||||
|
||||
URL = "/api/v3/plugins/authenticate/ytm"
|
||||
|
||||
def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
|
||||
api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
|
||||
tmp_path / "not-installed")
|
||||
assert api_v3_client.post(self.URL).status_code == 404
|
||||
|
||||
def test_missing_script_is_404(self, api_v3_client, plugin_dir):
|
||||
(plugin_dir / "authenticate_ytm.py").unlink()
|
||||
response = api_v3_client.post(self.URL)
|
||||
assert response.status_code == 404
|
||||
assert "script not found" in response.get_json()["message"]
|
||||
|
||||
def test_success(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed(0, "authorized")):
|
||||
response = api_v3_client.post(self.URL)
|
||||
assert response.status_code == 200
|
||||
assert response.get_json()["output"] == "authorized"
|
||||
|
||||
def test_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
|
||||
response = api_v3_client.post(self.URL)
|
||||
assert response.status_code == 400
|
||||
assert response.get_json()["output"] == "outerr"
|
||||
|
||||
def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run",
|
||||
side_effect=subprocess.TimeoutExpired("python3", 60)):
|
||||
assert api_v3_client.post(self.URL).status_code == 408
|
||||
|
||||
def test_runs_the_script_directly_without_a_shell(self, api_v3_client, plugin_dir):
|
||||
with patch.object(subprocess, "run", return_value=completed()) as run:
|
||||
api_v3_client.post(self.URL)
|
||||
args, kwargs = run.call_args
|
||||
assert args[0][0] == "python3"
|
||||
assert args[0][1].endswith("authenticate_ytm.py")
|
||||
assert kwargs.get("shell") in (None, False)
|
||||
assert kwargs["timeout"] == 60
|
||||
@@ -24,7 +24,7 @@ and reported success.
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -41,7 +41,8 @@ def restart_path(api_v3_module):
|
||||
|
||||
plugin_manager and config_manager are set to None so the route takes
|
||||
the simplest path to that branch rather than tripping over unrelated
|
||||
MagicMock plumbing; _ensure_cache_manager, _get_display_service_status,
|
||||
MagicMock plumbing. The cache is the blueprint's cache_manager, which
|
||||
api_v3_module already set to a MagicMock. _get_display_service_status,
|
||||
_stop_display_service and _ensure_display_service_running are bound by
|
||||
value in display.py (see its own docstring), so they are patched on
|
||||
that submodule rather than on the package.
|
||||
@@ -49,16 +50,13 @@ def restart_path(api_v3_module):
|
||||
api_v3_module.api_v3.plugin_manager = None
|
||||
api_v3_module.api_v3.config_manager = None
|
||||
|
||||
with patch("web_interface.blueprints.api_v3.display._ensure_cache_manager") as ensure_cache, \
|
||||
patch("web_interface.blueprints.api_v3.display._get_display_service_status") as get_status, \
|
||||
with patch("web_interface.blueprints.api_v3.display._get_display_service_status") as get_status, \
|
||||
patch("web_interface.blueprints.api_v3.display._stop_display_service") as stop_service, \
|
||||
patch("web_interface.blueprints.api_v3.display._ensure_display_service_running") as ensure_running:
|
||||
ensure_cache.return_value = MagicMock()
|
||||
# Active before the request: service_was_running becomes True.
|
||||
get_status.return_value = {"active": True}
|
||||
ensure_running.return_value = {"active": True}
|
||||
yield {
|
||||
"ensure_cache": ensure_cache,
|
||||
"get_status": get_status,
|
||||
"stop_service": stop_service,
|
||||
"ensure_running": ensure_running,
|
||||
|
||||
@@ -54,16 +54,6 @@ class TestResetPluginConfig:
|
||||
assert api_v3_client.post(self.URL, json={}).status_code != 500
|
||||
|
||||
|
||||
class TestDeleteOfTheDayJson:
|
||||
URL = "/api/v3/plugins/of-the-day/json/delete"
|
||||
|
||||
def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
|
||||
assert api_v3_client.post(self.URL).status_code != 500
|
||||
|
||||
def test_json_body_still_works(self, api_v3_client, api_v3_module):
|
||||
assert api_v3_client.post(self.URL, json={}).status_code != 500
|
||||
|
||||
|
||||
class TestPluginLimits:
|
||||
URL = "/api/v3/plugins/clock/limits"
|
||||
|
||||
|
||||
@@ -182,57 +182,6 @@ class TestServePluginStatic:
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
class TestDeleteOfTheDayJson:
|
||||
"""POST /api/v3/plugins/of-the-day/json/delete
|
||||
|
||||
file_id came from the request body and was interpolated into
|
||||
``f"{file_id}.json"`` and then unlinked, with no validation at all. A
|
||||
file_id of "../../../../etc/something" deleted that file. This is the one
|
||||
finding in the batch that destroyed data rather than exposing it.
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def plugin_tree(self, tmp_path, api_v3_module):
|
||||
plugin_dir = tmp_path / "plugin-repos" / "ledmatrix-of-the-day"
|
||||
(plugin_dir / "of_the_day").mkdir(parents=True)
|
||||
(plugin_dir / "of_the_day" / "quotes.json").write_text("{}", encoding="utf-8")
|
||||
outside = tmp_path / "victim.json"
|
||||
outside.write_text("important", encoding="utf-8")
|
||||
|
||||
api_v3_module.api_v3.plugin_manager = MagicMock()
|
||||
api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(plugin_dir)
|
||||
return plugin_dir, outside
|
||||
|
||||
URL = "/api/v3/plugins/of-the-day/json/delete"
|
||||
|
||||
def test_a_real_file_in_the_plugin_is_still_deleted(
|
||||
self, api_v3_client, plugin_tree
|
||||
):
|
||||
plugin_dir, _ = plugin_tree
|
||||
target = plugin_dir / "of_the_day" / "quotes.json"
|
||||
response = api_v3_client.post(self.URL, json={"file_id": "quotes"})
|
||||
assert response.status_code == 200
|
||||
assert not target.exists()
|
||||
|
||||
def test_a_traversing_file_id_deletes_nothing(self, api_v3_client, plugin_tree):
|
||||
_, outside = plugin_tree
|
||||
response = api_v3_client.post(
|
||||
self.URL, json={"file_id": "../../../victim"}
|
||||
)
|
||||
assert response.status_code == 400
|
||||
assert outside.exists(), "file outside the plugin directory was deleted"
|
||||
assert outside.read_text(encoding="utf-8") == "important"
|
||||
|
||||
@pytest.mark.parametrize("file_id", ["..", "a/b", "/etc/x", "x" + BACKSLASH + "y"])
|
||||
def test_other_shapes_of_traversal_are_refused(
|
||||
self, api_v3_client, plugin_tree, file_id
|
||||
):
|
||||
_, outside = plugin_tree
|
||||
response = api_v3_client.post(self.URL, json={"file_id": file_id})
|
||||
assert response.status_code == 400
|
||||
assert outside.exists()
|
||||
|
||||
|
||||
class TestDiskCacheKeys:
|
||||
"""The cache key becomes a filename, and POST /api/v3/cache/delete passes
|
||||
the request body's key straight through CacheManager.clear_cache to
|
||||
|
||||
@@ -527,13 +527,11 @@ class TestDisplayAPI:
|
||||
if response.status_code in [200, 201]:
|
||||
assert api_v3.cache_manager.set.called
|
||||
|
||||
@patch('web_interface.blueprints.api_v3.display._ensure_cache_manager')
|
||||
def test_stop_on_demand_display(self, mock_ensure_cache, client):
|
||||
def test_stop_on_demand_display(self, client):
|
||||
"""Test stopping on-demand display."""
|
||||
|
||||
# Mock the cache manager returned by _ensure_cache_manager
|
||||
mock_cache_manager = MagicMock()
|
||||
mock_ensure_cache.return_value = mock_cache_manager
|
||||
from web_interface.blueprints.api_v3 import api_v3
|
||||
|
||||
mock_cache_manager = api_v3.cache_manager = MagicMock()
|
||||
|
||||
response = client.post('/api/v3/display/on-demand/stop')
|
||||
|
||||
|
||||
@@ -23,7 +23,7 @@ def two_processes(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(CacheManager, '_get_writable_cache_dir', lambda self: str(tmp_path))
|
||||
monkeypatch.setattr(CacheManager, 'start_cleanup_thread', lambda self: None)
|
||||
display, web = CacheManager(), CacheManager()
|
||||
monkeypatch.setattr(api_pkg, 'cache_manager', web)
|
||||
monkeypatch.setattr(api_pkg.api_v3, 'cache_manager', web, raising=False)
|
||||
monkeypatch.setattr(api_pkg, '_get_display_service_status', lambda: {'active': True})
|
||||
return display
|
||||
|
||||
|
||||
@@ -282,8 +282,7 @@ class TestPluginReinstall:
|
||||
def test_missing_store_manager_is_reported_per_plugin(self, client, restore):
|
||||
restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
|
||||
api_v3.plugin_store_manager = None
|
||||
with patch("web_interface.blueprints.api_v3.backup.plugin_store_manager", None):
|
||||
body = post(client).get_json()
|
||||
body = post(client).get_json()
|
||||
assert body["data"]["plugins_failed"][0]["error"] == "Store manager unavailable"
|
||||
|
||||
|
||||
|
||||
@@ -1,135 +1,16 @@
|
||||
"""
|
||||
Tests for src/web_interface/validators.py.
|
||||
Tests for validate_file_upload in src/web_interface/validators.py.
|
||||
|
||||
dedup_unique_arrays is already covered by test_dedup_unique_arrays.py and
|
||||
is not repeated here; this file covers the other eight functions, none of
|
||||
which had any tests.
|
||||
dedup_unique_arrays is covered by test_dedup_unique_arrays.py.
|
||||
|
||||
Regression coverage for three fixed bugs:
|
||||
- validate_numeric_range accepted True/False, since bool subclasses int.
|
||||
- validate_file_upload lowercased the filename's extension but not the
|
||||
caller's allowed_extensions list, so ['.TTF'] rejected 'font.ttf'.
|
||||
- validate_image_url only checked for '..' inside the relative-path
|
||||
branch, so http://host/../secret passed validation untouched.
|
||||
Regression coverage: validate_file_upload lowercased the filename's
|
||||
extension but not the caller's allowed_extensions list, so ['.TTF']
|
||||
rejected 'font.ttf'.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.web_interface.validators import (
|
||||
escape_html,
|
||||
sanitize_plugin_config,
|
||||
validate_file_upload,
|
||||
validate_font_awesome_class,
|
||||
validate_image_url,
|
||||
validate_mime_type,
|
||||
validate_numeric_range,
|
||||
validate_string_length,
|
||||
)
|
||||
|
||||
|
||||
class TestEscapeHtml:
|
||||
def test_escapes_all_five_entities(self):
|
||||
assert escape_html("""<a href="x">O'Neill & co</a>""") == (
|
||||
"<a href="x">O'Neill & co</a>")
|
||||
|
||||
def test_ampersand_is_escaped_first_so_nothing_double_escapes(self):
|
||||
# If '<' were replaced before '&', the '&' of '<' would be
|
||||
# escaped again into '&lt;'.
|
||||
assert escape_html("<") == "<"
|
||||
assert escape_html("&") == "&"
|
||||
assert escape_html("&<") == "&<"
|
||||
|
||||
def test_plain_text_unchanged(self):
|
||||
assert escape_html("hello world") == "hello world"
|
||||
|
||||
def test_non_string_is_coerced(self):
|
||||
assert escape_html(42) == "42"
|
||||
assert escape_html(None) == "None"
|
||||
|
||||
def test_script_tag_neutralized(self):
|
||||
assert "<script>" not in escape_html("<script>alert(1)</script>")
|
||||
|
||||
|
||||
class TestValidateImageUrl:
|
||||
@pytest.mark.parametrize("url", [
|
||||
"javascript:alert(1)",
|
||||
"JavaScript:alert(1)",
|
||||
"JAVASCRIPT:alert(1)",
|
||||
"data:text/html;base64,PHNjcmlwdD4=",
|
||||
"vbscript:msgbox(1)",
|
||||
"file:///etc/passwd",
|
||||
])
|
||||
def test_dangerous_protocols_rejected(self, url):
|
||||
valid, error = validate_image_url(url)
|
||||
assert valid is False and "protocol" in error.lower()
|
||||
|
||||
@pytest.mark.parametrize("url", [
|
||||
"http://x/a.png?onerror=alert(1)",
|
||||
"http://x/a.png#onload=alert(1)",
|
||||
"http://x/onclick=alert(1).png",
|
||||
])
|
||||
def test_event_handlers_rejected(self, url):
|
||||
valid, error = validate_image_url(url)
|
||||
assert valid is False and "Event handlers" in error
|
||||
|
||||
@pytest.mark.parametrize("url", ["", None, 123, []])
|
||||
def test_empty_or_non_string_rejected(self, url):
|
||||
assert validate_image_url(url)[0] is False
|
||||
|
||||
def test_http_and_https_allowed(self):
|
||||
assert validate_image_url("http://example.com/logo.png") == (True, None)
|
||||
assert validate_image_url("https://example.com/logo.png") == (True, None)
|
||||
|
||||
def test_other_schemes_rejected(self):
|
||||
valid, error = validate_image_url("ftp://example.com/logo.png")
|
||||
assert valid is False and "http://" in error
|
||||
|
||||
def test_relative_path_allowed(self):
|
||||
assert validate_image_url("/static/logo.png") == (True, None)
|
||||
|
||||
def test_protocol_relative_url_rejected(self):
|
||||
assert validate_image_url("//evil.com/logo.png")[0] is False
|
||||
|
||||
def test_relative_traversal_rejected(self):
|
||||
assert validate_image_url("/static/../../etc/passwd")[0] is False
|
||||
|
||||
def test_absolute_url_traversal_rejected(self):
|
||||
# Regression: the '..' check used to sit inside the leading-slash
|
||||
# branch, so an absolute URL skipped it entirely.
|
||||
valid, error = validate_image_url("http://example.com/../secret")
|
||||
assert valid is False and "traversal" in error.lower()
|
||||
|
||||
def test_bare_traversal_rejected(self):
|
||||
assert validate_image_url("../../etc/passwd")[0] is False
|
||||
|
||||
|
||||
class TestValidateFontAwesomeClass:
|
||||
@pytest.mark.parametrize("cls", ["fa-star", "fas fa-star", "fa-solid fa-house"])
|
||||
def test_valid_classes_accepted(self, cls):
|
||||
assert validate_font_awesome_class(cls) == (True, None)
|
||||
|
||||
@pytest.mark.parametrize("cls", ["star", "glyphicon-star", ""])
|
||||
def test_classes_without_fa_prefix_rejected(self, cls):
|
||||
assert validate_font_awesome_class(cls)[0] is False
|
||||
|
||||
def test_injection_attempt_rejected(self):
|
||||
assert validate_font_awesome_class('fa-star" onload="alert(1)')[0] is False
|
||||
|
||||
def test_angle_brackets_rejected(self):
|
||||
assert validate_font_awesome_class("<script>fa-star</script>")[0] is False
|
||||
|
||||
def test_non_string_rejected(self):
|
||||
valid, error = validate_font_awesome_class(None)
|
||||
assert valid is False and "string" in error
|
||||
|
||||
def test_explicit_fa_check_is_unreachable_but_harmless(self):
|
||||
# Characterized, not fixed: the regex already requires 'fa-', so the
|
||||
# follow-up `if 'fa-' not in class_name` can never fire. Anything
|
||||
# lacking 'fa-' is rejected by the pattern first, with the pattern's
|
||||
# own message.
|
||||
valid, error = validate_font_awesome_class("star")
|
||||
assert valid is False
|
||||
assert error == "Invalid Font Awesome class name format"
|
||||
from src.web_interface.validators import validate_file_upload
|
||||
|
||||
|
||||
class TestValidateFileUpload:
|
||||
@@ -164,121 +45,3 @@ class TestValidateFileUpload:
|
||||
|
||||
def test_no_extension_list_skips_the_check(self):
|
||||
assert validate_file_upload("anything.xyz") == (True, None)
|
||||
|
||||
|
||||
class TestValidateMimeType:
|
||||
def test_known_type_accepted(self):
|
||||
assert validate_mime_type("logo.png", ["image/png"]) == (True, None)
|
||||
|
||||
def test_mismatched_type_rejected(self):
|
||||
valid, error = validate_mime_type("logo.png", ["image/jpeg"])
|
||||
assert valid is False and "not allowed" in error
|
||||
|
||||
def test_undeterminable_type_rejected(self):
|
||||
valid, error = validate_mime_type("mystery.zzz", ["image/png"])
|
||||
assert valid is False and "Could not determine" in error
|
||||
|
||||
def test_guess_type_failure_is_caught(self, monkeypatch):
|
||||
import mimetypes
|
||||
monkeypatch.setattr(mimetypes, "guess_type",
|
||||
lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("boom")))
|
||||
valid, error = validate_mime_type("logo.png", ["image/png"])
|
||||
assert valid is False and "Error validating MIME type" in error
|
||||
|
||||
|
||||
class TestValidateNumericRange:
|
||||
def test_value_in_range(self):
|
||||
assert validate_numeric_range(5, min_val=0, max_val=10) == (True, None)
|
||||
|
||||
def test_boundaries_are_inclusive(self):
|
||||
assert validate_numeric_range(0, min_val=0, max_val=10) == (True, None)
|
||||
assert validate_numeric_range(10, min_val=0, max_val=10) == (True, None)
|
||||
|
||||
def test_below_minimum_rejected(self):
|
||||
valid, error = validate_numeric_range(-1, min_val=0)
|
||||
assert valid is False and "at least" in error
|
||||
|
||||
def test_above_maximum_rejected(self):
|
||||
valid, error = validate_numeric_range(11, max_val=10)
|
||||
assert valid is False and "at most" in error
|
||||
|
||||
def test_floats_accepted(self):
|
||||
assert validate_numeric_range(2.5, min_val=0, max_val=10) == (True, None)
|
||||
|
||||
def test_no_bounds_accepts_any_number(self):
|
||||
assert validate_numeric_range(-9999) == (True, None)
|
||||
|
||||
@pytest.mark.parametrize("value", ["5", None, [], {}])
|
||||
def test_non_numeric_rejected(self, value):
|
||||
valid, error = validate_numeric_range(value, min_val=0, max_val=10)
|
||||
assert valid is False and error == "Value must be a number"
|
||||
|
||||
@pytest.mark.parametrize("value", [True, False])
|
||||
def test_booleans_rejected(self, value):
|
||||
# Regression: bool subclasses int, so True passed the isinstance
|
||||
# check and then compared as 1 against the range.
|
||||
valid, error = validate_numeric_range(value, min_val=0, max_val=10)
|
||||
assert valid is False and error == "Value must be a number"
|
||||
|
||||
|
||||
class TestValidateStringLength:
|
||||
def test_within_range(self):
|
||||
assert validate_string_length("hello", min_length=1, max_length=10) == (True, None)
|
||||
|
||||
def test_boundaries_are_inclusive(self):
|
||||
assert validate_string_length("abc", min_length=3, max_length=3) == (True, None)
|
||||
|
||||
def test_too_short_rejected(self):
|
||||
valid, error = validate_string_length("", min_length=1)
|
||||
assert valid is False and "at least" in error
|
||||
|
||||
def test_too_long_rejected(self):
|
||||
valid, error = validate_string_length("abcdef", max_length=3)
|
||||
assert valid is False and "at most" in error
|
||||
|
||||
def test_non_string_rejected(self):
|
||||
valid, error = validate_string_length(123, max_length=10)
|
||||
assert valid is False and "must be a string" in error
|
||||
|
||||
def test_no_bounds_accepts_anything(self):
|
||||
assert validate_string_length("") == (True, None)
|
||||
|
||||
|
||||
class TestSanitizePluginConfig:
|
||||
def test_valid_keys_and_scalars_kept(self):
|
||||
config = {"enabled": True, "count": 3, "ratio": 1.5, "name": "clock"}
|
||||
assert sanitize_plugin_config(config) == config
|
||||
|
||||
@pytest.mark.parametrize("key", ["has space", "has-dash", "has.dot", "has/slash", ""])
|
||||
def test_invalid_key_names_dropped(self, key):
|
||||
assert sanitize_plugin_config({key: "value", "good": 1}) == {"good": 1}
|
||||
|
||||
def test_non_string_keys_dropped(self):
|
||||
assert sanitize_plugin_config({1: "a", "good": 2}) == {"good": 2}
|
||||
|
||||
def test_nested_dicts_recursed(self):
|
||||
result = sanitize_plugin_config({"outer": {"inner": 1, "bad key": 2}})
|
||||
assert result == {"outer": {"inner": 1}}
|
||||
|
||||
def test_list_of_scalars_preserved(self):
|
||||
assert sanitize_plugin_config({"teams": ["PHI", "NYG"]})["teams"] == ["PHI", "NYG"]
|
||||
|
||||
def test_list_of_dicts_recursed(self):
|
||||
result = sanitize_plugin_config({"items": [{"ok": 1, "bad key": 2}]})
|
||||
assert result["items"] == [{"ok": 1}]
|
||||
|
||||
def test_unknown_value_types_dropped(self):
|
||||
assert sanitize_plugin_config({"weird": {1, 2, 3}, "good": 1}) == {"good": 1}
|
||||
|
||||
def test_none_values_dropped(self):
|
||||
assert sanitize_plugin_config({"nothing": None, "good": 1}) == {"good": 1}
|
||||
|
||||
def test_strings_are_not_html_escaped(self):
|
||||
# Pinned, not a bug: escaping here would persist the escaped form in
|
||||
# config.json. Output escaping belongs to the template layer, which
|
||||
# the function's docstring now says explicitly.
|
||||
payload = "<script>alert(1)</script>"
|
||||
assert sanitize_plugin_config({"title": payload})["title"] == payload
|
||||
|
||||
def test_empty_config(self):
|
||||
assert sanitize_plugin_config({}) == {}
|
||||
|
||||
Reference in New Issue
Block a user