mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-02 17:28:05 +00:00
fix(tools-tab): address code review findings
- Add _GIT = shutil.which('git') alongside _SUDO/_JOURNALCTL; return
503 in force_git_reset and get_git_info if git is unavailable
- Check git branch/status returncodes in get_git_info(); return a clear
500 error instead of silently treating a failed run as a clean repo
- Cap pip stdout+stderr at 50 KB via _truncate_output() helper to
avoid OOM on verbose dependency resolution or build failures
- Scrub embedded HTTPS credentials from remote_url via
_scrub_git_remote_url() using urllib.parse before returning to UI
- Fix clear_pycache to track and report failed deletions separately
instead of counting them as successes (removed ignore_errors=True,
wrapped in try/except OSError)
Skipped: plugin_manager-vs-api_v3.plugin_manager (api_v3 is the
Blueprint object; accessing .plugin_manager on it would fail — module-
level variable is the correct pattern used throughout this blueprint);
pages_v3 broad-except (identical to every other _load_*_partial in the
file); base.html HTMX fallback (loadTabContent handles all tabs
generically; named fallbacks only exist for tabs needing JS re-init);
tools.html auth (pre-existing architectural decision — reboot/shutdown
on the same endpoint are also unauthenticated).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
bafa2048a0
commit
344f64ebba
@@ -14,6 +14,7 @@ import logging
|
|||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, Any
|
from typing import Dict, Any
|
||||||
|
from urllib.parse import urlparse, urlunparse
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -28,6 +29,32 @@ from src.error_aggregator import get_error_aggregator
|
|||||||
|
|
||||||
_SUDO = shutil.which('sudo')
|
_SUDO = shutil.which('sudo')
|
||||||
_JOURNALCTL = shutil.which('journalctl')
|
_JOURNALCTL = shutil.which('journalctl')
|
||||||
|
_GIT = shutil.which('git')
|
||||||
|
|
||||||
|
# Cap subprocess output returned to the browser — pip can produce MBs on build failures.
|
||||||
|
_MAX_OUTPUT_BYTES = 51_200 # 50 KB
|
||||||
|
|
||||||
|
|
||||||
|
def _truncate_output(stdout: str, stderr: str) -> str:
|
||||||
|
"""Combine stdout+stderr and truncate to _MAX_OUTPUT_BYTES (keeping the tail)."""
|
||||||
|
combined = (stdout + stderr).strip()
|
||||||
|
if len(combined) > _MAX_OUTPUT_BYTES:
|
||||||
|
combined = '[...output truncated...]\n' + combined[-_MAX_OUTPUT_BYTES:]
|
||||||
|
return combined
|
||||||
|
|
||||||
|
|
||||||
|
def _scrub_git_remote_url(url: str) -> str:
|
||||||
|
"""Strip embedded username/password from an HTTPS remote URL before returning it to the UI."""
|
||||||
|
try:
|
||||||
|
p = urlparse(url)
|
||||||
|
if p.scheme in ('http', 'https') and (p.username or p.password):
|
||||||
|
netloc = p.hostname or ''
|
||||||
|
if p.port:
|
||||||
|
netloc += f':{p.port}'
|
||||||
|
return urlunparse(p._replace(netloc=netloc))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return url
|
||||||
|
|
||||||
# Will be initialized when blueprint is registered
|
# Will be initialized when blueprint is registered
|
||||||
config_manager = None
|
config_manager = None
|
||||||
@@ -1636,7 +1663,7 @@ def execute_system_action():
|
|||||||
return jsonify({
|
return jsonify({
|
||||||
'status': 'success' if result.returncode == 0 else 'error',
|
'status': 'success' if result.returncode == 0 else 'error',
|
||||||
'message': 'Base requirements installed successfully' if result.returncode == 0 else 'pip install failed',
|
'message': 'Base requirements installed successfully' if result.returncode == 0 else 'pip install failed',
|
||||||
'output': (result.stdout + result.stderr).strip()
|
'output': _truncate_output(result.stdout, result.stderr)
|
||||||
})
|
})
|
||||||
elif action == 'install_plugin_requirements':
|
elif action == 'install_plugin_requirements':
|
||||||
plugins_dir = Path(plugin_manager.plugins_dir) if plugin_manager else PROJECT_ROOT / 'plugin-repos'
|
plugins_dir = Path(plugin_manager.plugins_dir) if plugin_manager else PROJECT_ROOT / 'plugin-repos'
|
||||||
@@ -1652,7 +1679,7 @@ def execute_system_action():
|
|||||||
results.append({
|
results.append({
|
||||||
'plugin': p.name,
|
'plugin': p.name,
|
||||||
'ok': r.returncode == 0,
|
'ok': r.returncode == 0,
|
||||||
'output': (r.stdout + r.stderr).strip()
|
'output': _truncate_output(r.stdout, r.stderr)
|
||||||
})
|
})
|
||||||
ok_count = sum(1 for r in results if r['ok'])
|
ok_count = sum(1 for r in results if r['ok'])
|
||||||
all_ok = all(r['ok'] for r in results) if results else True
|
all_ok = all(r['ok'] for r in results) if results else True
|
||||||
@@ -1662,15 +1689,17 @@ def execute_system_action():
|
|||||||
'details': results
|
'details': results
|
||||||
})
|
})
|
||||||
elif action == 'force_git_reset':
|
elif action == 'force_git_reset':
|
||||||
|
if not _GIT:
|
||||||
|
return jsonify({'status': 'error', 'message': 'git not found on this system'}), 503
|
||||||
project_dir = str(PROJECT_ROOT)
|
project_dir = str(PROJECT_ROOT)
|
||||||
fetch = subprocess.run(
|
fetch = subprocess.run(
|
||||||
['git', 'fetch', 'origin'],
|
[_GIT, 'fetch', 'origin'],
|
||||||
capture_output=True, text=True, timeout=30, cwd=project_dir
|
capture_output=True, text=True, timeout=30, cwd=project_dir
|
||||||
)
|
)
|
||||||
if fetch.returncode != 0:
|
if fetch.returncode != 0:
|
||||||
return jsonify({'status': 'error', 'message': 'git fetch failed', 'output': fetch.stderr.strip()})
|
return jsonify({'status': 'error', 'message': 'git fetch failed', 'output': fetch.stderr.strip()})
|
||||||
reset = subprocess.run(
|
reset = subprocess.run(
|
||||||
['git', 'reset', '--hard', 'origin/main'],
|
[_GIT, 'reset', '--hard', 'origin/main'],
|
||||||
capture_output=True, text=True, timeout=30, cwd=project_dir
|
capture_output=True, text=True, timeout=30, cwd=project_dir
|
||||||
)
|
)
|
||||||
return jsonify({
|
return jsonify({
|
||||||
@@ -1680,11 +1709,18 @@ def execute_system_action():
|
|||||||
})
|
})
|
||||||
elif action == 'clear_pycache':
|
elif action == 'clear_pycache':
|
||||||
cleared = 0
|
cleared = 0
|
||||||
|
failed = 0
|
||||||
for d in PROJECT_ROOT.rglob('__pycache__'):
|
for d in PROJECT_ROOT.rglob('__pycache__'):
|
||||||
if d.is_dir():
|
if d.is_dir():
|
||||||
shutil.rmtree(d, ignore_errors=True)
|
try:
|
||||||
|
shutil.rmtree(d)
|
||||||
cleared += 1
|
cleared += 1
|
||||||
return jsonify({'status': 'success', 'message': f'Cleared {cleared} __pycache__ directories'})
|
except OSError:
|
||||||
|
failed += 1
|
||||||
|
msg = f'Cleared {cleared} __pycache__ directories'
|
||||||
|
if failed:
|
||||||
|
msg += f' ({failed} could not be removed)'
|
||||||
|
return jsonify({'status': 'success', 'message': msg})
|
||||||
else:
|
else:
|
||||||
return jsonify({'status': 'error', 'message': 'Unknown action'}), 400
|
return jsonify({'status': 'error', 'message': 'Unknown action'}), 400
|
||||||
|
|
||||||
@@ -1710,18 +1746,26 @@ def execute_system_action():
|
|||||||
@api_v3.route('/system/git-info', methods=['GET'])
|
@api_v3.route('/system/git-info', methods=['GET'])
|
||||||
def get_git_info():
|
def get_git_info():
|
||||||
"""Return branch, dirty state, recent commits and remote URL for the Tools tab."""
|
"""Return branch, dirty state, recent commits and remote URL for the Tools tab."""
|
||||||
|
if not _GIT:
|
||||||
|
return jsonify({'status': 'error', 'message': 'git not found on this system'}), 503
|
||||||
d = str(PROJECT_ROOT)
|
d = str(PROJECT_ROOT)
|
||||||
try:
|
try:
|
||||||
branch = subprocess.run(['git', 'branch', '--show-current'], capture_output=True, text=True, timeout=10, cwd=d)
|
branch = subprocess.run([_GIT, 'branch', '--show-current'], capture_output=True, text=True, timeout=10, cwd=d)
|
||||||
status = subprocess.run(['git', 'status', '--short', '--untracked-files=no'], capture_output=True, text=True, timeout=15, cwd=d)
|
if branch.returncode != 0:
|
||||||
log = subprocess.run(['git', 'log', '--oneline', '-5'], capture_output=True, text=True, timeout=10, cwd=d)
|
return jsonify({'status': 'error', 'message': f'git branch failed: {branch.stderr.strip()}'}), 500
|
||||||
remote = subprocess.run(['git', 'remote', 'get-url', 'origin'], capture_output=True, text=True, timeout=10, cwd=d)
|
|
||||||
|
status = subprocess.run([_GIT, 'status', '--short', '--untracked-files=no'], capture_output=True, text=True, timeout=15, cwd=d)
|
||||||
|
if status.returncode != 0:
|
||||||
|
return jsonify({'status': 'error', 'message': f'git status failed: {status.stderr.strip()}'}), 500
|
||||||
|
|
||||||
|
log = subprocess.run([_GIT, 'log', '--oneline', '-5'], capture_output=True, text=True, timeout=10, cwd=d)
|
||||||
|
remote = subprocess.run([_GIT, 'remote', 'get-url', 'origin'], capture_output=True, text=True, timeout=10, cwd=d)
|
||||||
return jsonify({
|
return jsonify({
|
||||||
'branch': branch.stdout.strip(),
|
'branch': branch.stdout.strip(),
|
||||||
'dirty': bool(status.stdout.strip()),
|
'dirty': bool(status.stdout.strip()),
|
||||||
'status': status.stdout.strip(),
|
'status': status.stdout.strip(),
|
||||||
'recent_commits': log.stdout.strip(),
|
'recent_commits': log.stdout.strip() if log.returncode == 0 else '',
|
||||||
'remote_url': remote.stdout.strip(),
|
'remote_url': _scrub_git_remote_url(remote.stdout.strip()) if remote.returncode == 0 else '',
|
||||||
})
|
})
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error("get_git_info failed: %s", e, exc_info=True)
|
logger.error("get_git_info failed: %s", e, exc_info=True)
|
||||||
|
|||||||
Reference in New Issue
Block a user