feat(extensions): solid update detection (builtin sources, trust hints, refresh, count)

This commit is contained in:
Juan Denis committed 2026-11-03 20:20:35 -05:00
1 parent e401cbaf98
commit c5a90e2c82
4 files changed
+225 -32

No files matched your search

+20 -7
View File
@@ -643,10 +643,21 @@ def list_updates():
"""List installed plugins that have a newer version in the registry.
Each entry: {slug, plugin_id, installed_version, available_version,
update_available, compatible, source}.
update_available, compatible, source, display_name, update_source,
incompatible_reason, requires_ack, risk}. ``?refresh=1`` refetches the
registry index first.
"""
from datetime import datetime, timezone
from app.services import registry_service
from app.services.plugin_service import check_for_updates
return jsonify({'updates': check_for_updates()})
refresh = request.args.get('refresh', '').lower() in ('1', 'true', 'yes')
updates = check_for_updates(refresh=refresh)
return jsonify({
'updates': updates,
'count': sum(1 for u in updates if u['update_available'] and u['compatible']),
'checked_at': datetime.now(timezone.utc).isoformat(),
'registry_source': registry_service.registry_source_label(),
})
@plugins_bp.route('/<int:plugin_id>/update', methods=['POST'])
@@ -670,21 +681,23 @@ def update_plugin_route(plugin_id):
if not plugin:
return jsonify({'error': 'Plugin not found'}), 404
entry = registry_service.get_entry(plugin.slug)
if not entry:
builtin = plugin.source_type == 'builtin' # local bundled code: no consent
entry = None if builtin else registry_service.get_entry(plugin.slug)
if not builtin and not entry:
return jsonify({
'error': f"No registry entry for '{plugin.slug}' — cannot update "
f"from the registry."
}), 400
# Hidden means not updatable either: unreviewed entries only exist in
# development contexts (see registry_service._show_unreviewed).
if entry.get('trust') == 'unreviewed' \
if entry and entry.get('trust') == 'unreviewed' \
and not registry_service._show_unreviewed():
return jsonify({'error': 'Extension not found in the registry'}), 404
body = json_object(optional=True)
acknowledged = body.get('acknowledge_risk') is True
reason, message = registry_service.consent_gate(entry, acknowledged)
reason, message = (None, None) if builtin else \
registry_service.consent_gate(entry, acknowledged)
if reason:
return jsonify({
'error': f'{message} Resend with acknowledge_risk: true to proceed.',
@@ -703,7 +716,7 @@ def update_plugin_route(plugin_id):
target_id=plugin.id,
details={
'name': plugin.name, 'version': plugin.version,
'source': 'registry-update',
'source': 'builtin-update' if builtin else 'registry-update',
# Forensics (audit I7): the signature verdict of the bytes
# that just landed.
'signature': sig.get('status') or 'unsigned',
+67 -24
View File
@@ -2133,37 +2133,71 @@ def install_registry_extension(slug, user_id=None):
)
def check_for_updates():
"""Compare installed plugins to the registry; return update candidates.
def check_for_updates(refresh=False):
"""Compare installed plugins to the registry / bundled manifests.
Update v1 is registry-driven: a plugin whose slug is in the registry with a
higher version than installed is flagged. Panel-version compatibility is
surfaced so the UI can gray out an update the panel can't run yet. The
installed plugin's ``source_type`` is included so callers can see the
provenance the update would cross (audit M2 — a registry entry only ever
matches on slug; the update route's consent gate is the enforcement).
Registry-driven for url/registry rows (matched on slug); ``builtin`` rows
compare against the bundled manifest in builtin-extensions/. Rows with no
known source of truth (url/local/upload/git without a registry entry) are
skipped. A malformed version means "no update"; one bad row never breaks
the list. ``requires_ack``/``risk`` mirror the update route's consent gate.
"""
from app.services import registry_service
from app.utils.version import compare_versions, version_satisfies, get_panel_version
if refresh:
registry_service.refresh(force=True)
panel = get_panel_version()
bundled = {b['slug']: b['manifest'] for b in list_builtin_extensions()}
out = []
for p in InstalledPlugin.query.all():
entry = registry_service.get_entry(p.slug)
if not entry:
continue
available = entry.get('version')
out.append({
'slug': p.slug,
'plugin_id': p.id,
'installed_version': p.version,
'available_version': available,
'update_available': compare_versions(available, p.version) > 0,
'compatible': version_satisfies(
panel, entry.get('min_panel_version'), entry.get('max_panel_version')
),
'source': entry.get('source'),
'installed_source_type': p.source_type,
})
try:
if p.source_type == 'builtin':
m = bundled.get(p.slug)
if not m:
continue
available = m.get('version')
row = {
'display_name': m.get('display_name') or p.name,
'update_source': 'builtin', 'source': None,
'compatible': True, 'requires_ack': False, 'risk': None,
}
minv = m.get('min_panel_version')
maxv = m.get('max_panel_version')
else:
entry = registry_service.get_entry(p.slug)
if not entry:
continue
available = entry.get('version')
minv = entry.get('min_panel_version')
maxv = entry.get('max_panel_version')
risk, _ = registry_service.consent_gate(entry, False)
row = {
'display_name': entry.get('display_name') or p.name,
'update_source': 'registry', 'source': entry.get('source'),
'requires_ack': bool(risk), 'risk': risk,
}
try:
newer = compare_versions(available, p.version) > 0
except Exception: # noqa: BLE001 - malformed version = no update
newer = False
compatible = version_satisfies(panel, minv, maxv)
row['compatible'] = bool(compatible)
row['incompatible_reason'] = None if compatible else (
f"Needs panel {minv or '*'}–{maxv or '*'} (this panel is {panel})."
)
if minv:
row['min_panel_version'] = minv
out.append({
'slug': p.slug,
'plugin_id': p.id,
'installed_version': p.version,
'available_version': available,
'update_available': newer,
'installed_source_type': p.source_type,
**row,
})
except Exception as exc: # noqa: BLE001 - isolate bad rows
logger.warning(f'Update check skipped {p.slug}: {exc}')
return out
@@ -2173,6 +2207,15 @@ def update_plugin(plugin_id, user_id=None):
plugin = InstalledPlugin.query.get(plugin_id)
if not plugin:
raise ValueError('Plugin not found')
folder = next((b['path'] for b in list_builtin_extensions()
if b['slug'] == plugin.slug), None) \
if plugin.source_type == 'builtin' else None
# A builtin row with no bundled folder (e.g. stamped by hand) falls
# through to the registry path below.
if folder:
return install_from_path(
folder, user_id=user_id, force=True, source_type='builtin',
)
entry = registry_service.get_entry(plugin.slug)
if not entry:
raise ValueError(
+137
View File
@@ -0,0 +1,137 @@
"""Update detection: registry + builtin rows, compat, consent hints, API shape."""
import io
import json
import zipfile
import pytest
from app import db
from app.models.plugin import InstalledPlugin
from app.services import plugin_service, registry_service
pytestmark = pytest.mark.fresh_app
@pytest.fixture(autouse=True)
def _reset_registry_cache():
registry_service._cache.update({'ts': 0.0, 'entries': None, 'source': None})
yield
registry_service._cache.update({'ts': 0.0, 'entries': None, 'source': None})
@pytest.fixture
def plugin_dirs(tmp_path, monkeypatch):
d = {n: tmp_path / n for n in ('b', 'f', 'builtin')}
for p in d.values():
p.mkdir()
monkeypatch.setattr(plugin_service, 'BACKEND_PLUGINS_DIR', str(d['b']))
monkeypatch.setattr(plugin_service, 'FRONTEND_PLUGINS_DIR', str(d['f']))
monkeypatch.setattr(plugin_service, 'BUILTIN_EXTENSIONS_DIR', str(d['builtin']))
return d
def _manifest(slug, version):
return {'name': slug, 'display_name': 'Ext ' + slug, 'version': version,
'category': 'utility',
'contributions': {'nav': [{'id': slug, 'label': 'X', 'route': '/x'}]}}
def _zip(slug, version):
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w') as zf:
zf.writestr('plugin.json', json.dumps(_manifest(slug, version)))
zf.writestr('frontend/index.jsx', 'export function P(){return null;}\n')
return buf.getvalue()
def _install_registry_row(monkeypatch, version='1.0.0'):
data = _zip('regext', version)
monkeypatch.setattr(plugin_service, '_download_zip', lambda url: io.BytesIO(data))
return plugin_service.install_from_url('https://x/regext.zip')
def _advertise(version, **extra):
registry_service._cache.update({
'ts': 9e18, 'source': 'test',
'entries': [registry_service._normalize({
'slug': 'regext', 'display_name': 'Registry Ext', 'version': version,
'source': 'https://x/regext.zip', 'min_panel_version': '0.0.1',
**extra})],
})
def _by_slug():
return {u['slug']: u for u in plugin_service.check_for_updates()}
def test_registry_update_fields(app, plugin_dirs, monkeypatch):
_install_registry_row(monkeypatch)
_advertise('2.0.0')
u = _by_slug()['regext']
assert u['update_available'] is True
assert u['display_name'] == 'Registry Ext'
assert u['update_source'] == 'registry'
assert isinstance(u['requires_ack'], bool)
assert u['requires_ack'] == (u['risk'] is not None)
assert u['incompatible_reason'] is None
def test_up_to_date(app, plugin_dirs, monkeypatch):
_install_registry_row(monkeypatch)
_advertise('1.0.0')
assert _by_slug()['regext']['update_available'] is False
def test_incompatible_panel(app, plugin_dirs, monkeypatch):
_install_registry_row(monkeypatch)
_advertise('2.0.0', min_panel_version='999.0.0')
u = _by_slug()['regext']
assert u['compatible'] is False
assert '999.0.0' in u['incompatible_reason']
def test_malformed_version_no_raise(app, plugin_dirs, monkeypatch):
p = _install_registry_row(monkeypatch)
_advertise('banana')
assert _by_slug()['regext']['update_available'] is False
p.version = 'also-bad'
db.session.commit()
assert _by_slug()['regext']['update_available'] is False
def test_builtin_update(app, plugin_dirs, monkeypatch):
folder = plugin_dirs['builtin'] / 'bi-ext'
(folder / 'frontend').mkdir(parents=True)
(folder / 'frontend' / 'index.jsx').write_text('export function P(){return null;}\n')
(folder / 'plugin.json').write_text(json.dumps(_manifest('bi-ext', '1.0.0')))
p = plugin_service.install_from_path(str(folder), source_type='builtin')
assert p.source_type == 'builtin'
# A registry entry with the same slug must be ignored for builtin rows.
assert plugin_service.check_for_updates()[0]['update_available'] is False
(folder / 'plugin.json').write_text(json.dumps(_manifest('bi-ext', '1.1.0')))
u = _by_slug()['bi-ext']
assert u['update_available'] is True
assert u['update_source'] == 'builtin'
assert u['requires_ack'] is False and u['risk'] is None
updated = plugin_service.update_plugin(p.id)
assert updated.version == '1.1.0'
assert updated.source_type == 'builtin'
assert InstalledPlugin.query.filter_by(slug='bi-ext').count() == 1
def test_api_shape_and_refresh(app, client, auth_headers, plugin_dirs, monkeypatch):
calls = []
_install_registry_row(monkeypatch)
_advertise('2.0.0')
monkeypatch.setattr(registry_service, 'refresh',
lambda force=False: calls.append(force) or registry_service._cache['entries'])
r = client.get('/api/v1/plugins/updates', headers=auth_headers)
assert r.status_code == 200
body = r.get_json()
assert body['count'] == 1 and body['checked_at']
assert 'registry_source' in body
assert True not in calls
client.get('/api/v1/plugins/updates?refresh=1', headers=auth_headers)
assert True in calls
+1 -1
View File
@@ -40,7 +40,7 @@ Relevant endpoints:
|---|---|
| `GET /api/v1/marketplace/registry` | list registry entries + live install state |
| `POST /api/v1/marketplace/registry/<slug>/install` | checksum-verified install (admin) |
| `GET /api/v1/plugins/updates` | installed plugins with a newer registry version |
| `GET /api/v1/plugins/updates` | installed plugins with a newer registry/bundled version; returns `{updates, count, checked_at, registry_source}`, `?refresh=1` refetches |
| `POST /api/v1/plugins/<id>/update` | update a plugin to the registry version (admin) |
---